From 7ae0b07f50102e97817138ce871c8e3906f1c8a5 Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Tue, 26 May 2026 22:53:02 +0800 Subject: [PATCH 1/7] feat(telemetry): trace daemon prompt lifecycle Connect qwen serve HTTP routes, ACP bridge dispatch, and ACP child prompt execution through OpenTelemetry context propagation. The daemon injects reserved qwen.telemetry metadata internally so clients do not need to pass trace context. Closes #4554 Co-authored-by: Qwen-Coder --- packages/acp-bridge/src/bridge.test.ts | 56 ++ packages/acp-bridge/src/bridge.ts | 414 ++++++++------- packages/acp-bridge/src/bridgeOptions.ts | 19 + .../src/acp-integration/session/Session.ts | 478 +++++++++--------- packages/cli/src/serve/runQwenServe.ts | 87 +++- packages/cli/src/serve/server.ts | 86 +++- .../core/src/telemetry/daemon-tracing.test.ts | 55 ++ packages/core/src/telemetry/daemon-tracing.ts | 307 +++++++++++ packages/core/src/telemetry/index.ts | 19 + packages/core/src/telemetry/metrics.ts | 9 +- packages/core/src/telemetry/runtime-config.ts | 24 + packages/core/src/telemetry/sdk.ts | 4 +- .../src/telemetry/session-tracing.test.ts | 26 + .../core/src/telemetry/session-tracing.ts | 79 +++ 14 files changed, 1238 insertions(+), 425 deletions(-) create mode 100644 packages/core/src/telemetry/daemon-tracing.test.ts create mode 100644 packages/core/src/telemetry/daemon-tracing.ts create mode 100644 packages/core/src/telemetry/runtime-config.ts diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index d09bf8a31d2..bc796f81c94 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -19,6 +19,7 @@ import type { Agent, InitializeResponse, LoadSessionResponse, + PromptRequest, PromptResponse, ResumeSessionResponse, } from '@agentclientprotocol/sdk'; @@ -41,6 +42,7 @@ import { import { MAX_WORKSPACE_PATH_LENGTH } from './workspacePaths.js'; import { createHttpAcpBridge } from './bridge.js'; import type { ChannelFactory } from './channel.js'; +import type { BridgeTelemetry } from './bridgeOptions.js'; import { createInMemoryChannel } from './inMemoryChannel.js'; import type { BridgeEvent } from './eventBus.js'; import { ApprovalMode } from '@qwen-code/qwen-code-core'; @@ -89,6 +91,60 @@ describe('createHttpAcpBridge', () => { ); }); + it('uses bridge telemetry for channel/session/prompt dispatch and prompt metadata injection', async () => { + const handle = makeChannel(); + const operations: string[] = []; + const telemetry: BridgeTelemetry = { + captureContext: () => ({ captured: true }), + async runWithContext(_captured, fn) { + return await fn(); + }, + async withSpan(operation, _attributes, fn) { + operations.push(operation); + return await fn(); + }, + event() {}, + injectPromptContext(request) { + const meta = + (request as { _meta?: Record })._meta ?? {}; + return { + ...request, + _meta: { + ...meta, + 'qwen.telemetry.traceparent': 'daemon-traceparent', + }, + }; + }, + }; + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + telemetry, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'hello' }], + _meta: { + keep: 'value', + 'qwen.telemetry.traceparent': 'client-spoof', + }, + } as PromptRequest); + + expect(operations).toEqual( + expect.arrayContaining([ + 'channel.spawn', + 'channel.initialize', + 'session.new', + 'prompt.dispatch', + ]), + ); + expect(handle.agent.promptCalls[0]!._meta).toMatchObject({ + keep: 'value', + 'qwen.telemetry.traceparent': 'daemon-traceparent', + }); + }); + it('forwards childEnvOverrides to the channelFactory at spawn time (#4247 R6 line 216)', async () => { // Round 6 (wenshao R5 line 216): pre-fix `runQwenServe` set // `process.env` globally to pass the MCP budget config to the diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 10d64ccf602..ca84a4daa46 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -76,7 +76,7 @@ import type { BridgeSessionSummary, HttpAcpBridge, } from './bridgeTypes.js'; -import type { BridgeOptions } from './bridgeOptions.js'; +import type { BridgeOptions, BridgeTelemetry } from './bridgeOptions.js'; import { defaultSpawnChannelFactory } from './spawnChannel.js'; import { writeStderrLine } from './internal/stderrLine.js'; import { BridgeClient } from './bridgeClient.js'; @@ -88,6 +88,30 @@ import { } from './permissionMediator.js'; import { PermissionForbiddenError } from './bridgeErrors.js'; +const DAEMON_TRACEPARENT_META_KEY = 'qwen.telemetry.traceparent'; +const DAEMON_TRACESTATE_META_KEY = 'qwen.telemetry.tracestate'; + +const NOOP_BRIDGE_TELEMETRY: BridgeTelemetry = { + captureContext: () => undefined, + async runWithContext(_captured, fn) { + return await fn(); + }, + async withSpan(_operation, _attributes, fn) { + return await fn(); + }, + event() {}, + injectPromptContext(request) { + const meta = (request as { _meta?: unknown })._meta; + if (!meta || typeof meta !== 'object' || Array.isArray(meta)) { + return request; + } + const nextMeta = { ...(meta as Record) }; + delete nextMeta[DAEMON_TRACEPARENT_META_KEY]; + delete nextMeta[DAEMON_TRACESTATE_META_KEY]; + return { ...request, _meta: nextMeta }; + }, +}; + /** * Stage 1 HTTP→ACP bridge factory + supporting helpers, lifted from * `cli/src/serve/httpAcpBridge.ts` to `@qwen-code/acp-bridge/bridge` @@ -712,6 +736,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { const contextFilename = opts.contextFilename ?? getCurrentGeminiMdFilename(); const persistApprovalMode = opts.persistApprovalMode; const persistDisabledTools = opts.persistDisabledTools; + const telemetry = opts.telemetry ?? NOOP_BRIDGE_TELEMETRY; // #3803 §02 single-workspace model: the bridge hosts AT MOST one // ATTACH-AVAILABLE channel and one default attach-target entry. @@ -948,7 +973,14 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { if (inFlightChannelSpawn) return await inFlightChannelSpawn; const promise = (async () => { - const channel = await channelFactory(boundWorkspace, childEnvOverrides); + const channel = await telemetry.withSpan( + 'channel.spawn', + { + 'qwen-code.daemon.bridge.operation': 'channel.spawn', + 'qwen-code.daemon.channel.reused': false, + }, + async () => await channelFactory(boundWorkspace, childEnvOverrides), + ); const client = new BridgeClient( // BfFut: ACP today carries a sessionId on every per-session // notification / request, so the no-sessionId branch is @@ -1065,6 +1097,13 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // context line in that flow, and the message confirms the // cleanup actually ran. if (!shuttingDown) { + telemetry.event('channel.exited', { + 'qwen-code.daemon.channel.exit_code': exitInfo?.exitCode ?? -1, + 'qwen-code.daemon.channel.session_count': sessions.length, + ...(exitInfo?.signalCode + ? { 'qwen-code.daemon.channel.signal': exitInfo.signalCode } + : {}), + }); writeStderrLine( `qwen serve: channel exited (code=${exitInfo?.exitCode ?? 'none'}, signal=${exitInfo?.signalCode ?? 'none'}, ${sessions.length} session(s) torn down)`, ); @@ -1104,16 +1143,23 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // shutdown) only need to mark dying + kill — the handler does // the alive-set cleanup when the OS reaps the child. try { - await withTimeout( - connection.initialize({ - protocolVersion: PROTOCOL_VERSION, - clientCapabilities: { - fs: { readTextFile: true, writeTextFile: true }, - }, - clientInfo: { name: 'qwen-serve-bridge', version: '0' }, - }), - initTimeoutMs, - 'initialize', + await telemetry.withSpan( + 'channel.initialize', + { + 'qwen-code.daemon.bridge.operation': 'channel.initialize', + }, + async () => + await withTimeout( + connection.initialize({ + protocolVersion: PROTOCOL_VERSION, + clientCapabilities: { + fs: { readTextFile: true, writeTextFile: true }, + }, + clientInfo: { name: 'qwen-serve-bridge', version: '0' }, + }), + initTimeoutMs, + 'initialize', + ), ); } catch (err) { // Mark the half-initialized channel as dying/unavailable, then @@ -1185,13 +1231,21 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { const ci = await ensureChannel(); let newSessionResp: { sessionId: string }; try { - newSessionResp = await withTimeout( - ci.connection.newSession({ - cwd: boundWorkspace, - mcpServers: [], - }), - initTimeoutMs, - 'newSession', + newSessionResp = await telemetry.withSpan( + 'session.new', + { + 'qwen-code.daemon.bridge.operation': 'session.new', + 'qwen-code.daemon.session_scope': effectiveScope, + }, + async () => + await withTimeout( + ci.connection.newSession({ + cwd: boundWorkspace, + mcpServers: [], + }), + initTimeoutMs, + 'newSession', + ), ); } catch (err) { // Only reap when this newSession was the channel's first/only @@ -2135,6 +2189,8 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { }, async sendPrompt(sessionId, req, signal, context) { + const capturedContext = telemetry.captureContext(); + const queuedAt = Date.now(); const entry = byId.get(sessionId); if (!entry) throw new SessionNotFoundError(sessionId); const originatorClientId = resolveTrustedClientId( @@ -2152,162 +2208,137 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // Force the body's sessionId to match the routing id — a client that // sent a stale id in the body would otherwise be dispatched to the // wrong agent process. - const normalized: PromptRequest = { ...req, sessionId }; - const result = entry.promptQueue.then(() => { - // If the caller aborted while we were queued behind earlier - // prompts, don't even start this one. - if (signal?.aborted) { - throw new DOMException('Prompt aborted', 'AbortError'); - } - if (originatorClientId === undefined) { - delete entry.activePromptOriginatorClientId; - } else { - entry.activePromptOriginatorClientId = originatorClientId; - } - // Echo the user prompt to the session bus so other SSE-subscribed - // clients see the input alongside the agent response. - // - // The interactive prompt path was the only one not emitting - // `user_message_chunk` — `Session#executePrompt` (the agent - // side) forwards the prompt directly to the LLM; the cron path - // (Session.ts:1402) and `HistoryReplayer` (line 65) emit it - // explicitly. Without this echo, multi-client UIs only saw - // assistant text from peer prompts — no record of who said what. - // - // Originator dedup: SDK consumers' `normalizeDaemonEvent` with - // `suppressOwnUserEcho: true` filters the echo when - // `event.originatorClientId === opts.clientId`. So the - // originator's local UI doesn't double-render its own input. - // - // Multi-modal: one envelope per content block. Non-text blocks - // pass through verbatim (the agent's Core multimodal echo is a - // separate follow-up tracked in PR #4353 §D); for now the - // common text path is the immediate fix. - // A fresh prompt starts: clear the D2 cancel-broadcast latch so a - // cancel for THIS turn can broadcast (a stale latch from the prior - // turn must not suppress it). - entry.cancelBroadcast = false; - echoPromptToSessionBus(entry, normalized, originatorClientId); - const promptPromise = entry.connection - .prompt(normalized) - .finally(() => { - delete entry.activePromptOriginatorClientId; - }); - - // Race against channel termination: if the underlying transport - // dies (child crashed, stream torn down) WHILE the prompt is in - // flight, the SDK's pending-request promise can hang because the - // wire never delivers a response. Make the prompt fail-fast in - // that case so the per-session FIFO doesn't poison the next - // queued prompt with an unbounded await. See - // `getTransportClosedReject` for the single-listener invariant. - // - // FIXME(stage-2): no absolute prompt deadline. A buggy agent - // that ignores `cancel()` while keeping the channel alive can - // hold this race open indefinitely — the abort path fires - // `cancel()` and resolves pending permissions, but the - // `promptPromise` itself only settles when the agent - // cooperates. Stage 2 should add a configurable per-prompt - // wall clock (e.g. `--prompt-deadline 30m`) into this race so - // a wedged agent can't slow-leak prompt promises. Tracked - // under #3803 follow-ups. - const racedPromise = Promise.race([ - promptPromise, - getTransportClosedReject(entry), - ]); - - // C3 (doudouOUC #4484 post-merge review): the user echo - // (`echoPromptToSessionBus`) was already published BEFORE the - // forward. If the forward itself fails (transport died, ACP child - // error) and it wasn't a user-initiated cancel that already - // broadcast, peers would be stuck having seen the echoed input with - // no response and no terminal signal — permanent silence. Emit a - // compensating `prompt_cancelled{reason:'forward_failed'}` so the - // turn visibly ends. The `…Once` latch dedups against the abort - // path (a normal cancel resolves rather than rejects, so this only - // fires on genuine forward failures). Side-effect only — the - // caller's `racedPromise` reference still surfaces the rejection. - void racedPromise - .then( - () => {}, - (err) => { - // Log the root cause — without this a production - // `prompt_cancelled{forward_failed}` has no diagnostic trail - // (transport death vs ACP child crash vs timeout). Matches - // the bridge's other `writeStderrLine` error paths. - writeStderrLine( - `sendPrompt: forward failed for session ${sessionId}: ${ - err instanceof Error ? err.message : String(err) - }`, - ); - broadcastPromptCancelledOnce( - entry, - sessionId, - originatorClientId, - 'forward_failed', - ); - cancelPendingForSession(sessionId); - entry.connection.cancel({ sessionId }).catch(() => {}); - }, - ) - .catch(() => {}); - - if (!signal) return racedPromise; - // Wire the abort: when the signal fires (e.g. SSE route's - // req.on('close')), tell the agent to wind down. ACP cancel is a - // notification — the active prompt resolves with - // stopReason: 'cancelled', then the next queued prompt can run. - // - // Also resolve any pending permission requests as `cancelled`. - // ACP spec requires `cancel` to settle outstanding - // `requestPermission` calls — `cancelSession()` already does - // this; the abort path here was missing the call. Without it, - // a client disconnecting while the agent is inside - // `requestPermission` leaves the permission promise unresolved - // forever (the agent is stuck waiting on a vote that no SSE - // subscriber will ever cast). - const onAbort = () => { - // Broadcast the cancel on the abort path too — client - // disconnect (SSE drop / tab close / laptop sleep) is the most - // common cancel trigger in production, and previously this path - // resolved permissions + forwarded ACP cancel WITHOUT telling - // peer SSE subscribers, leaving them in the exact - // silent-absence-of-chunks state this work set out to fix. - // `originatorClientId` here is the prompt's own originator (the - // client whose connection dropped). `…Once` dedups against an - // explicit `cancelSession` for the same turn (D2). - broadcastPromptCancelledOnce(entry, sessionId, originatorClientId); - cancelPendingForSession(sessionId); - entry.connection.cancel({ sessionId }).catch(() => { - // Cancel is fire-and-forget; the agent may already be dead. - }); - }; - if (signal.aborted) { - onAbort(); - } else { - signal.addEventListener('abort', onAbort, { once: true }); - // The aborted state can flip synchronously between the early-exit - // check at the top of `sendPrompt` and addEventListener — re-check - // after registration so a microsecond-window abort still fires - // `cancel()` instead of letting the prompt run uncancellable. - if (signal.aborted) onAbort(); - // Detach the listener once the prompt resolves so the - // AbortController can be GC'd. The `.finally()` returns a - // promise chained on `racedPromise`; if `racedPromise` - // rejects, that returned promise rejects too — and we - // never await it, so under Node's default - // unhandled-rejection behavior the daemon could terminate - // even though the route's own catch handles the original - // rejection. Attach `.catch(() => {})` to the - // listener-cleanup chain only — the caller's reference to - // `racedPromise` (via `return racedPromise` below) still - // surfaces failures normally. - racedPromise - .finally(() => signal.removeEventListener('abort', onAbort)) - .catch(() => {}); - } - return racedPromise; + const normalized: PromptRequest = telemetry.injectPromptContext({ + ...req, + sessionId, }); + const result = entry.promptQueue.then(() => + telemetry.runWithContext( + capturedContext, + async () => + await telemetry.withSpan( + 'prompt.dispatch', + { + 'qwen-code.daemon.bridge.operation': 'prompt.dispatch', + 'session.id': sessionId, + 'qwen-code.daemon.prompt.queue_wait_ms': Date.now() - queuedAt, + }, + async () => { + // If the caller aborted while we were queued behind earlier + // prompts, don't even start this one. + if (signal?.aborted) { + throw new DOMException('Prompt aborted', 'AbortError'); + } + if (originatorClientId === undefined) { + delete entry.activePromptOriginatorClientId; + } else { + entry.activePromptOriginatorClientId = originatorClientId; + } + // Echo the user prompt to the session bus so other SSE-subscribed + // clients see the input alongside the agent response. + // + // The interactive prompt path was the only one not emitting + // `user_message_chunk` — `Session#executePrompt` (the agent + // side) forwards the prompt directly to the LLM; the cron path + // (Session.ts:1402) and `HistoryReplayer` (line 65) emit it + // explicitly. Without this echo, multi-client UIs only saw + // assistant text from peer prompts — no record of who said what. + // + // Originator dedup: SDK consumers' `normalizeDaemonEvent` with + // `suppressOwnUserEcho: true` filters the echo when + // `event.originatorClientId === opts.clientId`. So the + // originator's local UI doesn't double-render its own input. + // + // Multi-modal: one envelope per content block. Non-text blocks + // pass through verbatim (the agent's Core multimodal echo is a + // separate follow-up tracked in PR #4353 §D); for now the + // common text path is the immediate fix. + entry.cancelBroadcast = false; + echoPromptToSessionBus(entry, normalized, originatorClientId); + const promptPromise = entry.connection + .prompt(normalized) + .finally(() => { + delete entry.activePromptOriginatorClientId; + }); + + // Race against channel termination: if the underlying transport + // dies (child crashed, stream torn down) WHILE the prompt is in + // flight, the SDK's pending-request promise can hang because the + // wire never delivers a response. Make the prompt fail-fast in + // that case so the per-session FIFO doesn't poison the next + // queued prompt with an unbounded await. See + // `getTransportClosedReject` for the single-listener invariant. + // + // FIXME(stage-2): no absolute prompt deadline. A buggy agent + // that ignores `cancel()` while keeping the channel alive can + // hold this race open indefinitely — the abort path fires + // `cancel()` and resolves pending permissions, but the + // `promptPromise` itself only settles when the agent + // cooperates. Stage 2 should add a configurable per-prompt + // wall clock (e.g. `--prompt-deadline 30m`) into this race so + // a wedged agent can't slow-leak prompt promises. Tracked + // under #3803 follow-ups. + const racedPromise = Promise.race([ + promptPromise, + getTransportClosedReject(entry), + ]); + + // C3 (doudouOUC #4484 post-merge review): the user echo + // (`echoPromptToSessionBus`) was already published BEFORE the + // forward. If the forward itself fails (transport died, ACP child + // error) and it wasn't a user-initiated cancel that already + // broadcast, peers would be stuck having seen the echoed input with + // no response and no terminal signal — permanent silence. Emit a + // compensating `prompt_cancelled{reason:'forward_failed'}` so the + // turn visibly ends. The `…Once` latch dedups against the abort + // path (a normal cancel resolves rather than rejects, so this only + // fires on genuine forward failures). Side-effect only — the + // caller's `racedPromise` reference still surfaces the rejection. + void racedPromise + .then( + () => {}, + (err) => { + writeStderrLine( + `sendPrompt: forward failed for session ${sessionId}: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + broadcastPromptCancelledOnce( + entry, + sessionId, + originatorClientId, + 'forward_failed', + ); + cancelPendingForSession(sessionId); + entry.connection.cancel({ sessionId }).catch(() => {}); + }, + ) + .catch(() => {}); + + if (!signal) return racedPromise; + const onAbort = () => { + broadcastPromptCancelledOnce( + entry, + sessionId, + originatorClientId, + ); + cancelPendingForSession(sessionId); + entry.connection.cancel({ sessionId }).catch(() => {}); + }; + if (signal.aborted) { + onAbort(); + } else { + signal.addEventListener('abort', onAbort, { once: true }); + if (signal.aborted) onAbort(); + racedPromise + .finally(() => signal.removeEventListener('abort', onAbort)) + .catch(() => {}); + } + return racedPromise; + }, + ), + ), + ); const promptId = context?.promptId; result.then( (promptResult) => { @@ -2392,12 +2423,21 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { const notif: CancelNotification = req ? { ...req, sessionId } : { sessionId }; - try { - await entry.connection.cancel(notif); - } catch (err) { - if (isNotCurrentlyGeneratingCancelError(err)) return; - throw err; - } + await telemetry.withSpan( + 'session.cancel', + { + 'qwen-code.daemon.bridge.operation': 'session.cancel', + 'session.id': sessionId, + }, + async () => { + try { + await entry.connection.cancel(notif); + } catch (err) { + if (isNotCurrentlyGeneratingCancelError(err)) return; + throw err; + } + }, + ); }, subscribeEvents(sessionId, subOpts) { @@ -2587,6 +2627,10 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { ? ` by client ${JSON.stringify(originatorClientId)}` : ''), ); + telemetry.event('session.close', { + 'qwen-code.daemon.bridge.operation': 'session.close', + 'session.id': sessionId, + }); if (defaultEntry === entry) defaultEntry = undefined; // #4325 fix: resolve the channel via `channelInfoForEntry(entry)` // (search `aliveChannels` for the entry's actual channel) instead @@ -2665,7 +2709,15 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // late cancellation frames from the agent are intentionally dropped. entry.events.close(); try { - await entry.connection.cancel({ sessionId }); + await telemetry.withSpan( + 'session.close.cancel_active_prompt', + { + 'qwen-code.daemon.bridge.operation': + 'session.close.cancel_active_prompt', + 'session.id': sessionId, + }, + async () => await entry.connection.cancel({ sessionId }), + ); } catch { /* no active prompt or session already torn down */ } diff --git a/packages/acp-bridge/src/bridgeOptions.ts b/packages/acp-bridge/src/bridgeOptions.ts index f237594bf47..bb73ad7d450 100644 --- a/packages/acp-bridge/src/bridgeOptions.ts +++ b/packages/acp-bridge/src/bridgeOptions.ts @@ -92,6 +92,23 @@ export interface DaemonStatusProvider { ): Promise; } +export type BridgeTelemetryAttributes = Record< + string, + string | number | boolean +>; + +export interface BridgeTelemetry { + captureContext(): unknown; + runWithContext(captured: unknown, fn: () => Promise): Promise; + withSpan( + operation: string, + attributes: BridgeTelemetryAttributes, + fn: () => Promise, + ): Promise; + event(name: string, attributes: BridgeTelemetryAttributes): void; + injectPromptContext(request: T): T; +} + /** * Construction options for `createHttpAcpBridge`. Most fields are * tuning knobs with sensible defaults; `boundWorkspace` is the only @@ -266,6 +283,8 @@ export interface BridgeOptions { * still query the routes; they'll see empty/idle cells. */ statusProvider?: DaemonStatusProvider; + /** Optional daemon telemetry seam. Omitted callers get no-op spans/logs. */ + telemetry?: BridgeTelemetry; /** * Optional fs injection seam (#4175 PR F1 step 5, originally the diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index 0860c086b75..4882646f6e5 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/packages/cli/src/acp-integration/session/Session.ts @@ -72,6 +72,8 @@ import { recordFallbackApprove, shouldFallback, shouldRunAutoModeForCall, + extractDaemonTraceContext, + withInteractionSpan, } from '@qwen-code/qwen-code-core'; import { getCommandSubcommandNames } from '../../services/commandMetadata.js'; import { getEffectiveSupportedModes } from '../../services/commandUtils.js'; @@ -687,270 +689,290 @@ export class Session implements SessionContext { this.turn += 1; const promptId = this.config.getSessionId() + '########' + this.turn; + const parentContext = extractDaemonTraceContext(params); - // Extract text from all text blocks to construct the full prompt text for logging - const promptText = params.prompt - .filter((block) => block.type === 'text') - .map((block) => (block.type === 'text' ? block.text : '')) - .join(' '); - - // Log user prompt - logUserPrompt( + return await withInteractionSpan( this.config, - new UserPromptEvent( - promptText.length, + { promptId, - this.config.getContentGeneratorConfig()?.authType, - promptText, - ), - ); - - // record user message for session management - this.config.getChatRecordingService()?.recordUserMessage(promptText); - - // Check if the input contains a slash command - // Extract text from the first text block if present - const firstTextBlock = params.prompt.find( - (block) => block.type === 'text', - ); - const inputText = firstTextBlock?.text || ''; - - let parts: Part[] | null; - - if (isSlashCommand(inputText)) { - // Handle slash command in ACP mode using capability-based filtering - const slashCommandResult = await handleSlashCommand( - inputText, - pendingSend, - this.config, - this.settings, - ); + model: this.config.getModel(), + messageType: 'acp_prompt', + ...(parentContext ? { parentContext } : {}), + }, + async () => { + // Extract text from all text blocks to construct the full prompt text for logging + const promptText = params.prompt + .filter((block) => block.type === 'text') + .map((block) => (block.type === 'text' ? block.text : '')) + .join(' '); + + // Log user prompt + logUserPrompt( + this.config, + new UserPromptEvent( + promptText.length, + promptId, + this.config.getContentGeneratorConfig()?.authType, + promptText, + ), + ); - parts = await this.#processSlashCommandResult( - slashCommandResult, - params.prompt, - ); + // record user message for session management + this.config + .getChatRecordingService() + ?.recordUserMessage(promptText); - // If parts is null, the command was fully handled (e.g., /summary completed) - // Return early without sending to the model - if (parts === null) { - return { stopReason: 'end_turn' }; - } - } else { - // Normal processing for non-slash commands - parts = await this.#resolvePrompt(params.prompt, pendingSend.signal); - } + // Check if the input contains a slash command + // Extract text from the first text block if present + const firstTextBlock = params.prompt.find( + (block) => block.type === 'text', + ); + const inputText = firstTextBlock?.text || ''; - // Fire UserPromptSubmit hook through MessageBus (aligned with core path in client.ts) - const hooksEnabled = !this.config.getDisableAllHooks?.(); - const messageBus = this.config.getMessageBus?.(); - if ( - hooksEnabled && - messageBus && - this.config.hasHooksForEvent?.('UserPromptSubmit') - ) { - const response = await messageBus.request< - HookExecutionRequest, - HookExecutionResponse - >( - { - type: MessageBusType.HOOK_EXECUTION_REQUEST, - eventName: 'UserPromptSubmit', - input: { - prompt: promptText, - }, - signal: pendingSend.signal, - }, - MessageBusType.HOOK_EXECUTION_RESPONSE, - ); - const hookOutput = response.output - ? createHookOutput('UserPromptSubmit', response.output) - : undefined; + let parts: Part[] | null; - if ( - hookOutput?.isBlockingDecision() || - hookOutput?.shouldStopExecution() - ) { - // Hook blocked the prompt - send notification to UI and return - const blockReason = - hookOutput?.getEffectiveReason() || 'No reason provided'; - await this.messageEmitter.emitAgentMessage( - `🚫 **UserPromptSubmit blocked**: ${blockReason}`, - ); - return { stopReason: 'end_turn' }; - } + if (isSlashCommand(inputText)) { + // Handle slash command in ACP mode using capability-based filtering + const slashCommandResult = await handleSlashCommand( + inputText, + pendingSend, + this.config, + this.settings, + ); - // Add additional context from hooks to the request - const additionalContext = hookOutput?.getAdditionalContext(); - if (additionalContext) { - parts = [...parts, { text: additionalContext }]; - } - } + parts = await this.#processSlashCommandResult( + slashCommandResult, + params.prompt, + ); - // Prepend session-level system reminders (plan mode / subagent / - // arena) so the model sees them, matching the behaviour of - // `GeminiClient.sendMessageStream` in the CLI/TUI path. Without this, - // plan mode in ACP has no effect because the model never learns it - // should avoid edits (#1151). - const systemReminders = await this.#buildInitialSystemReminders(); - if (systemReminders.length > 0) { - parts = [...systemReminders, ...parts]; - } + // If parts is null, the command was fully handled (e.g., /summary completed) + // Return early without sending to the model + if (parts === null) { + return { stopReason: 'end_turn' }; + } + } else { + // Normal processing for non-slash commands + parts = await this.#resolvePrompt( + params.prompt, + pendingSend.signal, + ); + } - // Phase C: one-shot worktree restore notice, set by acpAgent on - // --resume / loadSession when the session's worktree is still alive. - // Prepended exactly once, then cleared so it doesn't repeat on - // subsequent turns. - if (this.pendingWorktreeNotice) { - parts = [ - { - text: `\n${this.pendingWorktreeNotice}\n\n\n`, - }, - ...parts, - ]; - this.pendingWorktreeNotice = null; - } + // Fire UserPromptSubmit hook through MessageBus (aligned with core path in client.ts) + const hooksEnabled = !this.config.getDisableAllHooks?.(); + const messageBus = this.config.getMessageBus?.(); + if ( + hooksEnabled && + messageBus && + this.config.hasHooksForEvent?.('UserPromptSubmit') + ) { + const response = await messageBus.request< + HookExecutionRequest, + HookExecutionResponse + >( + { + type: MessageBusType.HOOK_EXECUTION_REQUEST, + eventName: 'UserPromptSubmit', + input: { + prompt: promptText, + }, + signal: pendingSend.signal, + }, + MessageBusType.HOOK_EXECUTION_RESPONSE, + ); + const hookOutput = response.output + ? createHookOutput('UserPromptSubmit', response.output) + : undefined; - let nextMessage: Content | null = { role: 'user', parts }; + if ( + hookOutput?.isBlockingDecision() || + hookOutput?.shouldStopExecution() + ) { + // Hook blocked the prompt - send notification to UI and return + const blockReason = + hookOutput?.getEffectiveReason() || 'No reason provided'; + await this.messageEmitter.emitAgentMessage( + `🚫 **UserPromptSubmit blocked**: ${blockReason}`, + ); + return { stopReason: 'end_turn' }; + } - while (nextMessage !== null) { - if (pendingSend.signal.aborted) { - this.#getCurrentChat().addHistory(nextMessage); - return { stopReason: 'cancelled' }; - } + // Add additional context from hooks to the request + const additionalContext = hookOutput?.getAdditionalContext(); + if (additionalContext) { + parts = [...parts, { text: additionalContext }]; + } + } - const functionCalls: FunctionCall[] = []; - let usageMetadata: GenerateContentResponseUsageMetadata | null = null; - const streamStartTime = Date.now(); + // Prepend session-level system reminders (plan mode / subagent / + // arena) so the model sees them, matching the behaviour of + // `GeminiClient.sendMessageStream` in the CLI/TUI path. Without this, + // plan mode in ACP has no effect because the model never learns it + // should avoid edits (#1151). + const systemReminders = await this.#buildInitialSystemReminders(); + if (systemReminders.length > 0) { + parts = [...systemReminders, ...parts]; + } - try { - const sendResult = await this.#sendMessageStreamWithAutoCompression( - promptId, - nextMessage?.parts ?? [], - pendingSend.signal, - ); - if (!sendResult.responseStream) { - this.#preserveUnsentMessageHistory( - nextMessage, - sendResult.stopReason === 'cancelled', - ); - return { stopReason: sendResult.stopReason }; + // Phase C: one-shot worktree restore notice, set by acpAgent on + // --resume / loadSession when the session's worktree is still alive. + // Prepended exactly once, then cleared so it doesn't repeat on + // subsequent turns. + if (this.pendingWorktreeNotice) { + parts = [ + { + text: `\n${this.pendingWorktreeNotice}\n\n\n`, + }, + ...parts, + ]; + this.pendingWorktreeNotice = null; } - const responseStream = sendResult.responseStream; - nextMessage = null; - for await (const resp of responseStream) { + let nextMessage: Content | null = { role: 'user', parts }; + + while (nextMessage !== null) { if (pendingSend.signal.aborted) { + this.#getCurrentChat().addHistory(nextMessage); return { stopReason: 'cancelled' }; } - if ( - resp.type === StreamEventType.CHUNK && - resp.value.candidates && - resp.value.candidates.length > 0 - ) { - const candidate = resp.value.candidates[0]; - for (const part of candidate.content?.parts ?? []) { - if (!part.text) { - continue; + const functionCalls: FunctionCall[] = []; + let usageMetadata: GenerateContentResponseUsageMetadata | null = + null; + const streamStartTime = Date.now(); + + try { + const sendResult = + await this.#sendMessageStreamWithAutoCompression( + promptId, + nextMessage?.parts ?? [], + pendingSend.signal, + ); + if (!sendResult.responseStream) { + this.#preserveUnsentMessageHistory( + nextMessage, + sendResult.stopReason === 'cancelled', + ); + return { stopReason: sendResult.stopReason }; + } + const responseStream = sendResult.responseStream; + nextMessage = null; + + for await (const resp of responseStream) { + if (pendingSend.signal.aborted) { + return { stopReason: 'cancelled' }; } - this.messageEmitter.emitMessage( - part.text, - 'assistant', - part.thought, + if ( + resp.type === StreamEventType.CHUNK && + resp.value.candidates && + resp.value.candidates.length > 0 + ) { + const candidate = resp.value.candidates[0]; + for (const part of candidate.content?.parts ?? []) { + if (!part.text) { + continue; + } + + this.messageEmitter.emitMessage( + part.text, + 'assistant', + part.thought, + ); + } + } + + if ( + resp.type === StreamEventType.CHUNK && + resp.value.usageMetadata + ) { + usageMetadata = resp.value.usageMetadata; + } + + if ( + resp.type === StreamEventType.CHUNK && + resp.value.functionCalls + ) { + functionCalls.push(...resp.value.functionCalls); + } + } + } catch (error) { + // Fire StopFailure hook (fire-and-forget, replaces Stop event for API errors) + // Aligned with useGeminiStream.ts handleFinishedWithErrorEvent + const errorStatus = getErrorStatus(error); + const errorMessage = + error instanceof Error ? error.message : String(error); + const errorType = classifyApiError({ + message: errorMessage, + status: errorStatus, + }); + + const hookSystem = this.config.getHookSystem?.(); + const hooksEnabledForStopFailure = + !this.config.getDisableAllHooks?.(); + if ( + hooksEnabledForStopFailure && + hookSystem && + this.config.hasHooksForEvent?.('StopFailure') + ) { + // Fire-and-forget: don't wait for hook to complete + hookSystem + .fireStopFailureEvent(errorType, errorMessage) + .catch((err) => { + debugLogger.warn(`StopFailure hook failed: ${err}`); + }); + } + + if (errorStatus === 429) { + throw new RequestError( + 429, + 'Rate limit exceeded. Try again later.', ); } - } - if ( - resp.type === StreamEventType.CHUNK && - resp.value.usageMetadata - ) { - usageMetadata = resp.value.usageMetadata; + throw error; } - if ( - resp.type === StreamEventType.CHUNK && - resp.value.functionCalls - ) { - functionCalls.push(...resp.value.functionCalls); - } - } - } catch (error) { - // Fire StopFailure hook (fire-and-forget, replaces Stop event for API errors) - // Aligned with useGeminiStream.ts handleFinishedWithErrorEvent - const errorStatus = getErrorStatus(error); - const errorMessage = - error instanceof Error ? error.message : String(error); - const errorType = classifyApiError({ - message: errorMessage, - status: errorStatus, - }); + if (usageMetadata) { + this.#recordPromptTokenCount(usageMetadata); + // Kick off rewrite in background (non-blocking, runs parallel to tools) + if (this.messageRewriter) { + this.messageRewriter.flushTurn(pendingSend.signal); + } - const hookSystem = this.config.getHookSystem?.(); - const hooksEnabledForStopFailure = - !this.config.getDisableAllHooks?.(); - if ( - hooksEnabledForStopFailure && - hookSystem && - this.config.hasHooksForEvent?.('StopFailure') - ) { - // Fire-and-forget: don't wait for hook to complete - hookSystem - .fireStopFailureEvent(errorType, errorMessage) - .catch((err) => { - debugLogger.warn(`StopFailure hook failed: ${err}`); - }); - } + const durationMs = Date.now() - streamStartTime; + await this.messageEmitter.emitUsageMetadata( + usageMetadata, + '', + durationMs, + ); + } - if (errorStatus === 429) { - throw new RequestError( - 429, - 'Rate limit exceeded. Try again later.', - ); + if (functionCalls.length > 0) { + const toolResponseParts = await this.runToolCalls( + pendingSend.signal, + promptId, + functionCalls, + ); + nextMessage = { role: 'user', parts: toolResponseParts }; + } } - throw error; - } - - if (usageMetadata) { - this.#recordPromptTokenCount(usageMetadata); - // Kick off rewrite in background (non-blocking, runs parallel to tools) + // Wait for any pending rewrite before returning if (this.messageRewriter) { - this.messageRewriter.flushTurn(pendingSend.signal); + await this.messageRewriter.waitForPendingRewrites(); } - const durationMs = Date.now() - streamStartTime; - await this.messageEmitter.emitUsageMetadata( - usageMetadata, - '', - durationMs, - ); - } - - if (functionCalls.length > 0) { - const toolResponseParts = await this.runToolCalls( - pendingSend.signal, + // Fire Stop hook loop (aligned with core path in client.ts) + // This is triggered after model response completes with no pending tool calls + return this.#handleStopHookLoop( + pendingSend, promptId, - functionCalls, + hooksEnabled, + messageBus, ); - nextMessage = { role: 'user', parts: toolResponseParts }; - } - } - - // Wait for any pending rewrite before returning - if (this.messageRewriter) { - await this.messageRewriter.waitForPendingRewrites(); - } - - // Fire Stop hook loop (aligned with core path in client.ts) - // This is triggered after model response completes with no pending tool calls - return this.#handleStopHookLoop( - pendingSend, - promptId, - hooksEnabled, - messageBus, + }, + (result) => (result.stopReason === 'cancelled' ? 'cancelled' : 'ok'), ); }, ); diff --git a/packages/cli/src/serve/runQwenServe.ts b/packages/cli/src/serve/runQwenServe.ts index cc4f5dd911b..f959df284af 100644 --- a/packages/cli/src/serve/runQwenServe.ts +++ b/packages/cli/src/serve/runQwenServe.ts @@ -16,6 +16,17 @@ import { createHttpAcpBridge, type HttpAcpBridge, } from './httpAcpBridge.js'; +import { + DEFAULT_OTLP_ENDPOINT, + DEFAULT_TELEMETRY_TARGET, + createDaemonBridgeTelemetry, + hashDaemonWorkspace, + initializeTelemetry, + resolveTelemetrySettings, + shutdownTelemetry, + type TelemetryRuntimeConfig, + type TelemetrySettings, +} from '@qwen-code/qwen-code-core'; import { createBridgeFileSystemAdapter } from './bridgeFileSystemAdapter.js'; import { createDaemonStatusProvider } from './daemonStatusProvider.js'; import { isLoopbackBind } from './loopbackBinds.js'; @@ -31,6 +42,7 @@ import { SERVE_CAPABILITY_REGISTRY } from './capabilities.js'; import type { ServeOptions } from './types.js'; import type { WorkspaceFileSystemFactory } from './fs/index.js'; import type { PermissionPolicy } from '@qwen-code/acp-bridge'; +import { getCliVersion } from '../utils/version.js'; const QWEN_SERVER_TOKEN_ENV = 'QWEN_SERVER_TOKEN'; const QWEN_SERVE_PROMPT_DEADLINE_MS_ENV = 'QWEN_SERVE_PROMPT_DEADLINE_MS'; @@ -83,6 +95,33 @@ function parseDeadlineEnv( return parsed; } +function createDaemonTelemetryRuntimeConfig( + telemetry: TelemetrySettings, + cliVersion: string, + daemonSessionId: string, +): TelemetryRuntimeConfig { + return { + getTelemetryEnabled: () => telemetry.enabled ?? false, + getTelemetryOtlpEndpoint: () => + telemetry.otlpEndpoint ?? DEFAULT_OTLP_ENDPOINT, + getTelemetryOtlpProtocol: () => telemetry.otlpProtocol ?? 'grpc', + getTelemetryOtlpTracesEndpoint: () => telemetry.otlpTracesEndpoint, + getTelemetryOtlpLogsEndpoint: () => telemetry.otlpLogsEndpoint, + getTelemetryOtlpMetricsEndpoint: () => telemetry.otlpMetricsEndpoint, + getTelemetryTarget: () => telemetry.target ?? DEFAULT_TELEMETRY_TARGET, + getTelemetryOutfile: () => telemetry.outfile, + getTelemetryIncludeSensitiveSpanAttributes: () => + telemetry.includeSensitiveSpanAttributes ?? false, + getTelemetryResourceAttributes: () => telemetry.resourceAttributes ?? {}, + getTelemetryMetricsIncludeSessionId: () => + telemetry.metrics?.includeSessionId ?? false, + getTelemetryResourceAttributeWarnings: () => + telemetry.resourceAttributeWarnings ?? [], + getCliVersion: () => cliVersion, + getSessionId: () => daemonSessionId, + }; +} + /** * Wenshao review #4335 / 3271978374 — boot-time policy validation * errors. Replaces the previous substring-matching of "invalid @@ -624,8 +663,9 @@ export async function runQwenServe( let contextFilenameForInit: string | undefined; let permissionPolicy: PermissionPolicy | undefined; let permissionConsensusQuorum: number | undefined; + let bootSettings: ReturnType | undefined; try { - const bootSettings = loadSettings(boundWorkspace); + bootSettings = loadSettings(boundWorkspace); contextFilenameForInit = extractContextFilename( bootSettings.merged.context?.fileName, ); @@ -660,6 +700,20 @@ export async function runQwenServe( ); } + const daemonWorkspaceHash = hashDaemonWorkspace(boundWorkspace); + const daemonTelemetrySettings = await resolveTelemetrySettings({ + env: process.env, + settings: bootSettings?.merged.telemetry, + }); + initializeTelemetry( + createDaemonTelemetryRuntimeConfig( + daemonTelemetrySettings, + await getCliVersion(), + `daemon:${daemonWorkspaceHash}:${process.pid}`, + ), + ); + const daemonTelemetry = createDaemonBridgeTelemetry(); + // F3 Commit 2 — allocate the audit ring + publisher in the daemon // host (here) rather than inside the bridge factory, because the // ring is the seam future PRs will lift up to expose `GET @@ -710,6 +764,7 @@ export async function runQwenServe( childEnvOverrides, channelFactory, onDiagnosticLine: diagnosticSink, + telemetry: daemonTelemetry, // F3 Commit 5 — wire the validated policy/quorum from // settings into the bridge. Bridge factory does its own // defensive `Number.isInteger` recheck on the quorum so a @@ -1032,15 +1087,27 @@ export async function runQwenServe( const finish = (err?: Error | null) => { if (settled) return; settled = true; - // Drain finished (or timed out) — safe to detach now. - process.removeListener('SIGINT', onSignal); - process.removeListener('SIGTERM', onSignal); - // Server.close error takes precedence (operator-visible - // listener problem); fall back to the bridge error - // captured during shutdown if any. - const finalErr = err ?? bridgeShutdownError; - if (finalErr) rej(finalErr); - else res(); + void shutdownTelemetry() + .catch((telemetryErr) => { + writeStderrLine( + `qwen serve: telemetry shutdown error: ${ + telemetryErr instanceof Error + ? telemetryErr.message + : String(telemetryErr) + }`, + ); + }) + .finally(() => { + // Drain finished (or timed out) — safe to detach now. + process.removeListener('SIGINT', onSignal); + process.removeListener('SIGTERM', onSignal); + // Server.close error takes precedence (operator-visible + // listener problem); fall back to the bridge error + // captured during shutdown if any. + const finalErr = err ?? bridgeShutdownError; + if (finalErr) rej(finalErr); + else res(); + }); }; // PR 21: dispose the device-flow registry FIRST so any diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 07ba77d81df..6d4a5a0a174 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -7,12 +7,17 @@ import * as crypto from 'node:crypto'; import * as path from 'node:path'; import express from 'express'; -import type { Application } from 'express'; +import type { Application, NextFunction, Request, Response } from 'express'; import type { ApprovalMode } from '@qwen-code/qwen-code-core'; import { APPROVAL_MODES, SessionService, TrustGateError, + emitDaemonLog, + hashDaemonWorkspace, + recordDaemonError, + recordDaemonHttpResponse, + withDaemonRequestSpan, } from '@qwen-code/qwen-code-core'; import { writeStderrLine } from '../utils/stdioHelpers.js'; import type { DaemonLogger } from './daemonLogger.js'; @@ -272,6 +277,73 @@ export interface ServeAppDeps { daemonLog?: DaemonLogger; } +function resolveDaemonTelemetryRoute( + req: Request, +): { route: string; sessionId?: string } | undefined { + if (req.method === 'POST' && req.path === '/session') { + return { route: 'POST /session' }; + } + const sessionAction = req.path.match( + /^\/session\/([^/]+)\/(load|resume|prompt|cancel)$/, + ); + const sessionActionId = sessionAction?.[1]; + const sessionActionName = sessionAction?.[2]; + if (sessionActionId && sessionActionName && req.method === 'POST') { + return { + route: `POST /session/:id/${sessionActionName}`, + sessionId: sessionActionId, + }; + } + const deleteSession = req.path.match(/^\/session\/([^/]+)$/); + const deleteSessionId = deleteSession?.[1]; + if (deleteSessionId && req.method === 'DELETE') { + return { route: 'DELETE /session/:id', sessionId: deleteSessionId }; + } + if (req.method === 'GET' && /^\/workspace\/.+\/sessions$/.test(req.path)) { + return { route: 'GET /workspace/:id/sessions' }; + } + return undefined; +} + +function daemonTelemetryMiddleware( + boundWorkspace: string, +): (req: Request, res: Response, next: NextFunction) => void { + const workspaceHash = hashDaemonWorkspace(boundWorkspace); + return (req, res, next) => { + const route = resolveDaemonTelemetryRoute(req); + if (!route) { + next(); + return; + } + void withDaemonRequestSpan( + { + method: req.method, + route: route.route, + workspaceHash, + ...(route.sessionId ? { sessionId: route.sessionId } : {}), + }, + async (span) => + await new Promise((resolve, reject) => { + let done = false; + const finish = () => { + if (done) return; + done = true; + recordDaemonHttpResponse(span, res.statusCode); + resolve(); + }; + res.once('finish', finish); + res.once('close', finish); + try { + next(); + } catch (error) { + recordDaemonError(span, error); + reject(error); + } + }), + ).catch(next); + }; +} + /** * Issue #4514 T2.9. Sentinel passed as `AbortController.abort(reason)` * when a prompt exceeds its server-configured wallclock. The catch @@ -796,6 +868,8 @@ export function createServeApp( requireAuth: opts.requireAuth === true, }); + app.use(daemonTelemetryMiddleware(boundWorkspace)); + app.get('/capabilities', (_req, res) => { const envelope: CapabilitiesEnvelope = { v: CAPABILITIES_SCHEMA_VERSION, @@ -3368,6 +3442,16 @@ function sendBridgeErrorImpl( // structured daemon logger (which tees to stderr + log file). When // absent (tests, direct embeds), fall back to the legacy stderr-only // `writeStderrLine` path. + recordDaemonError(undefined, err, { + ...(ctx?.route ? { 'http.route': ctx.route } : {}), + ...(ctx?.sessionId ? { 'session.id': ctx.sessionId } : {}), + }); + emitDaemonLog('Daemon bridge error.', { + ...(ctx?.route ? { 'http.route': ctx.route } : {}), + ...(ctx?.sessionId ? { 'session.id': ctx.sessionId } : {}), + 'error.type': err instanceof Error ? err.name : typeof err, + 'error.message': err instanceof Error ? err.message : String(err), + }); if (daemonLog) { daemonLog.error( err instanceof Error ? err.message : String(err), diff --git a/packages/core/src/telemetry/daemon-tracing.test.ts b/packages/core/src/telemetry/daemon-tracing.test.ts new file mode 100644 index 00000000000..6185f709de8 --- /dev/null +++ b/packages/core/src/telemetry/daemon-tracing.test.ts @@ -0,0 +1,55 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, expect, it } from 'vitest'; +import { trace } from '@opentelemetry/api'; +import { + DAEMON_TRACEPARENT_META_KEY, + DAEMON_TRACESTATE_META_KEY, + extractDaemonTraceContext, + hashDaemonWorkspace, + injectDaemonTraceContext, +} from './daemon-tracing.js'; + +describe('daemon-tracing', () => { + it('extracts daemon trace context from reserved prompt metadata keys', () => { + const traceId = '1'.repeat(32); + const spanId = '2'.repeat(16); + const extracted = extractDaemonTraceContext({ + _meta: { + [DAEMON_TRACEPARENT_META_KEY]: `00-${traceId}-${spanId}-01`, + [DAEMON_TRACESTATE_META_KEY]: 'vendor=value', + }, + }); + + expect(extracted).toBeDefined(); + expect(trace.getSpanContext(extracted!)?.traceId).toBe(traceId); + expect(trace.getSpanContext(extracted!)?.spanId).toBe(spanId); + }); + + it('strips reserved metadata when no active daemon span exists', () => { + const injected = injectDaemonTraceContext({ + prompt: [], + _meta: { + keep: true, + [DAEMON_TRACEPARENT_META_KEY]: 'client-spoof', + }, + }); + + const meta = injected._meta as Record; + expect(meta['keep']).toBe(true); + expect(meta[DAEMON_TRACEPARENT_META_KEY]).toBeUndefined(); + expect(meta[DAEMON_TRACESTATE_META_KEY]).toBeUndefined(); + expect(extractDaemonTraceContext(injected)).toBeUndefined(); + }); + + it('hashes workspace paths without exposing the raw path', () => { + const hash = hashDaemonWorkspace('/tmp/project'); + + expect(hash).toMatch(/^[0-9a-f]{16}$/); + expect(hash).not.toContain('project'); + }); +}); diff --git a/packages/core/src/telemetry/daemon-tracing.ts b/packages/core/src/telemetry/daemon-tracing.ts new file mode 100644 index 00000000000..8b97726b923 --- /dev/null +++ b/packages/core/src/telemetry/daemon-tracing.ts @@ -0,0 +1,307 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { createHash } from 'node:crypto'; +import { + context as otelContext, + propagation, + SpanKind, + SpanStatusCode, + trace, + type Attributes, + type Context, + type Span, +} from '@opentelemetry/api'; +import { logs, type LogAttributes } from '@opentelemetry/api-logs'; +import { SERVICE_NAME } from './constants.js'; +import { isTelemetrySdkInitialized } from './sdk.js'; +import { truncateSpanError } from './session-tracing.js'; + +export const DAEMON_TRACEPARENT_META_KEY = 'qwen.telemetry.traceparent'; +export const DAEMON_TRACESTATE_META_KEY = 'qwen.telemetry.tracestate'; + +const SPAN_DAEMON_REQUEST = 'qwen-code.daemon.request'; +const SPAN_DAEMON_BRIDGE = 'qwen-code.daemon.bridge'; +const EVENT_DAEMON_ERROR = 'qwen-code.daemon.error'; + +type DaemonAttributes = Record; + +interface CapturedDaemonContext { + context: Context; +} + +export interface DaemonRequestSpanOptions { + method: string; + route: string; + workspaceHash?: string; + sessionId?: string; +} + +function toOtelAttributes(attrs: DaemonAttributes): Attributes { + return attrs; +} + +function errorMessage(error: unknown): string { + if (error instanceof Error) return error.message; + return String(error); +} + +function errorType(error: unknown): string { + if (error instanceof Error) return error.name || 'Error'; + return typeof error; +} + +function activeSpanContextIsValid(): boolean { + const span = trace.getSpan(otelContext.active()); + if (!span) return false; + const ctx = span.spanContext(); + return ctx.traceId !== '0'.repeat(32) && ctx.spanId !== '0'.repeat(16); +} + +function stripReservedTraceMeta(meta: unknown): Record { + const out = + meta && typeof meta === 'object' && !Array.isArray(meta) + ? { ...(meta as Record) } + : {}; + delete out[DAEMON_TRACEPARENT_META_KEY]; + delete out[DAEMON_TRACESTATE_META_KEY]; + return out; +} + +export function hashDaemonWorkspace(workspace: string): string { + return createHash('sha256').update(workspace).digest('hex').slice(0, 16); +} + +export async function withDaemonSpan( + name: string, + attributes: DaemonAttributes, + fn: (span: Span) => Promise, + options: { autoOkOnSuccess?: boolean } = {}, +): Promise { + const autoOkOnSuccess = options.autoOkOnSuccess ?? true; + const tracer = trace.getTracer(SERVICE_NAME); + return await tracer.startActiveSpan( + name, + { kind: SpanKind.INTERNAL, attributes: toOtelAttributes(attributes) }, + async (span) => { + try { + const result = await fn(span); + if (autoOkOnSuccess) { + span.setStatus({ code: SpanStatusCode.OK }); + } + return result; + } catch (error) { + recordDaemonError(span, error); + throw error; + } finally { + span.end(); + } + }, + ); +} + +export async function withDaemonRequestSpan( + options: DaemonRequestSpanOptions, + fn: (span: Span) => Promise, +): Promise { + return await withDaemonSpan( + SPAN_DAEMON_REQUEST, + { + 'http.request.method': options.method, + 'http.route': options.route, + 'qwen-code.daemon.operation': 'http_request', + ...(options.workspaceHash + ? { 'qwen-code.workspace.hash': options.workspaceHash } + : {}), + ...(options.sessionId ? { 'session.id': options.sessionId } : {}), + }, + fn, + { autoOkOnSuccess: false }, + ); +} + +export async function withDaemonBridgeSpan( + operation: string, + attributes: DaemonAttributes, + fn: () => Promise, +): Promise { + return await withDaemonSpan( + SPAN_DAEMON_BRIDGE, + { + 'qwen-code.daemon.operation': operation, + ...attributes, + }, + async () => await fn(), + ); +} + +export function recordDaemonHttpResponse( + span: Span | undefined, + statusCode: number, +): void { + try { + span?.setAttribute('http.response.status_code', statusCode); + if (statusCode >= 500) { + span?.setStatus({ + code: SpanStatusCode.ERROR, + message: `HTTP ${statusCode}`, + }); + } else { + span?.setStatus({ code: SpanStatusCode.OK }); + } + } catch { + // Telemetry must not affect request handling. + } +} + +export function recordDaemonError( + span: Span | undefined, + error: unknown, + attributes: DaemonAttributes = {}, +): void { + const target = span ?? trace.getSpan(otelContext.active()); + if (!target) return; + try { + const message = truncateSpanError(errorMessage(error)); + target.recordException(error instanceof Error ? error : new Error(message)); + target.setAttributes({ + 'error.type': errorType(error), + 'error.message': message, + ...attributes, + }); + target.setStatus({ code: SpanStatusCode.ERROR, message }); + } catch { + // Telemetry must not affect request handling. + } +} + +export function emitDaemonLog( + body: string, + attributes: LogAttributes = {}, +): void { + if (!isTelemetrySdkInitialized()) return; + try { + logs.getLogger(SERVICE_NAME).emit({ + body, + attributes: { + 'event.name': EVENT_DAEMON_ERROR, + 'event.timestamp': new Date().toISOString(), + ...attributes, + }, + }); + } catch { + // Telemetry must not affect daemon behavior. + } +} + +export function captureDaemonTelemetryContext(): CapturedDaemonContext { + return { context: otelContext.active() }; +} + +export async function runWithDaemonTelemetryContext( + captured: unknown, + fn: () => Promise, +): Promise { + const ctx = + captured && + typeof captured === 'object' && + 'context' in captured && + (captured as CapturedDaemonContext).context + ? (captured as CapturedDaemonContext).context + : undefined; + if (!ctx) return await fn(); + return await otelContext.with(ctx, fn); +} + +export function injectDaemonTraceContext(request: T): T { + const currentMeta = (request as { _meta?: unknown })._meta; + const nextMeta = stripReservedTraceMeta(currentMeta); + + if (activeSpanContextIsValid()) { + const carrier: Record = {}; + propagation.inject(otelContext.active(), carrier); + if (carrier['traceparent']) { + nextMeta[DAEMON_TRACEPARENT_META_KEY] = carrier['traceparent']; + } + if (carrier['tracestate']) { + nextMeta[DAEMON_TRACESTATE_META_KEY] = carrier['tracestate']; + } + } + + return { + ...request, + _meta: nextMeta, + }; +} + +export function extractDaemonTraceContext( + source: unknown, +): Context | undefined { + const meta = (source as { _meta?: unknown } | undefined)?._meta; + if (!meta || typeof meta !== 'object' || Array.isArray(meta)) { + return undefined; + } + const record = meta as Record; + const traceparent = record[DAEMON_TRACEPARENT_META_KEY]; + if (typeof traceparent !== 'string' || traceparent.length === 0) { + return undefined; + } + const carrier: Record = { traceparent }; + const tracestate = record[DAEMON_TRACESTATE_META_KEY]; + if (typeof tracestate === 'string' && tracestate.length > 0) { + carrier['tracestate'] = tracestate; + } + const extracted = propagation.extract(otelContext.active(), carrier); + if (trace.getSpanContext(extracted)) return extracted; + + const parts = traceparent.split('-'); + const traceId = parts[1]; + const spanId = parts[2]; + const flags = parts[3]; + if ( + parts[0] !== '00' || + !traceId?.match(/^[0-9a-f]{32}$/) || + !spanId?.match(/^[0-9a-f]{16}$/) || + !flags?.match(/^[0-9a-f]{2}$/) + ) { + return undefined; + } + return trace.setSpan( + otelContext.active(), + trace.wrapSpanContext({ + traceId, + spanId, + traceFlags: Number.parseInt(flags, 16), + }), + ); +} + +export function createDaemonBridgeTelemetry(): { + captureContext(): unknown; + runWithContext(captured: unknown, fn: () => Promise): Promise; + withSpan( + operation: string, + attributes: DaemonAttributes, + fn: () => Promise, + ): Promise; + event(name: string, attributes: DaemonAttributes): void; + injectPromptContext(request: T): T; +} { + return { + captureContext: captureDaemonTelemetryContext, + runWithContext: runWithDaemonTelemetryContext, + withSpan: withDaemonBridgeSpan, + event(name, attributes) { + const span = trace.getSpan(otelContext.active()); + try { + span?.addEvent(name, attributes); + } catch { + // Telemetry must not affect bridge behavior. + } + }, + injectPromptContext: injectDaemonTraceContext, + }; +} diff --git a/packages/core/src/telemetry/index.ts b/packages/core/src/telemetry/index.ts index 6ad5cb13c34..527c18c7d0d 100644 --- a/packages/core/src/telemetry/index.ts +++ b/packages/core/src/telemetry/index.ts @@ -139,6 +139,7 @@ export { sanitizeHookName } from './sanitize.js'; export { startInteractionSpan, endInteractionSpan, + withInteractionSpan, startLLMRequestSpan, endLLMRequestSpan, startToolSpan, @@ -156,6 +157,7 @@ export { export type { StartInteractionOptions, EndInteractionOptions, + InteractionSpanResultStatus, LLMRequestMetadata, ToolSpanMetadata, ToolBlockedDecision, @@ -164,6 +166,23 @@ export type { StartHookSpanOptions, HookSpanMetadata, } from './session-tracing.js'; +export type { TelemetryRuntimeConfig } from './runtime-config.js'; +export { + DAEMON_TRACEPARENT_META_KEY, + DAEMON_TRACESTATE_META_KEY, + captureDaemonTelemetryContext, + createDaemonBridgeTelemetry, + emitDaemonLog, + extractDaemonTraceContext, + hashDaemonWorkspace, + injectDaemonTraceContext, + recordDaemonError, + recordDaemonHttpResponse, + runWithDaemonTelemetryContext, + withDaemonBridgeSpan, + withDaemonRequestSpan, + withDaemonSpan, +} from './daemon-tracing.js'; export { addUserPromptAttributes, addSystemPromptAttributes, diff --git a/packages/core/src/telemetry/metrics.ts b/packages/core/src/telemetry/metrics.ts index 7d9de142ee5..7bf0fe91bde 100644 --- a/packages/core/src/telemetry/metrics.ts +++ b/packages/core/src/telemetry/metrics.ts @@ -8,6 +8,7 @@ import type { Attributes, Meter, Counter, Histogram } from '@opentelemetry/api'; import { diag, metrics, ValueType } from '@opentelemetry/api'; import { SERVICE_NAME, EVENT_CHAT_COMPRESSION } from './constants.js'; import type { Config } from '../config/config.js'; +import type { TelemetryRuntimeConfig } from './runtime-config.js'; import type { ModelSlashCommandEvent } from './types.js'; const TOOL_CALL_COUNT = `${SERVICE_NAME}.tool.call.count`; @@ -59,7 +60,7 @@ const baseMetricDefinition = { // can enable QWEN_TELEMETRY_METRICS_INCLUDE_SESSION_ID or // telemetry.metrics.includeSessionId. Spans and logs always carry // session.id for trace/log correlation. - getCommonAttributes: (config: Config): Attributes => { + getCommonAttributes: (config: TelemetryRuntimeConfig): Attributes => { const out: Attributes = {}; if (config.getTelemetryMetricsIncludeSessionId()) { out['session.id'] = config.getSessionId(); @@ -397,7 +398,7 @@ export function getMeter(): Meter | undefined { return cliMeter; } -export function initializeMetrics(config: Config): void { +export function initializeMetrics(config: TelemetryRuntimeConfig): void { if (isMetricsInitialized) return; const meter = getMeter(); @@ -639,7 +640,9 @@ export function recordModelSlashCommand( // Performance Monitoring Functions -export function initializePerformanceMonitoring(config: Config): void { +export function initializePerformanceMonitoring( + config: TelemetryRuntimeConfig, +): void { const meter = getMeter(); if (!meter) return; diff --git a/packages/core/src/telemetry/runtime-config.ts b/packages/core/src/telemetry/runtime-config.ts new file mode 100644 index 00000000000..1ffcc221379 --- /dev/null +++ b/packages/core/src/telemetry/runtime-config.ts @@ -0,0 +1,24 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { TelemetryTarget } from './index.js'; + +export interface TelemetryRuntimeConfig { + getTelemetryEnabled(): boolean; + getTelemetryOtlpEndpoint(): string | undefined; + getTelemetryOtlpProtocol(): 'grpc' | 'http'; + getTelemetryOtlpTracesEndpoint(): string | undefined; + getTelemetryOtlpLogsEndpoint(): string | undefined; + getTelemetryOtlpMetricsEndpoint(): string | undefined; + getTelemetryTarget(): TelemetryTarget; + getTelemetryOutfile(): string | undefined; + getTelemetryIncludeSensitiveSpanAttributes(): boolean; + getTelemetryResourceAttributes(): Record; + getTelemetryMetricsIncludeSessionId(): boolean; + getTelemetryResourceAttributeWarnings(): readonly string[]; + getCliVersion(): string | undefined; + getSessionId(): string; +} diff --git a/packages/core/src/telemetry/sdk.ts b/packages/core/src/telemetry/sdk.ts index 1412d06de82..7e8be6229cc 100644 --- a/packages/core/src/telemetry/sdk.ts +++ b/packages/core/src/telemetry/sdk.ts @@ -20,7 +20,7 @@ import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-node'; import { BatchLogRecordProcessor } from '@opentelemetry/sdk-logs'; import { PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics'; import { HttpInstrumentation } from '@opentelemetry/instrumentation-http'; -import type { Config } from '../config/config.js'; +import type { TelemetryRuntimeConfig } from './runtime-config.js'; import { SERVICE_NAME } from './constants.js'; import { initializeMetrics } from './metrics.js'; import { @@ -147,7 +147,7 @@ function validateUrl(url: string | undefined): string | undefined { } } -export function initializeTelemetry(config: Config): void { +export function initializeTelemetry(config: TelemetryRuntimeConfig): void { if (telemetryInitialized || !config.getTelemetryEnabled()) { return; } diff --git a/packages/core/src/telemetry/session-tracing.test.ts b/packages/core/src/telemetry/session-tracing.test.ts index c0119982010..eb96b7e0fda 100644 --- a/packages/core/src/telemetry/session-tracing.test.ts +++ b/packages/core/src/telemetry/session-tracing.test.ts @@ -125,6 +125,7 @@ import type { Config } from '../config/config.js'; import { startInteractionSpan, endInteractionSpan, + withInteractionSpan, startLLMRequestSpan, endLLMRequestSpan, startToolSpan, @@ -191,6 +192,31 @@ describe('session-tracing', () => { expect(mockSpans[0]!.statuses[0]!.code).toBe(SpanStatusCode.OK); }); + it('runs scoped interaction spans without mutating the global interaction context', async () => { + const config = createMockConfig({ sessionId: 'scoped-session' }); + const result = await withInteractionSpan( + config, + { + promptId: 'prompt-scoped', + model: 'test-model', + messageType: 'acp_prompt', + parentContext: { parent: 'daemon' } as never, + }, + async () => 'done', + ); + + expect(result).toBe('done'); + expect(mockSpans).toHaveLength(1); + expect(mockSpans[0]!.name).toBe('qwen-code.interaction'); + expect(mockSpans[0]!.parentContext).toEqual({ parent: 'daemon' }); + expect(mockSpans[0]!.attributes['session.id']).toBe('scoped-session'); + expect(mockSpans[0]!.attributes['qwen-code.message_type']).toBe( + 'acp_prompt', + ); + expect(mockSpans[0]!.ended).toBe(true); + expect(mockSpans[0]!.statuses.at(-1)?.code).toBe(SpanStatusCode.OK); + }); + it('ends interaction span with error status', () => { const config = createMockConfig(); startInteractionSpan(config, { diff --git a/packages/core/src/telemetry/session-tracing.ts b/packages/core/src/telemetry/session-tracing.ts index 118192bea91..d712f84cd79 100644 --- a/packages/core/src/telemetry/session-tracing.ts +++ b/packages/core/src/telemetry/session-tracing.ts @@ -43,6 +43,8 @@ export interface EndInteractionOptions { errorMessage?: string; } +export type InteractionSpanResultStatus = 'ok' | 'cancelled'; + export interface LLMRequestMetadata { inputTokens?: number; outputTokens?: number; @@ -345,6 +347,83 @@ export function endInteractionSpan( interactionContext.enterWith(undefined); } +export async function withInteractionSpan( + config: Config, + options: StartInteractionOptions & { parentContext?: Context }, + fn: () => Promise, + getResultStatus?: (result: T) => InteractionSpanResultStatus, +): Promise { + if (!isTelemetrySdkInitialized()) return await fn(); + + ensureCleanupInterval(); + interactionSequence++; + + const attributes: Attributes = { + 'session.id': config.getSessionId(), + 'qwen-code.prompt_id': options.promptId, + 'qwen-code.message_type': options.messageType, + 'qwen-code.model': options.model, + 'qwen-code.approval_mode': config.getApprovalMode(), + 'interaction.sequence': interactionSequence, + }; + + const parentContext = + options.parentContext ?? resolveParentContext(undefined); + const span = getTracer().startSpan( + SPAN_INTERACTION, + { + kind: SpanKind.INTERNAL, + attributes, + }, + parentContext, + ); + const spanId = getSpanId(span); + const spanContextObj: SpanContext = { + span, + startTime: Date.now(), + attributes: attributes as Record, + type: 'interaction', + }; + activeSpans.set(spanId, new WeakRef(spanContextObj)); + strongSpans.set(spanId, spanContextObj); + + const activeContext = trace.setSpan(parentContext, span); + return await otelContext.with(activeContext, async () => + interactionContext.run(spanContextObj, async () => { + let terminalStatus: InteractionStatus = 'ok'; + try { + const result = await fn(); + terminalStatus = getResultStatus?.(result) ?? 'ok'; + return result; + } catch (error) { + terminalStatus = 'error'; + span.setStatus({ + code: SpanStatusCode.ERROR, + message: truncateSpanError( + error instanceof Error ? error.message : String(error), + ), + }); + throw error; + } finally { + if (!spanContextObj.ended) { + spanContextObj.ended = true; + const duration = Date.now() - spanContextObj.startTime; + span.setAttributes({ + 'interaction.duration_ms': duration, + 'qwen-code.turn_status': terminalStatus, + }); + if (terminalStatus !== 'error') { + span.setStatus({ code: SpanStatusCode.OK }); + } + span.end(); + activeSpans.delete(spanId); + strongSpans.delete(spanId); + } + } + }), + ); +} + // --- LLM Request Spans --- export function startLLMRequestSpan(model: string, promptId: string): Span { From 91aa3914fca62fa0a199cbcb8e03c85f3cac2c85 Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Wed, 27 May 2026 00:01:26 +0800 Subject: [PATCH 2/7] fix(telemetry): emit daemon bridge events as spans Record bridge telemetry events as short daemon bridge spans when they fire outside an active request or prompt context, so asynchronous channel exits remain observable. Co-authored-by: Qwen-Coder --- .../core/src/telemetry/daemon-tracing.test.ts | 47 ++++++++++++++++++- packages/core/src/telemetry/daemon-tracing.ts | 20 +++++++- 2 files changed, 63 insertions(+), 4 deletions(-) diff --git a/packages/core/src/telemetry/daemon-tracing.test.ts b/packages/core/src/telemetry/daemon-tracing.test.ts index 6185f709de8..34a88782827 100644 --- a/packages/core/src/telemetry/daemon-tracing.test.ts +++ b/packages/core/src/telemetry/daemon-tracing.test.ts @@ -4,17 +4,27 @@ * SPDX-License-Identifier: Apache-2.0 */ -import { describe, expect, it } from 'vitest'; -import { trace } from '@opentelemetry/api'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { + SpanStatusCode, + trace, + type Span, + type Tracer, +} from '@opentelemetry/api'; import { DAEMON_TRACEPARENT_META_KEY, DAEMON_TRACESTATE_META_KEY, + createDaemonBridgeTelemetry, extractDaemonTraceContext, hashDaemonWorkspace, injectDaemonTraceContext, } from './daemon-tracing.js'; describe('daemon-tracing', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + it('extracts daemon trace context from reserved prompt metadata keys', () => { const traceId = '1'.repeat(32); const spanId = '2'.repeat(16); @@ -52,4 +62,37 @@ describe('daemon-tracing', () => { expect(hash).toMatch(/^[0-9a-f]{16}$/); expect(hash).not.toContain('project'); }); + + it('emits bridge events as standalone spans without an active span', () => { + const addEvent = vi.fn(); + const setStatus = vi.fn(); + const end = vi.fn(); + const startSpan = vi.fn( + () => ({ addEvent, setStatus, end }) as unknown as Span, + ); + vi.spyOn(trace, 'getSpan').mockReturnValue(undefined); + vi.spyOn(trace, 'getTracer').mockReturnValue({ + startSpan, + } as unknown as Tracer); + + createDaemonBridgeTelemetry().event('channel.exited', { + 'qwen-code.daemon.channel.session_count': 2, + }); + + expect(startSpan).toHaveBeenCalledWith( + 'qwen-code.daemon.bridge', + expect.objectContaining({ + attributes: expect.objectContaining({ + 'event.name': 'channel.exited', + 'qwen-code.daemon.operation': 'event.channel.exited', + 'qwen-code.daemon.channel.session_count': 2, + }), + }), + ); + expect(addEvent).toHaveBeenCalledWith('channel.exited', { + 'qwen-code.daemon.channel.session_count': 2, + }); + expect(setStatus).toHaveBeenCalledWith({ code: SpanStatusCode.OK }); + expect(end).toHaveBeenCalled(); + }); }); diff --git a/packages/core/src/telemetry/daemon-tracing.ts b/packages/core/src/telemetry/daemon-tracing.ts index 8b97726b923..7795369b72a 100644 --- a/packages/core/src/telemetry/daemon-tracing.ts +++ b/packages/core/src/telemetry/daemon-tracing.ts @@ -295,9 +295,25 @@ export function createDaemonBridgeTelemetry(): { runWithContext: runWithDaemonTelemetryContext, withSpan: withDaemonBridgeSpan, event(name, attributes) { - const span = trace.getSpan(otelContext.active()); try { - span?.addEvent(name, attributes); + const activeSpan = trace.getSpan(otelContext.active()); + if (activeSpan) { + activeSpan.addEvent(name, attributes); + return; + } + const span = trace + .getTracer(SERVICE_NAME) + .startSpan(SPAN_DAEMON_BRIDGE, { + kind: SpanKind.INTERNAL, + attributes: { + 'event.name': name, + 'qwen-code.daemon.operation': `event.${name}`, + ...attributes, + }, + }); + span.addEvent(name, attributes); + span.setStatus({ code: SpanStatusCode.OK }); + span.end(); } catch { // Telemetry must not affect bridge behavior. } From 085aa49706f278ccaf00a8cc45db79ccbad041ff Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Wed, 27 May 2026 19:40:34 +0800 Subject: [PATCH 3/7] =?UTF-8?q?fix(telemetry):=20address=20wenshao=20revie?= =?UTF-8?q?w=20=E2=80=94=2010=20fixes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - recordDaemonHttpResponse: don't clobber ERROR with OK on non-5xx - finish(): remove signal listeners synchronously before async telemetry shutdown - extractDaemonTraceContext: reject all-zero IDs, include tracestate, set isRemote - propagation.inject: wrap in try/catch for consistency - injectPromptContext: move inside prompt.dispatch span for correct parent - withDaemonSpan: guard on isTelemetrySdkInitialized() - toOtelAttributes: remove identity function, pass attributes directly - injectDaemonTraceContext: early-return when no active span (avoid empty _meta) - emitDaemonLog: remove redundant event.timestamp attribute - NOOP_BRIDGE_TELEMETRY: drop async, add short-circuit for missing keys --- packages/acp-bridge/src/bridge.ts | 27 ++++++++----- packages/cli/src/serve/runQwenServe.ts | 5 +-- packages/core/src/telemetry/daemon-tracing.ts | 39 ++++++++++++------- 3 files changed, 46 insertions(+), 25 deletions(-) diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index ca84a4daa46..7db80af7168 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -93,11 +93,11 @@ const DAEMON_TRACESTATE_META_KEY = 'qwen.telemetry.tracestate'; const NOOP_BRIDGE_TELEMETRY: BridgeTelemetry = { captureContext: () => undefined, - async runWithContext(_captured, fn) { - return await fn(); + runWithContext(_captured, fn) { + return fn(); }, - async withSpan(_operation, _attributes, fn) { - return await fn(); + withSpan(_operation, _attributes, fn) { + return fn(); }, event() {}, injectPromptContext(request) { @@ -105,7 +105,14 @@ const NOOP_BRIDGE_TELEMETRY: BridgeTelemetry = { if (!meta || typeof meta !== 'object' || Array.isArray(meta)) { return request; } - const nextMeta = { ...(meta as Record) }; + const record = meta as Record; + if ( + !(DAEMON_TRACEPARENT_META_KEY in record) && + !(DAEMON_TRACESTATE_META_KEY in record) + ) { + return request; + } + const nextMeta = { ...record }; delete nextMeta[DAEMON_TRACEPARENT_META_KEY]; delete nextMeta[DAEMON_TRACESTATE_META_KEY]; return { ...request, _meta: nextMeta }; @@ -2208,10 +2215,6 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // Force the body's sessionId to match the routing id — a client that // sent a stale id in the body would otherwise be dispatched to the // wrong agent process. - const normalized: PromptRequest = telemetry.injectPromptContext({ - ...req, - sessionId, - }); const result = entry.promptQueue.then(() => telemetry.runWithContext( capturedContext, @@ -2224,6 +2227,12 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { 'qwen-code.daemon.prompt.queue_wait_ms': Date.now() - queuedAt, }, async () => { + const normalized: PromptRequest = telemetry.injectPromptContext( + { + ...req, + sessionId, + }, + ); // If the caller aborted while we were queued behind earlier // prompts, don't even start this one. if (signal?.aborted) { diff --git a/packages/cli/src/serve/runQwenServe.ts b/packages/cli/src/serve/runQwenServe.ts index f959df284af..55f7edad3ec 100644 --- a/packages/cli/src/serve/runQwenServe.ts +++ b/packages/cli/src/serve/runQwenServe.ts @@ -1087,6 +1087,8 @@ export async function runQwenServe( const finish = (err?: Error | null) => { if (settled) return; settled = true; + process.removeListener('SIGINT', onSignal); + process.removeListener('SIGTERM', onSignal); void shutdownTelemetry() .catch((telemetryErr) => { writeStderrLine( @@ -1098,9 +1100,6 @@ export async function runQwenServe( ); }) .finally(() => { - // Drain finished (or timed out) — safe to detach now. - process.removeListener('SIGINT', onSignal); - process.removeListener('SIGTERM', onSignal); // Server.close error takes precedence (operator-visible // listener problem); fall back to the bridge error // captured during shutdown if any. diff --git a/packages/core/src/telemetry/daemon-tracing.ts b/packages/core/src/telemetry/daemon-tracing.ts index 7795369b72a..efb265562b8 100644 --- a/packages/core/src/telemetry/daemon-tracing.ts +++ b/packages/core/src/telemetry/daemon-tracing.ts @@ -11,7 +11,7 @@ import { SpanKind, SpanStatusCode, trace, - type Attributes, + TraceState, type Context, type Span, } from '@opentelemetry/api'; @@ -40,10 +40,6 @@ export interface DaemonRequestSpanOptions { sessionId?: string; } -function toOtelAttributes(attrs: DaemonAttributes): Attributes { - return attrs; -} - function errorMessage(error: unknown): string { if (error instanceof Error) return error.message; return String(error); @@ -54,11 +50,14 @@ function errorType(error: unknown): string { return typeof error; } +const INVALID_TRACE_ID = '0'.repeat(32); +const INVALID_SPAN_ID = '0'.repeat(16); + function activeSpanContextIsValid(): boolean { const span = trace.getSpan(otelContext.active()); if (!span) return false; const ctx = span.spanContext(); - return ctx.traceId !== '0'.repeat(32) && ctx.spanId !== '0'.repeat(16); + return ctx.traceId !== INVALID_TRACE_ID && ctx.spanId !== INVALID_SPAN_ID; } function stripReservedTraceMeta(meta: unknown): Record { @@ -81,11 +80,14 @@ export async function withDaemonSpan( fn: (span: Span) => Promise, options: { autoOkOnSuccess?: boolean } = {}, ): Promise { + if (!isTelemetrySdkInitialized()) { + return await fn(trace.getSpan(otelContext.active()) as Span); + } const autoOkOnSuccess = options.autoOkOnSuccess ?? true; const tracer = trace.getTracer(SERVICE_NAME); return await tracer.startActiveSpan( name, - { kind: SpanKind.INTERNAL, attributes: toOtelAttributes(attributes) }, + { kind: SpanKind.INTERNAL, attributes }, async (span) => { try { const result = await fn(span); @@ -149,8 +151,6 @@ export function recordDaemonHttpResponse( code: SpanStatusCode.ERROR, message: `HTTP ${statusCode}`, }); - } else { - span?.setStatus({ code: SpanStatusCode.OK }); } } catch { // Telemetry must not affect request handling. @@ -188,7 +188,6 @@ export function emitDaemonLog( body, attributes: { 'event.name': EVENT_DAEMON_ERROR, - 'event.timestamp': new Date().toISOString(), ...attributes, }, }); @@ -218,9 +217,15 @@ export async function runWithDaemonTelemetryContext( export function injectDaemonTraceContext(request: T): T { const currentMeta = (request as { _meta?: unknown })._meta; - const nextMeta = stripReservedTraceMeta(currentMeta); - if (activeSpanContextIsValid()) { + if (!activeSpanContextIsValid()) { + return currentMeta + ? { ...request, _meta: stripReservedTraceMeta(currentMeta) } + : request; + } + + const nextMeta = stripReservedTraceMeta(currentMeta); + try { const carrier: Record = {}; propagation.inject(otelContext.active(), carrier); if (carrier['traceparent']) { @@ -229,6 +234,8 @@ export function injectDaemonTraceContext(request: T): T { if (carrier['tracestate']) { nextMeta[DAEMON_TRACESTATE_META_KEY] = carrier['tracestate']; } + } catch { + // Telemetry must not affect prompt forwarding. } return { @@ -265,7 +272,9 @@ export function extractDaemonTraceContext( parts[0] !== '00' || !traceId?.match(/^[0-9a-f]{32}$/) || !spanId?.match(/^[0-9a-f]{16}$/) || - !flags?.match(/^[0-9a-f]{2}$/) + !flags?.match(/^[0-9a-f]{2}$/) || + traceId === INVALID_TRACE_ID || + spanId === INVALID_SPAN_ID ) { return undefined; } @@ -275,6 +284,10 @@ export function extractDaemonTraceContext( traceId, spanId, traceFlags: Number.parseInt(flags, 16), + isRemote: true, + ...(carrier['tracestate'] + ? { traceState: new TraceState(carrier['tracestate']) } + : {}), }), ); } From 9c37a3dd62309fcb91f711b75cde54abf8895000 Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Wed, 27 May 2026 21:50:31 +0800 Subject: [PATCH 4/7] fix(telemetry): remove TraceState constructor usage in manual fallback TraceState is a type-only export from @opentelemetry/api (not a runtime constructor). The manual fallback path now omits tracestate since the primary propagation.extract path already handles it. --- packages/core/src/telemetry/daemon-tracing.ts | 4 ---- 1 file changed, 4 deletions(-) diff --git a/packages/core/src/telemetry/daemon-tracing.ts b/packages/core/src/telemetry/daemon-tracing.ts index efb265562b8..7f78cf909c4 100644 --- a/packages/core/src/telemetry/daemon-tracing.ts +++ b/packages/core/src/telemetry/daemon-tracing.ts @@ -11,7 +11,6 @@ import { SpanKind, SpanStatusCode, trace, - TraceState, type Context, type Span, } from '@opentelemetry/api'; @@ -285,9 +284,6 @@ export function extractDaemonTraceContext( spanId, traceFlags: Number.parseInt(flags, 16), isRemote: true, - ...(carrier['tracestate'] - ? { traceState: new TraceState(carrier['tracestate']) } - : {}), }), ); } From db06c28357841f4898d2d6090197048b2cc7751e Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Wed, 27 May 2026 21:57:02 +0800 Subject: [PATCH 5/7] fix(telemetry): address wenshao review round 3 - withDaemonSpan: pass undefined (not getSpan result) when SDK off - stripReservedTraceMeta: skip copy when no reserved keys present - sendBridgeErrorImpl: truncate error.message in emitDaemonLog --- packages/cli/src/serve/server.ts | 5 ++++- packages/core/src/telemetry/daemon-tracing.ts | 15 ++++++++++----- 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 6d4a5a0a174..f8aca579d6e 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -3450,7 +3450,10 @@ function sendBridgeErrorImpl( ...(ctx?.route ? { 'http.route': ctx.route } : {}), ...(ctx?.sessionId ? { 'session.id': ctx.sessionId } : {}), 'error.type': err instanceof Error ? err.name : typeof err, - 'error.message': err instanceof Error ? err.message : String(err), + 'error.message': (err instanceof Error ? err.message : String(err)).slice( + 0, + 1024, + ), }); if (daemonLog) { daemonLog.error( diff --git a/packages/core/src/telemetry/daemon-tracing.ts b/packages/core/src/telemetry/daemon-tracing.ts index 7f78cf909c4..46f3fbff648 100644 --- a/packages/core/src/telemetry/daemon-tracing.ts +++ b/packages/core/src/telemetry/daemon-tracing.ts @@ -60,10 +60,15 @@ function activeSpanContextIsValid(): boolean { } function stripReservedTraceMeta(meta: unknown): Record { - const out = - meta && typeof meta === 'object' && !Array.isArray(meta) - ? { ...(meta as Record) } - : {}; + if (!meta || typeof meta !== 'object' || Array.isArray(meta)) return {}; + const record = meta as Record; + if ( + !(DAEMON_TRACEPARENT_META_KEY in record) && + !(DAEMON_TRACESTATE_META_KEY in record) + ) { + return { ...record }; + } + const out = { ...record }; delete out[DAEMON_TRACEPARENT_META_KEY]; delete out[DAEMON_TRACESTATE_META_KEY]; return out; @@ -80,7 +85,7 @@ export async function withDaemonSpan( options: { autoOkOnSuccess?: boolean } = {}, ): Promise { if (!isTelemetrySdkInitialized()) { - return await fn(trace.getSpan(otelContext.active()) as Span); + return await fn(undefined as unknown as Span); } const autoOkOnSuccess = options.autoOkOnSuccess ?? true; const tracer = trace.getTracer(SERVICE_NAME); From 6c35131e36e5ccb982e6021929f37bd2941a7b76 Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Thu, 28 May 2026 02:01:52 +0800 Subject: [PATCH 6/7] fix(telemetry): address wenshao review round 4 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - extractDaemonTraceContext: use ROOT_CONTEXT as extraction base to prevent incorrect parent-child when agent has its own active span - extractDaemonTraceContext (manual fallback): already has isRemote:true and ROOT_CONTEXT from previous fix — confirmed consistent - injectDaemonTraceContext: skip _meta assignment when original had no _meta and no trace headers were injected (match NOOP behavior) - withInteractionSpan: cancelled prompts get UNSET instead of OK so dashboards can distinguish cancelled from successful - emitDaemonLog: use OTel built-in timestamp field instead of custom attribute --- packages/core/src/telemetry/daemon-tracing.ts | 10 ++++++++-- packages/core/src/telemetry/session-tracing.ts | 2 +- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/packages/core/src/telemetry/daemon-tracing.ts b/packages/core/src/telemetry/daemon-tracing.ts index 46f3fbff648..7a78b60db1f 100644 --- a/packages/core/src/telemetry/daemon-tracing.ts +++ b/packages/core/src/telemetry/daemon-tracing.ts @@ -8,6 +8,7 @@ import { createHash } from 'node:crypto'; import { context as otelContext, propagation, + ROOT_CONTEXT, SpanKind, SpanStatusCode, trace, @@ -190,6 +191,7 @@ export function emitDaemonLog( try { logs.getLogger(SERVICE_NAME).emit({ body, + timestamp: new Date(), attributes: { 'event.name': EVENT_DAEMON_ERROR, ...attributes, @@ -242,6 +244,10 @@ export function injectDaemonTraceContext(request: T): T { // Telemetry must not affect prompt forwarding. } + if (!currentMeta && !nextMeta[DAEMON_TRACEPARENT_META_KEY]) { + return request; + } + return { ...request, _meta: nextMeta, @@ -265,7 +271,7 @@ export function extractDaemonTraceContext( if (typeof tracestate === 'string' && tracestate.length > 0) { carrier['tracestate'] = tracestate; } - const extracted = propagation.extract(otelContext.active(), carrier); + const extracted = propagation.extract(ROOT_CONTEXT, carrier); if (trace.getSpanContext(extracted)) return extracted; const parts = traceparent.split('-'); @@ -283,7 +289,7 @@ export function extractDaemonTraceContext( return undefined; } return trace.setSpan( - otelContext.active(), + ROOT_CONTEXT, trace.wrapSpanContext({ traceId, spanId, diff --git a/packages/core/src/telemetry/session-tracing.ts b/packages/core/src/telemetry/session-tracing.ts index d712f84cd79..bc5fc68ea02 100644 --- a/packages/core/src/telemetry/session-tracing.ts +++ b/packages/core/src/telemetry/session-tracing.ts @@ -412,7 +412,7 @@ export async function withInteractionSpan( 'interaction.duration_ms': duration, 'qwen-code.turn_status': terminalStatus, }); - if (terminalStatus !== 'error') { + if (terminalStatus === 'ok') { span.setStatus({ code: SpanStatusCode.OK }); } span.end(); From 9041d56f358a58b72f605a61d218d627c1506403 Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Thu, 28 May 2026 13:26:23 +0800 Subject: [PATCH 7/7] fix(telemetry): address wenshao review round 5 - Import DAEMON_TRACEPARENT/TRACESTATE_META_KEY from core instead of redeclaring locally in bridge.ts (drift risk) - Add isTelemetrySdkInitialized() guard to event() in createDaemonBridgeTelemetry for consistency with siblings - Remove setStatus(ERROR, "HTTP 500") from recordDaemonHttpResponse to avoid overwriting the descriptive error message already set by recordDaemonError --- packages/acp-bridge/src/bridge.ts | 5 ++--- packages/core/src/telemetry/daemon-tracing.test.ts | 4 ++++ packages/core/src/telemetry/daemon-tracing.ts | 7 +------ 3 files changed, 7 insertions(+), 9 deletions(-) diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 7db80af7168..63b6e0c6eb8 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -19,6 +19,8 @@ import type { } from '@agentclientprotocol/sdk'; import type { ApprovalMode } from '@qwen-code/qwen-code-core'; import { + DAEMON_TRACEPARENT_META_KEY, + DAEMON_TRACESTATE_META_KEY, TrustGateError, getCurrentGeminiMdFilename, ShellExecutionService, @@ -88,9 +90,6 @@ import { } from './permissionMediator.js'; import { PermissionForbiddenError } from './bridgeErrors.js'; -const DAEMON_TRACEPARENT_META_KEY = 'qwen.telemetry.traceparent'; -const DAEMON_TRACESTATE_META_KEY = 'qwen.telemetry.tracestate'; - const NOOP_BRIDGE_TELEMETRY: BridgeTelemetry = { captureContext: () => undefined, runWithContext(_captured, fn) { diff --git a/packages/core/src/telemetry/daemon-tracing.test.ts b/packages/core/src/telemetry/daemon-tracing.test.ts index 34a88782827..51630b9015c 100644 --- a/packages/core/src/telemetry/daemon-tracing.test.ts +++ b/packages/core/src/telemetry/daemon-tracing.test.ts @@ -11,6 +11,10 @@ import { type Span, type Tracer, } from '@opentelemetry/api'; + +vi.mock('./sdk.js', () => ({ + isTelemetrySdkInitialized: () => true, +})); import { DAEMON_TRACEPARENT_META_KEY, DAEMON_TRACESTATE_META_KEY, diff --git a/packages/core/src/telemetry/daemon-tracing.ts b/packages/core/src/telemetry/daemon-tracing.ts index 7a78b60db1f..478d7465e13 100644 --- a/packages/core/src/telemetry/daemon-tracing.ts +++ b/packages/core/src/telemetry/daemon-tracing.ts @@ -151,12 +151,6 @@ export function recordDaemonHttpResponse( ): void { try { span?.setAttribute('http.response.status_code', statusCode); - if (statusCode >= 500) { - span?.setStatus({ - code: SpanStatusCode.ERROR, - message: `HTTP ${statusCode}`, - }); - } } catch { // Telemetry must not affect request handling. } @@ -315,6 +309,7 @@ export function createDaemonBridgeTelemetry(): { runWithContext: runWithDaemonTelemetryContext, withSpan: withDaemonBridgeSpan, event(name, attributes) { + if (!isTelemetrySdkInitialized()) return; try { const activeSpan = trace.getSpan(otelContext.active()); if (activeSpan) {