diff --git a/packages/core/src/telemetry/log-to-span-processor.test.ts b/packages/core/src/telemetry/log-to-span-processor.test.ts index 26d9f098ab4..73a98dca6a6 100644 --- a/packages/core/src/telemetry/log-to-span-processor.test.ts +++ b/packages/core/src/telemetry/log-to-span-processor.test.ts @@ -751,4 +751,388 @@ describe('LogToSpanProcessor', () => { deriveTraceId('fresh-session'), ); }); + + describe('export failure diagnostics', () => { + function makeFailingProcessor(error: Error | undefined) { + const failingExporter = { + export: vi.fn((_spans, cb) => cb({ code: 1, error })), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter; + return new LogToSpanProcessor(failingExporter, 60000); + } + + async function flushOne(p: LogToSpanProcessor) { + p.onEmit({ + body: 'event', + hrTime: [1000, 0] as [number, number], + attributes: { 'event.name': 'event' }, + } as unknown as ReadableLogRecord); + await p.forceFlush(); + } + + it('falls back to error.name when message is empty (HTTP/2 / stripped reason phrase)', async () => { + await processor.shutdown(); + const err = Object.assign(new Error(''), { + name: 'OTLPExporterError', + code: 403, + data: 'Forbidden: invalid license', + }); + processor = makeFailingProcessor(err); + const stderrWrite = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + + try { + await flushOne(processor); + expect(stderrWrite).toHaveBeenCalledWith( + '[LogToSpan] export failed: code=1 error="OTLPExporterError" httpCode=403 data="Forbidden: invalid license"\n', + ); + } finally { + stderrWrite.mockRestore(); + } + }); + + it('JSON-escapes embedded newlines in message and data so the record stays on one line', async () => { + await processor.shutdown(); + const err = Object.assign(new Error('line1\nline2'), { + name: 'OTLPExporterError', + code: 500, + data: '{\n "error": "boom"\n}', + }); + const sink = vi.fn(); + processor = new LogToSpanProcessor( + { + export: vi.fn((_s, cb) => cb({ code: 1, error: err })), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter, + { flushIntervalMs: 60000, diagnosticsSink: sink }, + ); + + await flushOne(processor); + const msg = sink.mock.calls[0][0] as string; + expect(msg).not.toContain('\n'); + expect(msg).toContain('error="line1\\nline2"'); + expect(msg).toContain('data="{\\n \\"error\\": \\"boom\\"\\n}"'); + }); + + it('truncates response data snippets to 200 characters before stringifying', async () => { + await processor.shutdown(); + const err = Object.assign(new Error(''), { + name: 'OTLPExporterError', + code: 500, + data: 'x'.repeat(500), + }); + processor = makeFailingProcessor(err); + const stderrWrite = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + + try { + await flushOne(processor); + const msg = stderrWrite.mock.calls[0][0] as string; + expect(msg).toContain('httpCode=500'); + expect(msg).toContain(`data="${'x'.repeat(200)}"`); + expect(msg).not.toContain('x'.repeat(201)); + } finally { + stderrWrite.mockRestore(); + } + }); + + it('omits httpCode when err.code is a non-numeric networking code (ECONNREFUSED)', async () => { + await processor.shutdown(); + const err = Object.assign(new Error('connect ECONNREFUSED 127.0.0.1'), { + code: 'ECONNREFUSED', + }); + processor = makeFailingProcessor(err); + const stderrWrite = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + + try { + await flushOne(processor); + const msg = stderrWrite.mock.calls[0][0] as string; + expect(msg).not.toContain('httpCode='); + expect(msg).toContain('error="connect ECONNREFUSED 127.0.0.1"'); + } finally { + stderrWrite.mockRestore(); + } + }); + + it('reports error="unknown" when result.error is missing', async () => { + await processor.shutdown(); + processor = makeFailingProcessor(undefined); + const stderrWrite = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + + try { + await flushOne(processor); + expect(stderrWrite).toHaveBeenCalledWith( + '[LogToSpan] export failed: code=1 error="unknown"\n', + ); + } finally { + stderrWrite.mockRestore(); + } + }); + + it('omits data field when err.data is a non-string truthy value (e.g. Buffer)', async () => { + await processor.shutdown(); + const err = Object.assign(new Error('fail'), { + code: 500, + data: Buffer.from('binary'), + }); + processor = makeFailingProcessor(err as unknown as Error); + const stderrWrite = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + + try { + await flushOne(processor); + const msg = stderrWrite.mock.calls[0][0] as string; + expect(msg).toContain('httpCode=500'); + expect(msg).not.toContain('data='); + } finally { + stderrWrite.mockRestore(); + } + }); + + it('falls back to "unknown" when both message and name are empty (e.g. minified Error)', async () => { + await processor.shutdown(); + const err = Object.assign(new Error(''), { name: '' }); + processor = makeFailingProcessor(err); + const stderrWrite = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + + try { + await flushOne(processor); + expect(stderrWrite).toHaveBeenCalledWith( + '[LogToSpan] export failed: code=1 error="unknown"\n', + ); + } finally { + stderrWrite.mockRestore(); + } + }); + + it('omits data field when err.data is an empty string (guards against length>0 loosening)', async () => { + await processor.shutdown(); + const err = Object.assign(new Error('fail'), { code: 500, data: '' }); + processor = makeFailingProcessor(err); + const stderrWrite = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + + try { + await flushOne(processor); + const msg = stderrWrite.mock.calls[0][0] as string; + expect(msg).toContain('httpCode=500'); + expect(msg).not.toContain('data='); + } finally { + stderrWrite.mockRestore(); + } + }); + + it('routes diagnostics to an injected sink without touching stderr', async () => { + await processor.shutdown(); + const sink = vi.fn(); + const stderrWrite = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const failingExporter = { + export: vi.fn((_spans, cb) => + cb({ code: 1, error: new Error('boom') }), + ), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter; + processor = new LogToSpanProcessor(failingExporter, { + flushIntervalMs: 60000, + diagnosticsSink: sink, + }); + + try { + await flushOne(processor); + expect(sink).toHaveBeenCalledWith( + '[LogToSpan] export failed: code=1 error="boom"', + ); + expect(stderrWrite).not.toHaveBeenCalled(); + } finally { + stderrWrite.mockRestore(); + } + }); + + it('routes buffer-overflow warnings through the injected sink', async () => { + await processor.shutdown(); + const sink = vi.fn(); + processor = new LogToSpanProcessor( + { + export: vi.fn((_s, cb) => cb({ code: 0 })), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter, + { flushIntervalMs: 60000, maxBufferSize: 2, diagnosticsSink: sink }, + ); + + for (const body of ['a', 'b', 'c']) { + processor.onEmit({ + body, + hrTime: [1000, 0] as [number, number], + attributes: { 'event.name': body }, + } as unknown as ReadableLogRecord); + } + + expect(sink).toHaveBeenCalledWith( + expect.stringContaining('[LogToSpan] buffer exceeded max size'), + ); + }); + + it('routes export timeout through the injected sink', async () => { + await processor.shutdown(); + vi.useFakeTimers(); + const sink = vi.fn(); + try { + processor = new LogToSpanProcessor( + { + // Never invoke the callback — force the timeout branch. + export: vi.fn(), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter, + { flushIntervalMs: 60000, diagnosticsSink: sink }, + ); + processor.onEmit({ + body: 'event', + hrTime: [1000, 0] as [number, number], + attributes: { 'event.name': 'event' }, + } as unknown as ReadableLogRecord); + + const flushPromise = processor.forceFlush(); + // EXPORT_TIMEOUT_MS is 30_000 — advance past it. + await vi.advanceTimersByTimeAsync(31_000); + await flushPromise; + + expect(sink).toHaveBeenCalledWith( + expect.stringMatching( + /^\[LogToSpan] export timeout after \d+ms \(\d+ span\(s\)\)$/, + ), + ); + } finally { + vi.useRealTimers(); + } + }); + + it('routes export-threw (synchronous exporter exception) through the injected sink', async () => { + await processor.shutdown(); + const sink = vi.fn(); + processor = new LogToSpanProcessor( + { + export: vi.fn(() => { + throw new Error('exporter exploded synchronously'); + }), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter, + { flushIntervalMs: 60000, diagnosticsSink: sink }, + ); + + await flushOne(processor); + expect(sink).toHaveBeenCalledWith( + '[LogToSpan] export threw: error="exporter exploded synchronously"', + ); + }); + + it('surfaces httpCode/data when a sync-thrown error carries OTLPExporterError fields', async () => { + await processor.shutdown(); + const sink = vi.fn(); + const err = Object.assign(new Error('Bad Request'), { + name: 'OTLPExporterError', + code: 400, + data: 'malformed payload', + }); + processor = new LogToSpanProcessor( + { + export: vi.fn(() => { + throw err; + }), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter, + { flushIntervalMs: 60000, diagnosticsSink: sink }, + ); + + await flushOne(processor); + expect(sink).toHaveBeenCalledWith( + '[LogToSpan] export threw: error="Bad Request" httpCode=400 data="malformed payload"', + ); + }); + + it('JSON-escapes export-threw payloads with embedded newlines (single-line invariant)', async () => { + await processor.shutdown(); + const sink = vi.fn(); + processor = new LogToSpanProcessor( + { + export: vi.fn(() => { + throw new Error('line1\nline2'); + }), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter, + { flushIntervalMs: 60000, diagnosticsSink: sink }, + ); + + await flushOne(processor); + const msg = sink.mock.calls[0][0] as string; + expect(msg).not.toContain('\n'); + expect(msg).toBe('[LogToSpan] export threw: error="line1\\nline2"'); + }); + + it('handles non-Error throws (e.g. throw "string") in the export-threw path', async () => { + await processor.shutdown(); + const sink = vi.fn(); + processor = new LogToSpanProcessor( + { + export: vi.fn(() => { + // Deliberate non-Error throw to exercise the String(err) branch. + // eslint-disable-next-line no-restricted-syntax + throw 'raw string thrown'; + }), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter, + { flushIntervalMs: 60000, diagnosticsSink: sink }, + ); + + await flushOne(processor); + expect(sink).toHaveBeenCalledWith( + '[LogToSpan] export threw: error="raw string thrown"', + ); + }); + + it('keeps processing exports after the sink throws', async () => { + await processor.shutdown(); + const sink = vi.fn(() => { + throw new Error('sink exploded'); + }); + const exportFn = vi.fn( + (_spans, cb: (r: { code: number; error?: Error }) => void) => + cb({ code: 1, error: new Error('boom') }), + ); + processor = new LogToSpanProcessor( + { + export: exportFn, + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter, + { flushIntervalMs: 60000, diagnosticsSink: sink }, + ); + + await flushOne(processor); + await flushOne(processor); + + expect(exportFn).toHaveBeenCalledTimes(2); + expect(sink).toHaveBeenCalledTimes(2); + }); + }); }); diff --git a/packages/core/src/telemetry/log-to-span-processor.ts b/packages/core/src/telemetry/log-to-span-processor.ts index fd6ada9d40f..28490ddfcce 100644 --- a/packages/core/src/telemetry/log-to-span-processor.ts +++ b/packages/core/src/telemetry/log-to-span-processor.ts @@ -45,10 +45,27 @@ const SENSITIVE_ATTRIBUTE_KEYS = new Set([ 'response_text', ]); +/** + * Sink for processor-internal diagnostic messages (export failures, buffer + * overflows, timeouts). Messages are passed without a trailing newline — the + * sink implementation decides how to terminate them. + * + * Default sink writes to stderr to keep diagnostics visible when the host + * environment has no other logging pipeline. Hosts running a TUI should + * inject a sink that routes to a file-based logger to avoid the message + * landing in the rendered terminal area. + */ +export type LogToSpanDiagnosticsSink = (message: string) => void; + +const defaultDiagnosticsSink: LogToSpanDiagnosticsSink = (message) => { + process.stderr.write(`${message}\n`); +}; + interface LogToSpanProcessorOptions { flushIntervalMs?: number; includeSensitiveSpanAttributes?: boolean; maxBufferSize?: number; + diagnosticsSink?: LogToSpanDiagnosticsSink; } /** @@ -61,6 +78,10 @@ interface LogToSpanProcessorOptions { * this processor directly constructs ReadableSpan objects and feeds * them to the exporter. * + * Internal diagnostics (export failures, buffer overflows, timeouts) are + * routed through {@link LogToSpanDiagnosticsSink} so TUI hosts can keep + * them off the rendered terminal area; see the `diagnosticsSink` option. + * * When a log record has a `duration_ms` attribute, the resulting span * will have a matching duration. Otherwise, the span is instantaneous. */ @@ -73,6 +94,7 @@ export class LogToSpanProcessor implements LogRecordProcessor { private cachedTraceId: string | undefined; private readonly includeSensitiveSpanAttributes: boolean; private readonly maxBufferSize: number; + private readonly diagnosticsSink: LogToSpanDiagnosticsSink; private lastBufferOverflowWarningMs: number | undefined; private droppedSpansSinceLastBufferWarning = 0; private totalDroppedSpans = 0; @@ -94,6 +116,7 @@ export class LogToSpanProcessor implements LogRecordProcessor { this.flushIntervalMs = flushIntervalMsOrOptions; this.includeSensitiveSpanAttributes = false; this.maxBufferSize = normalizeMaxBufferSize(maxBufferSize); + this.diagnosticsSink = defaultDiagnosticsSink; } else { this.flushIntervalMs = flushIntervalMsOrOptions.flushIntervalMs ?? 5000; this.includeSensitiveSpanAttributes = @@ -101,6 +124,8 @@ export class LogToSpanProcessor implements LogRecordProcessor { this.maxBufferSize = normalizeMaxBufferSize( flushIntervalMsOrOptions.maxBufferSize, ); + this.diagnosticsSink = + flushIntervalMsOrOptions.diagnosticsSink ?? defaultDiagnosticsSink; } this.flushTimer = setInterval(() => { void this.flush(); @@ -236,12 +261,26 @@ export class LogToSpanProcessor implements LogRecordProcessor { const droppedSinceLastWarning = this.droppedSpansSinceLastBufferWarning; this.droppedSpansSinceLastBufferWarning = 0; this.lastBufferOverflowWarningMs = now; + this.emitDiagnostic( + `[LogToSpan] buffer exceeded max size (${this.maxBufferSize}); dropped ${droppedSinceLastWarning} oldest span(s) since last warning, ${this.totalDroppedSpans} total`, + ); + } + + /** + * Route a diagnostic message to the configured sink, swallowing any sink + * error so a misbehaving sink can never interrupt telemetry ingestion. + * + * Tradeoff: when the sink itself is broken (e.g. file-logger failing on + * EACCES), bridge-specific diagnostics go dark. We accept that — the host + * surfaces overall logging health via `isDebugLoggingDegraded()`, and + * falling back to stderr here would re-introduce the TUI-pollution this + * sink injection was added to prevent. + */ + private emitDiagnostic(message: string): void { try { - process.stderr.write( - `[LogToSpan] buffer exceeded max size (${this.maxBufferSize}); dropped ${droppedSinceLastWarning} oldest span(s) since last warning, ${this.totalDroppedSpans} total\n`, - ); + this.diagnosticsSink(message); } catch { - // Logging diagnostics must not interrupt telemetry ingestion. + // Diagnostics must never interrupt telemetry ingestion. } } @@ -251,8 +290,8 @@ export class LogToSpanProcessor implements LogRecordProcessor { const spans = this.buffer.splice(0); const exportPromise = new Promise((resolve) => { const timeout = setTimeout(() => { - process.stderr.write( - `[LogToSpan] export timeout after ${EXPORT_TIMEOUT_MS}ms\n`, + this.emitDiagnostic( + `[LogToSpan] export timeout after ${EXPORT_TIMEOUT_MS}ms (${spans.length} span(s))`, ); resolve(); }, EXPORT_TIMEOUT_MS); @@ -264,8 +303,8 @@ export class LogToSpanProcessor implements LogRecordProcessor { (result) => { clearTimeout(timeout); if (result.code !== 0) { - process.stderr.write( - `[LogToSpan] export failed: code=${result.code} error=${result.error?.message ?? 'unknown'}\n`, + this.emitDiagnostic( + `[LogToSpan] export failed: code=${result.code} ${formatExportError(result.error)}`, ); } resolve(); @@ -273,9 +312,15 @@ export class LogToSpanProcessor implements LogRecordProcessor { ); } catch (err) { clearTimeout(timeout); - process.stderr.write( - `[LogToSpan] export threw: ${err instanceof Error ? err.message : String(err)}\n`, - ); + // Reuse formatExportError for Error instances so a sync-thrown + // OTLPExporterError surfaces httpCode/data the same way callback + // failures do. Non-Error throws fall back to JSON.stringify to + // preserve the single-line invariant. + const detail = + err instanceof Error + ? formatExportError(err) + : `error=${JSON.stringify(String(err))}`; + this.emitDiagnostic(`[LogToSpan] export threw: ${detail}`); resolve(); } }); @@ -406,6 +451,37 @@ function deriveSpanStatus(attrs: Record | undefined): { return { code: SpanStatusCode.OK }; } +// OTLPExporterError carries an HTTP status `code` and response `data`, but its +// `message` is the HTTP reason-phrase — which is empty on HTTP/2 or when the +// gateway strips it. Surface name/code/data so the operator has something to +// act on (e.g. a 403 from ARMS with empty body). +// +// Both `message` and `data` can carry embedded newlines or other characters +// that would break log parsing when the backend returns a JSON error body. +// JSON.stringify each field to keep the diagnostic on a single line — +// otherwise a torn record breaks downstream log greps and corrupts the +// file-logger format. The 200 figure is JS string length (UTF-16 code +// units), not bytes — non-ASCII payloads may stringify to more bytes; this +// is fine because the cap is a leak/noise budget, not a hard byte limit. +function formatExportError(err: Error | undefined): string { + if (!err) return 'error="unknown"'; + // `code` is typed as `number | string` because Node networking errors (e.g. + // ECONNREFUSED) surface a string here, while OTLPExporterError uses number. + // The `typeof === 'number'` guard below is load-bearing — don't relax it to + // a truthy check or string codes get mislabelled as HTTP statuses. + const extra = err as { code?: number | string; data?: string }; + const msg = err.message || err.name || 'unknown'; + const parts = [`error=${JSON.stringify(msg)}`]; + // `code` is only meaningful as an HTTP status. Networking errors surface + // string codes like 'ECONNREFUSED' on the same field — labelling those as + // `httpCode` would be a lie, so only emit for numeric codes. + if (typeof extra.code === 'number') parts.push(`httpCode=${extra.code}`); + if (typeof extra.data === 'string' && extra.data.length > 0) { + parts.push(`data=${JSON.stringify(extra.data.slice(0, 200))}`); + } + return parts.join(' '); +} + function hrTimeDiff(start: HrTime, end: HrTime): HrTime { let secs = end[0] - start[0]; let nanos = end[1] - start[1]; diff --git a/packages/core/src/telemetry/sdk.test.ts b/packages/core/src/telemetry/sdk.test.ts index 91a0a35532d..63091359a50 100644 --- a/packages/core/src/telemetry/sdk.test.ts +++ b/packages/core/src/telemetry/sdk.test.ts @@ -148,6 +148,7 @@ describe('Telemetry SDK', () => { getSessionId: () => 'test-session', getCliVersion: () => '1.0.0-test', getOutboundCorrelationPropagateTraceContext: () => false, + isInteractive: () => false, } as unknown as Config; }); @@ -343,9 +344,10 @@ describe('Telemetry SDK', () => { }); // Logs falls back to LogToSpanProcessor (bridges logs → spans) expect(OTLPLogExporterHttp).not.toHaveBeenCalled(); - expect(LogToSpanProcessor).toHaveBeenCalledWith(expect.anything(), { - includeSensitiveSpanAttributes: false, - }); + expect(LogToSpanProcessor).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ includeSensitiveSpanAttributes: false }), + ); expect(NodeSDK.prototype.start).toHaveBeenCalled(); }); @@ -362,9 +364,108 @@ describe('Telemetry SDK', () => { initializeTelemetry(mockConfig); - expect(LogToSpanProcessor).toHaveBeenCalledWith(expect.anything(), { - includeSensitiveSpanAttributes: true, - }); + expect(LogToSpanProcessor).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ includeSensitiveSpanAttributes: true }), + ); + }); + + it('in interactive mode, routes log-to-span diagnostics through the OTEL debug logger to avoid TUI pollution', async () => { + vi.spyOn(mockConfig, 'getTelemetryOtlpProtocol').mockReturnValue('http'); + vi.spyOn(mockConfig, 'getTelemetryOtlpEndpoint').mockReturnValue(''); + vi.spyOn(mockConfig, 'getTelemetryOtlpTracesEndpoint').mockReturnValue( + 'http://traces-host/token/api/otlp/traces', + ); + vi.spyOn(mockConfig, 'isInteractive').mockReturnValue(true); + + const mkdirSpy = vi.spyOn(fs, 'mkdir').mockResolvedValue(undefined); + const appendFileSpy = vi + .spyOn(fs, 'appendFile') + .mockResolvedValue(undefined); + const previousDebugLogFileEnv = process.env['QWEN_DEBUG_LOG_FILE']; + try { + process.env['QWEN_DEBUG_LOG_FILE'] = '1'; + setDebugLogSession({ getSessionId: () => 'log-to-span-sink-test' }); + + initializeTelemetry(mockConfig); + + const call = vi.mocked(LogToSpanProcessor).mock.calls.at(-1); + const opts = call?.[1] as { diagnosticsSink?: (m: string) => void }; + expect(typeof opts.diagnosticsSink).toBe('function'); + + opts.diagnosticsSink?.('[LogToSpan] sink wiring smoke test'); + + await vi.waitFor(() => { + expect(appendFileSpy).toHaveBeenCalledWith( + expect.stringContaining('log-to-span-sink-test'), + expectOtelDebugLogLine('WARN', '[LogToSpan] sink wiring smoke test'), + 'utf8', + ); + }); + } finally { + if (previousDebugLogFileEnv === undefined) { + delete process.env['QWEN_DEBUG_LOG_FILE']; + } else { + process.env['QWEN_DEBUG_LOG_FILE'] = previousDebugLogFileEnv; + } + setDebugLogSession(null); + resetDebugLoggingState(); + mkdirSpy.mockRestore(); + appendFileSpy.mockRestore(); + } + }); + + it('in non-interactive mode, leaves diagnostics on the default stderr sink so CI/scripts see export failures', async () => { + vi.spyOn(mockConfig, 'getTelemetryOtlpProtocol').mockReturnValue('http'); + vi.spyOn(mockConfig, 'getTelemetryOtlpEndpoint').mockReturnValue(''); + vi.spyOn(mockConfig, 'getTelemetryOtlpTracesEndpoint').mockReturnValue( + 'http://traces-host/token/api/otlp/traces', + ); + vi.spyOn(mockConfig, 'isInteractive').mockReturnValue(false); + + initializeTelemetry(mockConfig); + + const call = vi.mocked(LogToSpanProcessor).mock.calls.at(-1); + const opts = call?.[1] as { diagnosticsSink?: (m: string) => void }; + // No explicit sink → processor falls back to its default (stderr). + expect(opts.diagnosticsSink).toBeUndefined(); + + // End-to-end check: the real default sink must hit stderr, not silently + // drop. Construct a processor with no sink and trigger a failed export. + const { LogToSpanProcessor: RealProcessor } = await vi.importActual< + typeof import('./log-to-span-processor.js') + >('./log-to-span-processor.js'); + const failingExporter = { + export: ( + _spans: unknown, + cb: (r: { code: number; error?: Error }) => void, + ) => cb({ code: 1, error: new Error('boom') }), + shutdown: () => Promise.resolve(), + forceFlush: () => Promise.resolve(), + }; + const realProcessor = new RealProcessor( + failingExporter as unknown as ConstructorParameters< + typeof RealProcessor + >[0], + { flushIntervalMs: 60000 }, + ); + const stderrWrite = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + try { + realProcessor.onEmit({ + body: 'event', + hrTime: [1000, 0] as [number, number], + attributes: { 'event.name': 'event' }, + } as unknown as Parameters[0]); + await realProcessor.forceFlush(); + expect(stderrWrite).toHaveBeenCalledWith( + '[LogToSpan] export failed: code=1 error="boom"\n', + ); + } finally { + stderrWrite.mockRestore(); + await realProcessor.shutdown(); + } }); it('should warn and skip startup for gRPC per-signal endpoints without base endpoint', () => { @@ -1161,6 +1262,7 @@ describe('refreshSessionContext', () => { getSessionId: () => 'test-session', getCliVersion: () => '1.0.0-test', getOutboundCorrelationPropagateTraceContext: () => false, + isInteractive: () => false, } as unknown as Config; }); diff --git a/packages/core/src/telemetry/sdk.ts b/packages/core/src/telemetry/sdk.ts index 20b2a8ecf77..322fd1763f1 100644 --- a/packages/core/src/telemetry/sdk.ts +++ b/packages/core/src/telemetry/sdk.ts @@ -295,6 +295,20 @@ export function initializeTelemetry(config: Config): void { { includeSensitiveSpanAttributes: config.getTelemetryIncludeSensitiveSpanAttributes(), + // In interactive (TUI) mode, route bridge diagnostics to the OTEL + // debug log file so they don't break out of the Ink render area + // via raw stderr. In non-interactive mode, leave the default sink + // alone so CI / scripts can still see export failures on stderr — + // the canonical diagnostic channel for batch runs. + // + // Caveat for interactive mode: when the user has explicitly + // disabled file logging via QWEN_DEBUG_LOG_FILE=0, debugLogger.warn + // silently no-ops and bridge diagnostics are fully lost — accepted + // trade-off, since falling back to stderr would re-introduce the + // TUI pollution this injection was added to prevent. + ...(config.isInteractive() && { + diagnosticsSink: (message: string) => debugLogger.warn(message), + }), }, ); }