diff --git a/src/lib/core/baseProvider.ts b/src/lib/core/baseProvider.ts index a64232f92..64ebe3a10 100644 --- a/src/lib/core/baseProvider.ts +++ b/src/lib/core/baseProvider.ts @@ -7,9 +7,6 @@ import { IMAGE_GENERATION_MODELS } from "../core/constants.js"; import type { EvaluationData } from "../index.js"; import { MiddlewareFactory } from "../middleware/factory.js"; import type { NeuroLink } from "../neurolink.js"; -import { getMetricsAggregator } from "../observability/metricsAggregator.js"; -import { SpanStatus, SpanType } from "../observability/types/spanTypes.js"; -import { SpanSerializer } from "../observability/utils/spanSerializer.js"; import { ATTR, tracers } from "../telemetry/index.js"; import type { JsonValue, UnknownRecord } from "../types/common.js"; import type { @@ -28,7 +25,6 @@ import type { } from "../types/typeAliases.js"; import { isAbortError } from "../utils/errorHandling.js"; import { logger } from "../utils/logger.js"; -import { calculateCost } from "../utils/pricing.js"; import { composeAbortSignals, createTimeoutController, @@ -164,21 +160,6 @@ export abstract class BaseProvider implements AIProvider { ): Promise { let options = this.normalizeStreamOptions(optionsOrPrompt); - // Observability: create metrics span for provider.stream - const metricsSpan = SpanSerializer.createSpan( - SpanType.MODEL_GENERATION, - "provider.stream", - { - "ai.provider": this.providerName || "unknown", - "ai.model": this.modelName || options.model || "unknown", - "ai.temperature": options.temperature, - "ai.max_tokens": options.maxTokens, - }, - this._traceContext?.parentSpanId, - this._traceContext?.traceId, - ); - let metricsSpanRecorded = false; - // OTEL span for provider-level stream tracing const otelStreamSpan = tracers.provider.startSpan( "neurolink.provider.stream", @@ -298,15 +279,6 @@ export abstract class BaseProvider implements AIProvider { } } } catch (error) { - // Observability: record failed stream span - metricsSpanRecorded = true; - const endedStreamSpan = SpanSerializer.endSpan( - metricsSpan, - SpanStatus.ERROR, - error instanceof Error ? error.message : String(error), - ); - getMetricsAggregator().recordSpan(endedStreamSpan); - otelStreamSpan.setStatus({ code: SpanStatusCode.ERROR, message: error instanceof Error ? error.message : String(error), @@ -315,14 +287,8 @@ export abstract class BaseProvider implements AIProvider { throw error; } finally { - // Observability: record successful stream span (only if not already ended via error path) - if (!metricsSpanRecorded) { - const endedStreamSpan = SpanSerializer.endSpan( - metricsSpan, - SpanStatus.OK, - ); - getMetricsAggregator().recordSpan(endedStreamSpan); - + // End OTEL span on success (only if not already ended via error path) + if (otelStreamSpan.isRecording()) { otelStreamSpan.setStatus({ code: SpanStatusCode.OK }); otelStreamSpan.end(); } @@ -719,20 +685,6 @@ export abstract class BaseProvider implements AIProvider { this.validateOptions(options); const startTime = Date.now(); - // Observability: create metrics span for provider.generate - const metricsSpan = SpanSerializer.createSpan( - SpanType.MODEL_GENERATION, - "provider.generate", - { - "ai.provider": this.providerName || "unknown", - "ai.model": this.modelName || options.model || "unknown", - "ai.temperature": options.temperature, - "ai.max_tokens": options.maxTokens, - }, - this._traceContext?.parentSpanId, - this._traceContext?.traceId, - ); - // OTEL span for provider-level generate tracing // Use startActiveSpan pattern via context.with() so child spans become descendants const otelSpan = tracers.provider.startSpan("neurolink.provider.generate", { @@ -996,47 +948,8 @@ export abstract class BaseProvider implements AIProvider { } } - // Observability: record successful generate span with token/cost data - let enrichedGenerateSpan = { ...metricsSpan }; - if (enhancedResult?.usage) { - enrichedGenerateSpan = SpanSerializer.enrichWithTokenUsage( - enrichedGenerateSpan, - { - promptTokens: enhancedResult.usage.input || 0, - completionTokens: enhancedResult.usage.output || 0, - totalTokens: enhancedResult.usage.total || 0, - }, - ); - const cost = calculateCost(this.providerName, this.modelName, { - input: enhancedResult.usage.input || 0, - output: enhancedResult.usage.output || 0, - total: enhancedResult.usage.total || 0, - }); - if (cost && cost > 0) { - enrichedGenerateSpan = SpanSerializer.enrichWithCost( - enrichedGenerateSpan, - { - totalCost: cost, - }, - ); - } - } - const endedGenerateSpan = SpanSerializer.endSpan( - enrichedGenerateSpan, - SpanStatus.OK, - ); - getMetricsAggregator().recordSpan(endedGenerateSpan); - return await this.enhanceResult(enhancedResult, options, startTime); } catch (error) { - // Observability: record failed generate span - const endedGenerateSpan = SpanSerializer.endSpan( - metricsSpan, - SpanStatus.ERROR, - error instanceof Error ? error.message : String(error), - ); - getMetricsAggregator().recordSpan(endedGenerateSpan); - otelSpan.setStatus({ code: SpanStatusCode.ERROR, message: error instanceof Error ? error.message : String(error), diff --git a/src/lib/neurolink.ts b/src/lib/neurolink.ts index 670646d0d..ddcc30cfa 100644 --- a/src/lib/neurolink.ts +++ b/src/lib/neurolink.ts @@ -2394,7 +2394,16 @@ Current user's request: ${currentInput}`; span.spanId = traceCtx.parentSpanId; span.parentSpanId = undefined; } - span = SpanSerializer.endSpan(span, SpanStatus.OK); + // Mark failed generations with ERROR status so metrics count them correctly + const spanStatus = + data.success === false || data.error + ? SpanStatus.ERROR + : SpanStatus.OK; + span = SpanSerializer.endSpan( + span, + spanStatus, + data.error ? String(data.error) : undefined, + ); span.durationMs = responseTime; if (usage) { @@ -3199,6 +3208,28 @@ Current user's request: ${currentInput}`; code: SpanStatusCode.ERROR, message: error instanceof Error ? error.message : String(error), }); + // Emit generation:end on error so metrics listeners still record the failure. + // Note: variables declared inside try blocks are not accessible in error + // handlers, so we extract what we can from the original input. + const errProvider = + typeof optionsOrPrompt === "object" + ? (optionsOrPrompt as GenerateOptions).provider || "unknown" + : "unknown"; + const errModel = + typeof optionsOrPrompt === "object" + ? (optionsOrPrompt as GenerateOptions).model || "unknown" + : "unknown"; + try { + this.emitter.emit("generation:end", { + provider: errProvider, + model: errModel, + responseTime: 0, + error: error instanceof Error ? error.message : String(error), + success: false, + }); + } catch (emitError: unknown) { + void emitError; // non-blocking — error event emission is best-effort + } throw error; } finally { generateSpan.end(); diff --git a/src/lib/observability/exporters/langfuseExporter.ts b/src/lib/observability/exporters/langfuseExporter.ts index cf3ec936c..627a8c0f5 100644 --- a/src/lib/observability/exporters/langfuseExporter.ts +++ b/src/lib/observability/exporters/langfuseExporter.ts @@ -154,7 +154,9 @@ export class LangfuseExporter extends BaseExporter { name: span.name, userId: span.attributes["user.id"] as string | undefined, sessionId: span.attributes["session.id"] as string | undefined, - metadata: span.attributes, + // Only pick safe, non-PII attributes for metadata — intentionally excludes + // input, output, error.stack, and other user content to match Braintrust exporter + metadata: filterSafeMetadata(span.attributes), release: this.release, tags: this.extractTags(span), }; @@ -249,3 +251,6 @@ export class LangfuseExporter extends BaseExporter { return tags; } } + +// Safe metadata filtering imported from shared module to avoid duplication +import { filterSafeMetadata } from "../utils/safeMetadata.js"; diff --git a/src/lib/observability/metricsAggregator.ts b/src/lib/observability/metricsAggregator.ts index 334eee352..17cac0eb2 100644 --- a/src/lib/observability/metricsAggregator.ts +++ b/src/lib/observability/metricsAggregator.ts @@ -179,8 +179,12 @@ export class MetricsAggregator { recordSpan(span: SpanData): void { // Enforce maximum spans limit if (this.spans.length >= this.config.maxSpansRetained) { - this.spans.shift(); // Remove oldest span - // Note: We keep aggregated metrics, only raw spans are trimmed + const evicted = this.spans.shift(); // Remove oldest span + // Only trim latencyValues when the evicted span had a duration recorded + if (evicted?.durationMs !== undefined) { + this.latencyValues.shift(); + } + // Note: We keep aggregated metrics, only raw spans and latency values are trimmed } this.spans.push(span); diff --git a/src/lib/observability/spanProcessor.ts b/src/lib/observability/spanProcessor.ts index 4a99767af..2b00ef13f 100644 --- a/src/lib/observability/spanProcessor.ts +++ b/src/lib/observability/spanProcessor.ts @@ -102,6 +102,8 @@ export class RedactionProcessor implements SpanProcessor { "credentials", "private_key", "privateKey", + "stack", + "error.stack", ], ); this.redactedValue = config?.redactedValue ?? "[REDACTED]"; @@ -308,11 +310,24 @@ export class BatchProcessor implements SpanProcessor { }, this.flushIntervalMs); } + // Note: flush() is intentionally synchronous. The onBatchReady callback is + // typed as `(spans: SpanData[]) => void` — callers must not pass async + // exporters. If async export is needed, the callback should handle its own + // error reporting (e.g. fire-and-forget with promise error handlers). private flush(): void { if (this.batch.length > 0 && this.onBatchReady) { const spans = [...this.batch]; - this.batch = []; - this.onBatchReady(spans); + try { + this.onBatchReady(spans); + this.batch = []; + } catch (flushError: unknown) { + // Keep spans for next flush attempt, but cap backlog growth + void flushError; // acknowledged — error is expected during exporter outages + const maxBacklog = this.batchSize * 20; + if (this.batch.length > maxBacklog) { + this.batch = this.batch.slice(this.batch.length - maxBacklog); + } + } } } diff --git a/src/lib/observability/utils/safeMetadata.ts b/src/lib/observability/utils/safeMetadata.ts new file mode 100644 index 000000000..91dbb0096 --- /dev/null +++ b/src/lib/observability/utils/safeMetadata.ts @@ -0,0 +1,31 @@ +/** + * Safe metadata filtering for observability exporters. + * + * Only these attribute keys are forwarded to third-party backends as trace + * metadata. User prompts (input), LLM responses (output), error stacks, and + * any other potentially sensitive data are excluded to prevent PII leaks. + */ + +import type { SpanAttributes } from "../types/spanTypes.js"; + +// Only ai.* keys are forwarded as metadata. Stream metrics (chunk_count, +// content_length) should be accessed via span attributes directly, not via +// metadata sent to third-party backends. +export const SAFE_METADATA_KEYS = new Set([ + "ai.provider", + "ai.model", + "ai.temperature", + "ai.max_tokens", +]); + +export function filterSafeMetadata( + attributes: SpanAttributes, +): Record { + const filtered: Record = {}; + for (const key of SAFE_METADATA_KEYS) { + if (attributes[key] !== undefined) { + filtered[key] = attributes[key]; + } + } + return filtered; +} diff --git a/src/lib/observability/utils/spanSerializer.ts b/src/lib/observability/utils/spanSerializer.ts index 97798d1ff..fb84d15b4 100644 --- a/src/lib/observability/utils/spanSerializer.ts +++ b/src/lib/observability/utils/spanSerializer.ts @@ -142,7 +142,9 @@ export class SpanSerializer { name: span.name, startTime: span.startTime, endTime: span.endTime, - metadata: { ...span.attributes }, + // Only pick safe, non-PII attributes for metadata — intentionally excludes + // input, output, error.stack, and other user content for PII safety + metadata: filterSafeMetadata(span.attributes), level: span.status === SpanStatus.ERROR ? "ERROR" : "DEFAULT", statusMessage: span.statusMessage, input: span.attributes["input"], @@ -389,3 +391,6 @@ export class SpanSerializer { }); } } + +// Safe metadata filtering imported from shared module to avoid duplication +import { filterSafeMetadata } from "./safeMetadata.js"; diff --git a/src/lib/providers/googleVertex.ts b/src/lib/providers/googleVertex.ts index b7077e7c7..c70dffe71 100644 --- a/src/lib/providers/googleVertex.ts +++ b/src/lib/providers/googleVertex.ts @@ -1721,9 +1721,7 @@ export class GoogleVertexProvider extends BaseProvider { }, { role: "model", - parts: [ - { text: "Understood. I will follow these instructions." }, - ], + parts: [{ text: "OK" }], }, ...currentContents, ]; @@ -1999,9 +1997,7 @@ export class GoogleVertexProvider extends BaseProvider { }, { role: "model", - parts: [ - { text: "Understood. I will follow these instructions." }, - ], + parts: [{ text: "OK" }], }, ...currentContents, ]; diff --git a/test/zod-schema-test-function.ts b/test/zod-schema-test-function.ts index 757fa24d3..32d0cf62b 100644 --- a/test/zod-schema-test-function.ts +++ b/test/zod-schema-test-function.ts @@ -394,9 +394,12 @@ export async function testComplexZodSchemaMultiProvider( const providerName = providerOverride || "vertex"; const modelName = modelOverride || undefined; - // Check if this is a Gemini MODEL - Gemini cannot use tools + JSON schema together + // Check if this is a Gemini MODEL - Gemini cannot use tools + JSON schema together. // This is a documented Gemini limitation, not a bug in the SDK. - // Note: provider name alone is not sufficient - Vertex can run Claude too. + // When an explicit model is provided, we check its name. When no model is given, + // google-ai, googleAiStudio, and vertex all default to a Gemini model, so we + // treat them as Gemini. If someone passes e.g. --provider=vertex --model=claude-*, + // the modelName check will NOT match "gemini" and the test will correctly run. const isGeminiModel = modelName?.toLowerCase().includes("gemini") || (!modelName &&