abbyswag/qwen-code
1
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2025-12-19 09:33:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #284 from QwenLM/feat/usage_stats_logging

Browse Source 
feat: add usage statistics logging for Qwen integration
This commit is contained in:
tanzhenxin
2025-08-12 17:56:34 +08:00
committed by GitHub
parent fb9f2d292c 2655af079a
commit 6d5bb1b57c
16 changed files with 673 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/core/src/config/config.ts
View File

@@ -43,7 +43,7 @@ import {
DEFAULT_GEMINI_EMBEDDING_MODEL,
DEFAULT_GEMINI_FLASH_MODEL,
} from './models.js';
import { ClearcutLogger } from '../telemetry/clearcut-logger/clearcut-logger.js';
import { QwenLogger } from '../telemetry/qwen-logger/qwen-logger.js';
import { shouldAttemptBrowserLaunch } from '../utils/browser.js';
import { MCPOAuthConfig } from '../mcp/oauth-provider.js';
import { IdeClient } from '../ide/ide-client.js';
@@ -360,7 +360,7 @@ export class Config {
}
if (this.getUsageStatisticsEnabled()) {
ClearcutLogger.getInstance(this)?.logStartSessionEvent(
QwenLogger.getInstance(this)?.logStartSessionEvent(
new StartSessionEvent(this),
);
} else {

19
packages/core/src/core/__tests__/openaiTimeoutHandling.test.ts
View File

@@ -15,6 +15,7 @@ vi.mock('openai');
// Mock logger modules
vi.mock('../../telemetry/loggers.js', () => ({
logApiResponse: vi.fn(),
logApiError: vi.fn(),
}));
vi.mock('../../utils/openaiLogger.js', () => ({
@@ -290,28 +291,18 @@ describe('OpenAIContentGenerator Timeout Handling', () => {
});
describe('token estimation on timeout', () => {
it('should estimate tokens even when request times out', async () => {
it('should surface a clear timeout error when request times out', async () => {
const timeoutError = new Error('Request timeout');
mockOpenAIClient.chat.completions.create.mockRejectedValue(timeoutError);
// Mock countTokens to return a value
const mockCountTokens = vi.spyOn(generator, 'countTokens');
mockCountTokens.mockResolvedValue({ totalTokens: 100 });
const request = {
contents: [{ role: 'user' as const, parts: [{ text: 'Hello world' }] }],
model: 'gpt-4',
};
try {
await generator.generateContent(request, 'test-prompt-id');
} catch (_error) {
// Verify that countTokens was called for estimation
expect(mockCountTokens).toHaveBeenCalledWith({
contents: request.contents,
model: 'gpt-4',
});
}
await expect(
generator.generateContent(request, 'test-prompt-id'),
).rejects.toThrow(/Request timeout after \d+s/);
});
it('should fall back to character-based estimation if countTokens fails', async () => {

24
packages/core/src/core/geminiChat.ts
View File

@@ -158,14 +158,23 @@ export class GeminiChat {
prompt_id: string,
usageMetadata?: GenerateContentResponseUsageMetadata,
responseText?: string,
responseId?: string,
): Promise<void> {
const authType = this.config.getContentGeneratorConfig()?.authType;
// Don't log API responses for openaiContentGenerator
if (authType === AuthType.QWEN_OAUTH || authType === AuthType.USE_OPENAI) {
return;
}
logApiResponse(
this.config,
new ApiResponseEvent(
responseId || `gemini-${Date.now()}`,
this.config.getModel(),
durationMs,
prompt_id,
this.config.getContentGeneratorConfig()?.authType,
authType,
usageMetadata,
responseText,
),
@@ -176,18 +185,27 @@ export class GeminiChat {
durationMs: number,
error: unknown,
prompt_id: string,
responseId?: string,
): void {
const errorMessage = error instanceof Error ? error.message : String(error);
const errorType = error instanceof Error ? error.name : 'unknown';
const authType = this.config.getContentGeneratorConfig()?.authType;
// Don't log API errors for openaiContentGenerator
if (authType === AuthType.QWEN_OAUTH || authType === AuthType.USE_OPENAI) {
return;
}
logApiError(
this.config,
new ApiErrorEvent(
responseId,
this.config.getModel(),
errorMessage,
durationMs,
prompt_id,
this.config.getContentGeneratorConfig()?.authType,
authType,
errorType,
),
);
@@ -320,6 +338,7 @@ export class GeminiChat {
prompt_id,
response.usageMetadata,
JSON.stringify(response),
response.responseId,
);
this.sendPromise = (async () => {
@@ -563,6 +582,7 @@ export class GeminiChat {
prompt_id,
this.getFinalUsageMetadata(chunks),
JSON.stringify(chunks),
chunks[chunks.length - 1]?.responseId,
);
}
this.recordHistory(inputContent, outputContent);

1
packages/core/src/core/openaiContentGenerator.test.ts
View File

@@ -23,6 +23,7 @@ vi.mock('openai');
// Mock logger modules
vi.mock('../telemetry/loggers.js', () => ({
logApiResponse: vi.fn(),
logApiError: vi.fn(),
}));
vi.mock('../utils/openaiLogger.js', () => ({

124
packages/core/src/core/openaiContentGenerator.ts
View File

@@ -22,8 +22,8 @@ import {
} from '@google/genai';
import { AuthType, ContentGenerator } from './contentGenerator.js';
import OpenAI from 'openai';
import { logApiResponse } from '../telemetry/loggers.js';
import { ApiResponseEvent } from '../telemetry/types.js';
import { logApiError, logApiResponse } from '../telemetry/loggers.js';
import { ApiErrorEvent, ApiResponseEvent } from '../telemetry/types.js';
import { Config } from '../config/config.js';
import { openaiLogger } from '../utils/openaiLogger.js';
@@ -263,6 +263,7 @@ export class OpenAIContentGenerator implements ContentGenerator {
// Log API response event for UI telemetry
const responseEvent = new ApiResponseEvent(
response.responseId || 'unknown',
this.model,
durationMs,
userPromptId,
@@ -291,41 +292,21 @@ export class OpenAIContentGenerator implements ContentGenerator {
? error.message
: String(error);
// Estimate token usage even when there's an error
// This helps track costs and usage even for failed requests
let estimatedUsage;
try {
const tokenCountResult = await this.countTokens({
contents: request.contents,
model: this.model,
});
estimatedUsage = {
promptTokenCount: tokenCountResult.totalTokens,
candidatesTokenCount: 0, // No completion tokens since request failed
totalTokenCount: tokenCountResult.totalTokens,
};
} catch {
// If token counting also fails, provide a minimal estimate
const contentStr = JSON.stringify(request.contents);
const estimatedTokens = Math.ceil(contentStr.length / 4);
estimatedUsage = {
promptTokenCount: estimatedTokens,
candidatesTokenCount: 0,
totalTokenCount: estimatedTokens,
};
}
// Log API error event for UI telemetry with estimated usage
const errorEvent = new ApiResponseEvent(
// Log API error event for UI telemetry
const errorEvent = new ApiErrorEvent(
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(error as any).requestID || 'unknown',
this.model,
errorMessage,
durationMs,
userPromptId,
this.config.getContentGeneratorConfig()?.authType,
estimatedUsage,
undefined,
errorMessage,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(error as any).type,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(error as any).code,
);
logApiResponse(this.config, errorEvent);
logApiError(this.config, errorEvent);
// Log error interaction if enabled
if (this.config.getContentGeneratorConfig()?.enableOpenAILogging) {
@@ -414,6 +395,7 @@ export class OpenAIContentGenerator implements ContentGenerator {
// Log API response event for UI telemetry
const responseEvent = new ApiResponseEvent(
responses[responses.length - 1]?.responseId || 'unknown',
this.model,
durationMs,
userPromptId,
@@ -445,40 +427,21 @@ export class OpenAIContentGenerator implements ContentGenerator {
? error.message
: String(error);
// Estimate token usage even when there's an error in streaming
let estimatedUsage;
try {
const tokenCountResult = await this.countTokens({
contents: request.contents,
model: this.model,
});
estimatedUsage = {
promptTokenCount: tokenCountResult.totalTokens,
candidatesTokenCount: 0, // No completion tokens since request failed
totalTokenCount: tokenCountResult.totalTokens,
};
} catch {
// If token counting also fails, provide a minimal estimate
const contentStr = JSON.stringify(request.contents);
const estimatedTokens = Math.ceil(contentStr.length / 4);
estimatedUsage = {
promptTokenCount: estimatedTokens,
candidatesTokenCount: 0,
totalTokenCount: estimatedTokens,
};
}
// Log API error event for UI telemetry with estimated usage
const errorEvent = new ApiResponseEvent(
// Log API error event for UI telemetry
const errorEvent = new ApiErrorEvent(
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(error as any).requestID || 'unknown',
this.model,
errorMessage,
durationMs,
userPromptId,
this.config.getContentGeneratorConfig()?.authType,
estimatedUsage,
undefined,
errorMessage,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(error as any).type,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(error as any).code,
);
logApiResponse(this.config, errorEvent);
logApiError(this.config, errorEvent);
// Log error interaction if enabled
if (this.config.getContentGeneratorConfig()?.enableOpenAILogging) {
@@ -518,40 +481,21 @@ export class OpenAIContentGenerator implements ContentGenerator {
? error.message
: String(error);
// Estimate token usage even when there's an error in streaming setup
let estimatedUsage;
try {
const tokenCountResult = await this.countTokens({
contents: request.contents,
model: this.model,
});
estimatedUsage = {
promptTokenCount: tokenCountResult.totalTokens,
candidatesTokenCount: 0, // No completion tokens since request failed
totalTokenCount: tokenCountResult.totalTokens,
};
} catch {
// If token counting also fails, provide a minimal estimate
const contentStr = JSON.stringify(request.contents);
const estimatedTokens = Math.ceil(contentStr.length / 4);
estimatedUsage = {
promptTokenCount: estimatedTokens,
candidatesTokenCount: 0,
totalTokenCount: estimatedTokens,
};
}
// Log API error event for UI telemetry with estimated usage
const errorEvent = new ApiResponseEvent(
// Log API error event for UI telemetry
const errorEvent = new ApiErrorEvent(
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(error as any).requestID || 'unknown',
this.model,
errorMessage,
durationMs,
userPromptId,
this.config.getContentGeneratorConfig()?.authType,
estimatedUsage,
undefined,
errorMessage,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(error as any).type,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(error as any).code,
);
logApiResponse(this.config, errorEvent);
logApiError(this.config, errorEvent);
// Allow subclasses to suppress error logging for specific scenarios
if (!this.shouldSuppressErrorLogging(error, request)) {

2
packages/core/src/services/loopDetectionService.test.ts
View File

@@ -19,6 +19,8 @@ import { LoopDetectionService } from './loopDetectionService.js';
vi.mock('../telemetry/loggers.js', () => ({
logLoopDetected: vi.fn(),
logApiError: vi.fn(),
logApiResponse: vi.fn(),
}));
const TOOL_CALL_LOOP_THRESHOLD = 5;

2
packages/core/src/telemetry/clearcut-logger/event-metadata-key.ts
View File

@@ -4,7 +4,7 @@
* SPDX-License-Identifier: Apache-2.0
*/
// Defines valid event metadata keys for Clearcut logging.
// Defines valid event metadata keys for Qwen logging.
export enum EventMetadataKey {
GEMINI_CLI_KEY_UNKNOWN = 0,

17
packages/core/src/telemetry/integration.test.circular.ts
View File

@@ -9,11 +9,12 @@
*/
import { describe, it, expect } from 'vitest';
import { ClearcutLogger } from './clearcut-logger/clearcut-logger.js';
import { QwenLogger } from './qwen-logger/qwen-logger.js';
import { RumEvent } from './qwen-logger/event-types.js';
import { Config } from '../config/config.js';
describe('Circular Reference Integration Test', () => {
it('should handle HttpsProxyAgent-like circular references in clearcut logging', () => {
it('should handle HttpsProxyAgent-like circular references in qwen logging', () => {
// Create a mock config with proxy
const mockConfig = {
getTelemetryEnabled: () => true,
@@ -44,16 +45,20 @@ describe('Circular Reference Integration Test', () => {
proxyAgentLike.sockets['cloudcode-pa.googleapis.com:443'] = [socketLike];
// Create an event that would contain this circular structure
const problematicEvent = {
const problematicEvent: RumEvent = {
timestamp: Date.now(),
event_type: 'exception',
type: 'error',
name: 'api_error',
error: new Error('Network error'),
function_args: {
filePath: '/test/file.txt',
httpAgent: proxyAgentLike, // This would cause the circular reference
},
};
} as RumEvent;
// Test that ClearcutLogger can handle this
const logger = ClearcutLogger.getInstance(mockConfig);
// Test that QwenLogger can handle this
const logger = QwenLogger.getInstance(mockConfig);
expect(() => {
logger?.enqueueLogEvent(problematicEvent);

3
packages/core/src/telemetry/loggers.test.ts
View File

@@ -212,6 +212,7 @@ describe('loggers', () => {
toolUsePromptTokenCount: 2,
};
const event = new ApiResponseEvent(
'test-response-id',
'test-model',
100,
'prompt-id-1',
@@ -229,6 +230,7 @@ describe('loggers', () => {
'event.name': EVENT_API_RESPONSE,
'event.timestamp': '2025-01-01T00:00:00.000Z',
[SemanticAttributes.HTTP_STATUS_CODE]: 200,
response_id: 'test-response-id',
model: 'test-model',
status_code: 200,
duration_ms: 100,
@@ -275,6 +277,7 @@ describe('loggers', () => {
toolUsePromptTokenCount: 2,
};
const event = new ApiResponseEvent(
'test-response-id-2',
'test-model',
100,
'prompt-id-1',

22
packages/core/src/telemetry/loggers.ts
View File

@@ -39,7 +39,7 @@ import {
} from './metrics.js';
import { isTelemetrySdkInitialized } from './sdk.js';
import { uiTelemetryService, UiEvent } from './uiTelemetry.js';
import { ClearcutLogger } from './clearcut-logger/clearcut-logger.js';
import { QwenLogger } from './qwen-logger/qwen-logger.js';
import { safeJsonStringify } from '../utils/safeJsonStringify.js';
const shouldLogUserPrompts = (config: Config): boolean =>
@@ -55,7 +55,7 @@ export function logCliConfiguration(
config: Config,
event: StartSessionEvent,
): void {
ClearcutLogger.getInstance(config)?.logStartSessionEvent(event);
QwenLogger.getInstance(config)?.logStartSessionEvent(event);
if (!isTelemetrySdkInitialized()) return;
const attributes: LogAttributes = {
@@ -84,7 +84,7 @@ export function logCliConfiguration(
}
export function logUserPrompt(config: Config, event: UserPromptEvent): void {
ClearcutLogger.getInstance(config)?.logNewPromptEvent(event);
QwenLogger.getInstance(config)?.logNewPromptEvent(event);
if (!isTelemetrySdkInitialized()) return;
const attributes: LogAttributes = {
@@ -113,7 +113,7 @@ export function logToolCall(config: Config, event: ToolCallEvent): void {
'event.timestamp': new Date().toISOString(),
} as UiEvent;
uiTelemetryService.addEvent(uiEvent);
ClearcutLogger.getInstance(config)?.logToolCallEvent(event);
QwenLogger.getInstance(config)?.logToolCallEvent(event);
if (!isTelemetrySdkInitialized()) return;
const attributes: LogAttributes = {
@@ -146,7 +146,7 @@ export function logToolCall(config: Config, event: ToolCallEvent): void {
}
export function logApiRequest(config: Config, event: ApiRequestEvent): void {
ClearcutLogger.getInstance(config)?.logApiRequestEvent(event);
QwenLogger.getInstance(config)?.logApiRequestEvent(event);
if (!isTelemetrySdkInitialized()) return;
const attributes: LogAttributes = {
@@ -168,7 +168,7 @@ export function logFlashFallback(
config: Config,
event: FlashFallbackEvent,
): void {
ClearcutLogger.getInstance(config)?.logFlashFallbackEvent(event);
QwenLogger.getInstance(config)?.logFlashFallbackEvent(event);
if (!isTelemetrySdkInitialized()) return;
const attributes: LogAttributes = {
@@ -193,7 +193,7 @@ export function logApiError(config: Config, event: ApiErrorEvent): void {
'event.timestamp': new Date().toISOString(),
} as UiEvent;
uiTelemetryService.addEvent(uiEvent);
ClearcutLogger.getInstance(config)?.logApiErrorEvent(event);
QwenLogger.getInstance(config)?.logApiErrorEvent(event);
if (!isTelemetrySdkInitialized()) return;
const attributes: LogAttributes = {
@@ -235,7 +235,7 @@ export function logApiResponse(config: Config, event: ApiResponseEvent): void {
'event.timestamp': new Date().toISOString(),
} as UiEvent;
uiTelemetryService.addEvent(uiEvent);
ClearcutLogger.getInstance(config)?.logApiResponseEvent(event);
QwenLogger.getInstance(config)?.logApiResponseEvent(event);
if (!isTelemetrySdkInitialized()) return;
const attributes: LogAttributes = {
...getCommonAttributes(config),
@@ -298,7 +298,7 @@ export function logLoopDetected(
config: Config,
event: LoopDetectedEvent,
): void {
ClearcutLogger.getInstance(config)?.logLoopDetectedEvent(event);
QwenLogger.getInstance(config)?.logLoopDetectedEvent(event);
if (!isTelemetrySdkInitialized()) return;
const attributes: LogAttributes = {
@@ -318,7 +318,7 @@ export function logNextSpeakerCheck(
config: Config,
event: NextSpeakerCheckEvent,
): void {
ClearcutLogger.getInstance(config)?.logNextSpeakerCheck(event);
QwenLogger.getInstance(config)?.logNextSpeakerCheck(event);
if (!isTelemetrySdkInitialized()) return;
const attributes: LogAttributes = {
@@ -339,7 +339,7 @@ export function logSlashCommand(
config: Config,
event: SlashCommandEvent,
): void {
ClearcutLogger.getInstance(config)?.logSlashCommandEvent(event);
QwenLogger.getInstance(config)?.logSlashCommandEvent(event);
if (!isTelemetrySdkInitialized()) return;
const attributes: LogAttributes = {

83
packages/core/src/telemetry/qwen-logger/event-types.ts Normal file
View File

@@ -0,0 +1,83 @@
// RUM Protocol Data Structures
export interface RumApp {
id: string;
env: string;
version: string;
type: 'cli' | 'extension';
}
export interface RumUser {
id: string;
}
export interface RumSession {
id: string;
}
export interface RumView {
id: string;
name: string;
}
export interface RumEvent {
timestamp?: number;
event_type?: 'view' | 'action' | 'exception' | 'resource';
type: string; // Event type
name: string; // Event name
snapshots?: string; // JSON string of event snapshots
properties?: Record<string, unknown>;
// [key: string]: unknown;
}
export interface RumViewEvent extends RumEvent {
view_type?: string; // View rendering type
time_spent?: number; // Time spent on current view in ms
}
export interface RumActionEvent extends RumEvent {
target_name?: string; // Element user interacted with (for auto-collected actions only)
duration?: number; // Action duration in ms
method_info?: string; // Action callback, e.g.: onClick()
}
export interface RumExceptionEvent extends RumEvent {
source?: string; // Error source, e.g.: console, event
file?: string; // Error file
subtype?: string; // Secondary classification of error type
message?: string; // Concise, readable message explaining the event
stack?: string; // Stack trace or supplemental information about the error
caused_by?: string; // Exception cause
line?: number; // Line number where exception occurred
column?: number; // Column number where exception occurred
thread_id?: string; // Thread ID
binary_images?: string; // Error source
}
export interface RumResourceEvent extends RumEvent {
method?: string; // HTTP request method: POST, GET, etc.
status_code?: string; // Resource status code
message?: string; // Error message content, corresponds to resource.error_msg
url?: string; // Resource URL
provider_type?: string; // Resource provider type: first-party, cdn, ad, analytics
trace_id?: string; // Resource request TraceID
success?: number; // Resource loading success: 1 (default) success, 0 failure
duration?: number; // Total time spent loading resource in ms (responseEnd - redirectStart)
size?: number; // Resource size in bytes, corresponds to decodedBodySize
connect_duration?: number; // Time spent establishing connection to server in ms (connectEnd - connectStart)
ssl_duration?: number; // Time spent on TLS handshake in ms (connectEnd - secureConnectionStart), 0 if no SSL
dns_duration?: number; // Time spent resolving DNS name in ms (domainLookupEnd - domainLookupStart)
redirect_duration?: number; // Time spent on HTTP redirects in ms (redirectEnd - redirectStart)
first_byte_duration?: number; // Time waiting for first byte of response in ms (responseStart - requestStart)
download_duration?: number; // Time spent downloading response in ms (responseEnd - responseStart)
timing_data?: string; // JSON string of PerformanceResourceTiming
trace_data?: string; // Trace information snapshot JSON string
}
export interface RumPayload {
app: RumApp;
user: RumUser;
session: RumSession;
view: RumView;
events: RumEvent[];
_v: string;
}

484
packages/core/src/telemetry/qwen-logger/qwen-logger.ts Normal file
View File

@@ -0,0 +1,484 @@
/**
* @license
* Copyright 2025 Google LLC
* SPDX-License-Identifier: Apache-2.0
*/
import { Buffer } from 'buffer';
import * as https from 'https';
import { HttpsProxyAgent } from 'https-proxy-agent';
import { randomUUID } from 'crypto';
import {
StartSessionEvent,
EndSessionEvent,
UserPromptEvent,
ToolCallEvent,
ApiRequestEvent,
ApiResponseEvent,
ApiErrorEvent,
FlashFallbackEvent,
LoopDetectedEvent,
NextSpeakerCheckEvent,
SlashCommandEvent,
MalformedJsonResponseEvent,
} from '../types.js';
import {
RumEvent,
RumViewEvent,
RumActionEvent,
RumResourceEvent,
RumExceptionEvent,
RumPayload,
} from './event-types.js';
// Removed unused EventMetadataKey import
import { Config } from '../../config/config.js';
import { safeJsonStringify } from '../../utils/safeJsonStringify.js';
// Removed unused import
import { HttpError, retryWithBackoff } from '../../utils/retry.js';
import { getInstallationId } from '../../utils/user_id.js';
// Usage statistics collection endpoint
const USAGE_STATS_HOSTNAME = 'gb4w8c3ygj-default-sea.rum.aliyuncs.com';
const USAGE_STATS_PATH = '/';
const RUN_APP_ID = 'gb4w8c3ygj@851d5d500f08f92';
export interface LogResponse {
nextRequestWaitMs?: number;
}
// Singleton class for batch posting log events to RUM. When a new event comes in, the elapsed time
// is checked and events are flushed to RUM if at least a minute has passed since the last flush.
export class QwenLogger {
private static instance: QwenLogger;
private config?: Config;
private readonly events: RumEvent[] = [];
private last_flush_time: number = Date.now();
private flush_interval_ms: number = 1000 * 60; // Wait at least a minute before flushing events.
private userId: string;
private sessionId: string;
private viewId: string;
private isFlushInProgress: boolean = false;
private isShutdown: boolean = false;
private constructor(config?: Config) {
this.config = config;
this.userId = this.generateUserId();
this.sessionId =
typeof this.config?.getSessionId === 'function'
? this.config.getSessionId()
: '';
this.viewId = randomUUID();
}
private generateUserId(): string {
// Use installation ID as user ID for consistency
return `user-${getInstallationId()}`;
}
static getInstance(config?: Config): QwenLogger | undefined {
if (config === undefined || !config?.getUsageStatisticsEnabled())
return undefined;
if (!QwenLogger.instance) {
QwenLogger.instance = new QwenLogger(config);
}
process.on('exit', QwenLogger.instance.shutdown.bind(QwenLogger.instance));
return QwenLogger.instance;
}
enqueueLogEvent(event: RumEvent): void {
this.events.push(event);
}
createRumEvent(
eventType: 'view' | 'action' | 'exception' | 'resource',
type: string,
name: string,
properties: Partial<RumEvent>,
): RumEvent {
return {
timestamp: Date.now(),
event_type: eventType,
type,
name,
...(properties || {}),
};
}
createViewEvent(
type: string,
name: string,
properties: Partial<RumViewEvent>,
): RumEvent {
return this.createRumEvent('view', type, name, properties);
}
createActionEvent(
type: string,
name: string,
properties: Partial<RumActionEvent>,
): RumEvent {
return this.createRumEvent('action', type, name, properties);
}
createResourceEvent(
type: string,
name: string,
properties: Partial<RumResourceEvent>,
): RumEvent {
return this.createRumEvent('resource', type, name, properties);
}
createExceptionEvent(
type: string,
name: string,
properties: Partial<RumExceptionEvent>,
): RumEvent {
return this.createRumEvent('exception', type, name, properties);
}
createRumPayload(): RumPayload {
const version = process.env.CLI_VERSION || process.version;
return {
app: {
id: RUN_APP_ID,
env: process.env.DEBUG ? 'dev' : 'prod',
version,
type: 'cli',
},
user: {
id: this.userId,
},
session: {
id: this.sessionId,
},
view: {
id: this.viewId,
name: 'qwen-code-cli',
},
events: [...this.events],
_v: `qwen-code@${version}`,
};
}
flushIfNeeded(): void {
if (Date.now() - this.last_flush_time < this.flush_interval_ms) {
return;
}
// Prevent concurrent flush operations
if (this.isFlushInProgress) {
return;
}
this.flushToRum().catch((error) => {
console.debug('Error flushing to RUM:', error);
});
}
async flushToRum(): Promise<LogResponse> {
if (this.config?.getDebugMode()) {
console.log('Flushing log events to RUM.');
}
if (this.events.length === 0) {
return {};
}
this.isFlushInProgress = true;
const rumPayload = this.createRumPayload();
const flushFn = () =>
new Promise<Buffer>((resolve, reject) => {
const body = safeJsonStringify(rumPayload);
const options = {
hostname: USAGE_STATS_HOSTNAME,
path: USAGE_STATS_PATH,
method: 'POST',
headers: {
'Content-Length': Buffer.byteLength(body),
'Content-Type': 'text/plain;charset=UTF-8',
},
};
const bufs: Buffer[] = [];
const req = https.request(
{
...options,
agent: this.getProxyAgent(),
},
(res) => {
if (
res.statusCode &&
(res.statusCode < 200 || res.statusCode >= 300)
) {
const err: HttpError = new Error(
`Request failed with status ${res.statusCode}`,
);
err.status = res.statusCode;
res.resume();
return reject(err);
}
res.on('data', (buf) => bufs.push(buf));
res.on('end', () => resolve(Buffer.concat(bufs)));
},
);
req.on('error', reject);
req.end(body);
});
try {
await retryWithBackoff(flushFn, {
maxAttempts: 3,
initialDelayMs: 200,
shouldRetry: (err: unknown) => {
if (!(err instanceof Error)) return false;
const status = (err as HttpError).status as number | undefined;
// If status is not available, it's likely a network error
if (status === undefined) return true;
// Retry on 429 (Too many Requests) and 5xx server errors.
return status === 429 || (status >= 500 && status < 600);
},
});
this.events.splice(0, this.events.length);
this.last_flush_time = Date.now();
return {};
} catch (error) {
if (this.config?.getDebugMode()) {
console.error('RUM flush failed after multiple retries.', error);
}
return {};
} finally {
this.isFlushInProgress = false;
}
}
logStartSessionEvent(event: StartSessionEvent): void {
const applicationEvent = this.createViewEvent('session', 'session_start', {
properties: {
model: event.model,
},
snapshots: JSON.stringify({
embedding_model: event.embedding_model,
sandbox_enabled: event.sandbox_enabled,
core_tools_enabled: event.core_tools_enabled,
approval_mode: event.approval_mode,
api_key_enabled: event.api_key_enabled,
vertex_ai_enabled: event.vertex_ai_enabled,
debug_enabled: event.debug_enabled,
mcp_servers: event.mcp_servers,
telemetry_enabled: event.telemetry_enabled,
telemetry_log_user_prompts_enabled:
event.telemetry_log_user_prompts_enabled,
}),
});
// Flush start event immediately
this.enqueueLogEvent(applicationEvent);
this.flushToRum().catch((error: unknown) => {
console.debug('Error flushing to RUM:', error);
});
}
logNewPromptEvent(event: UserPromptEvent): void {
const rumEvent = this.createActionEvent('user_prompt', 'user_prompt', {
properties: {
auth_type: event.auth_type,
prompt_id: event.prompt_id,
},
snapshots: JSON.stringify({
prompt_length: event.prompt_length,
}),
});
this.enqueueLogEvent(rumEvent);
this.flushIfNeeded();
}
logToolCallEvent(event: ToolCallEvent): void {
const rumEvent = this.createActionEvent(
'tool_call',
`tool_call#${event.function_name}`,
{
properties: {
prompt_id: event.prompt_id,
},
snapshots: JSON.stringify({
function_name: event.function_name,
decision: event.decision,
success: event.success,
duration_ms: event.duration_ms,
error: event.error,
error_type: event.error_type,
}),
},
);
this.enqueueLogEvent(rumEvent);
this.flushIfNeeded();
}
logApiRequestEvent(_event: ApiRequestEvent): void {
// ignore for now
return;
// const rumEvent = this.createResourceEvent('api', 'api_request', {
// properties: {
// model: event.model,
// prompt_id: event.prompt_id,
// },
// });
// this.enqueueLogEvent(rumEvent);
// this.flushIfNeeded();
}
logApiResponseEvent(event: ApiResponseEvent): void {
const rumEvent = this.createResourceEvent('api', 'api_response', {
status_code: event.status_code?.toString() ?? '',
duration: event.duration_ms,
success: 1,
message: event.error,
trace_id: event.response_id,
properties: {
auth_type: event.auth_type,
model: event.model,
prompt_id: event.prompt_id,
},
snapshots: JSON.stringify({
input_token_count: event.input_token_count,
output_token_count: event.output_token_count,
cached_content_token_count: event.cached_content_token_count,
thoughts_token_count: event.thoughts_token_count,
tool_token_count: event.tool_token_count,
}),
});
this.enqueueLogEvent(rumEvent);
this.flushIfNeeded();
}
logApiErrorEvent(event: ApiErrorEvent): void {
const rumEvent = this.createResourceEvent('api', 'api_error', {
status_code: event.status_code?.toString() ?? '',
duration: event.duration_ms,
success: 0,
message: event.error,
trace_id: event.response_id,
properties: {
auth_type: event.auth_type,
model: event.model,
prompt_id: event.prompt_id,
},
snapshots: JSON.stringify({
error_type: event.error_type,
}),
});
this.enqueueLogEvent(rumEvent);
this.flushIfNeeded();
}
logFlashFallbackEvent(event: FlashFallbackEvent): void {
const rumEvent = this.createActionEvent('fallback', 'flash_fallback', {
properties: {
auth_type: event.auth_type,
},
});
this.enqueueLogEvent(rumEvent);
this.flushIfNeeded();
}
logLoopDetectedEvent(event: LoopDetectedEvent): void {
const rumEvent = this.createExceptionEvent('error', 'loop_detected', {
subtype: 'loop_detected',
properties: {
prompt_id: event.prompt_id,
},
snapshots: JSON.stringify({
loop_type: event.loop_type,
}),
});
this.enqueueLogEvent(rumEvent);
this.flushIfNeeded();
}
logNextSpeakerCheck(event: NextSpeakerCheckEvent): void {
const rumEvent = this.createActionEvent('check', 'next_speaker_check', {
properties: {
prompt_id: event.prompt_id,
},
snapshots: JSON.stringify({
finish_reason: event.finish_reason,
result: event.result,
}),
});
this.enqueueLogEvent(rumEvent);
this.flushIfNeeded();
}
logSlashCommandEvent(event: SlashCommandEvent): void {
const rumEvent = this.createActionEvent('command', 'slash_command', {
snapshots: JSON.stringify({
command: event.command,
subcommand: event.subcommand,
}),
});
this.enqueueLogEvent(rumEvent);
this.flushIfNeeded();
}
logMalformedJsonResponseEvent(event: MalformedJsonResponseEvent): void {
const rumEvent = this.createExceptionEvent(
'error',
'malformed_json_response',
{
subtype: 'malformed_json_response',
properties: {
model: event.model,
},
},
);
this.enqueueLogEvent(rumEvent);
this.flushIfNeeded();
}
logEndSessionEvent(_event: EndSessionEvent): void {
const applicationEvent = this.createViewEvent('session', 'session_end', {});
// Flush immediately on session end.
this.enqueueLogEvent(applicationEvent);
this.flushToRum().catch((error: unknown) => {
console.debug('Error flushing to RUM:', error);
});
}
getProxyAgent() {
const proxyUrl = this.config?.getProxy();
if (!proxyUrl) return undefined;
// undici which is widely used in the repo can only support http & https proxy protocol,
// https://github.com/nodejs/undici/issues/2224
if (proxyUrl.startsWith('http')) {
return new HttpsProxyAgent(proxyUrl);
} else {
throw new Error('Unsupported proxy type');
}
}
shutdown() {
if (this.isShutdown) return;
this.isShutdown = true;
const event = new EndSessionEvent(this.config);
this.logEndSessionEvent(event);
}
}

2
packages/core/src/telemetry/sdk.ts
View File

@@ -28,7 +28,6 @@ import { HttpInstrumentation } from '@opentelemetry/instrumentation-http';
import { Config } from '../config/config.js';
import { SERVICE_NAME } from './constants.js';
import { initializeMetrics } from './metrics.js';
import { ClearcutLogger } from './clearcut-logger/clearcut-logger.js';
import {
FileLogExporter,
FileMetricExporter,
@@ -141,7 +140,6 @@ export async function shutdownTelemetry(): Promise<void> {
return;
}
try {
ClearcutLogger.getInstance()?.shutdown();
await sdk.shutdown();
console.log('OpenTelemetry SDK shut down successfully.');
} catch (error) {

6
packages/core/src/telemetry/types.ts
View File

@@ -161,6 +161,7 @@ export class ApiRequestEvent {
export class ApiErrorEvent {
'event.name': 'api_error';
'event.timestamp': string; // ISO 8601
response_id?: string;
model: string;
error: string;
error_type?: string;
@@ -170,6 +171,7 @@ export class ApiErrorEvent {
auth_type?: string;
constructor(
response_id: string | undefined,
model: string,
error: string,
duration_ms: number,
@@ -180,6 +182,7 @@ export class ApiErrorEvent {
) {
this['event.name'] = 'api_error';
this['event.timestamp'] = new Date().toISOString();
this.response_id = response_id;
this.model = model;
this.error = error;
this.error_type = error_type;
@@ -193,6 +196,7 @@ export class ApiErrorEvent {
export class ApiResponseEvent {
'event.name': 'api_response';
'event.timestamp': string; // ISO 8601
response_id: string;
model: string;
status_code?: number | string;
duration_ms: number;
@@ -208,6 +212,7 @@ export class ApiResponseEvent {
auth_type?: string;
constructor(
response_id: string,
model: string,
duration_ms: number,
prompt_id: string,
@@ -218,6 +223,7 @@ export class ApiResponseEvent {
) {
this['event.name'] = 'api_response';
this['event.timestamp'] = new Date().toISOString();
this.response_id = response_id;
this.model = model;
this.duration_ms = duration_ms;
this.status_code = 200;

6
packages/core/test-setup.ts
View File

@@ -8,3 +8,9 @@ import { setSimulate429 } from './src/utils/testUtils.js';
// Disable 429 simulation globally for all tests
setSimulate429(false);
// Some dependencies (e.g., undici) expect a global File constructor in Node.
// Provide a minimal shim for test environment if missing.
if (typeof (globalThis as unknown as { File?: unknown }).File === 'undefined') {
(globalThis as unknown as { File: unknown }).File = class {} as unknown;
}