'use strict'; var base = require('@mastra/core/base'); var error = require('@mastra/core/error'); var logger = require('@mastra/core/logger'); var observability = require('@mastra/core/observability'); var utils = require('@mastra/core/utils'); var v4 = require('zod/v4'); var fs = require('fs'); var module$1 = require('module'); var path = require('path'); var web = require('stream/web'); var features = require('@mastra/core/features'); var storage = require('@mastra/core/storage'); var _documentCurrentScript = typeof document !== 'undefined' ? document.currentScript : null; function _interopDefault (e) { return e && e.__esModule ? e : { default: e }; } var fs__default = /*#__PURE__*/_interopDefault(fs); var path__default = /*#__PURE__*/_interopDefault(path); // src/default.ts function routeToHandler(handler, event, logger) { try { switch (event.type) { case observability.TracingEventType.SPAN_STARTED: case observability.TracingEventType.SPAN_UPDATED: case observability.TracingEventType.SPAN_ENDED: { const fn = handler.onTracingEvent ? handler.onTracingEvent.bind(handler) : handler.exportTracingEvent.bind(handler); return catchAsyncResult(fn(event), handler.name, "tracing", logger); } case "log": if (handler.onLogEvent) { return catchAsyncResult(handler.onLogEvent(event), handler.name, "log", logger); } break; case "metric": if (handler.onMetricEvent) { return catchAsyncResult(handler.onMetricEvent(event), handler.name, "metric", logger); } break; case "score": if (handler.onScoreEvent) { return catchAsyncResult(handler.onScoreEvent(event), handler.name, "score", logger); } if (handler.addScoreToTrace) { const score = event.score; if (!score.traceId) break; return catchAsyncResult( handler.addScoreToTrace({ traceId: score.traceId, ...score.spanId ? { spanId: score.spanId } : {}, score: score.score, ...score.reason ? { reason: score.reason } : {}, scorerName: score.scorerName ?? score.scorerId, ...score.metadata ? { metadata: score.metadata } : {} }), handler.name, "score", logger ); } break; case "feedback": if (handler.onFeedbackEvent) { return catchAsyncResult(handler.onFeedbackEvent(event), handler.name, "feedback", logger); } break; } } catch (err) { logger.error(`[Observability] Handler error [handler=${handler.name}]:`, err); } } function routeDropToHandler(handler, event, logger) { try { if (handler.onDroppedEvent) { return catchAsyncResult(handler.onDroppedEvent(event), handler.name, "drop", logger); } } catch (err) { logger.error(`[Observability] Handler error [handler=${handler.name}]:`, err); } } function catchAsyncResult(result, handlerName, signal, logger) { if (result && typeof result.then === "function") { return result.catch((err) => { logger.error(`[Observability] ${signal} handler error [handler=${handlerName}]:`, err); }); } } var BaseObservabilityEventBus = class _BaseObservabilityEventBus extends base.MastraBase { subscribers = /* @__PURE__ */ new Set(); /** In-flight async subscriber promises. Self-cleaning via .finally(). */ pendingSubscribers = /* @__PURE__ */ new Set(); constructor({ name } = {}) { super({ component: logger.RegisteredLogger.OBSERVABILITY, name: name ?? "EventBus" }); } /** * Dispatch an event to all subscribers synchronously. * Async handler promises are tracked internally and drained by {@link flush}. * * @param event - The event to broadcast to subscribers. */ emit(event) { for (const handler of this.subscribers) { try { const result = handler(event); if (result && typeof result.then === "function") { const promise = result.catch((err) => { this.logger.error("[ObservabilityEventBus] Handler error:", err); }); this.pendingSubscribers.add(promise); void promise.finally(() => this.pendingSubscribers.delete(promise)); } } catch (err) { this.logger.error("[ObservabilityEventBus] Handler error:", err); } } } /** * Register a handler to receive future events. * * @param handler - Callback invoked synchronously on each {@link emit}. * @returns An unsubscribe function that removes the handler. */ subscribe(handler) { this.subscribers.add(handler); return () => { this.subscribers.delete(handler); }; } /** Max flush drain iterations before bailing — prevents infinite loops when handlers re-emit. */ static MAX_FLUSH_ITERATIONS = 3; /** Await all in-flight async subscriber promises, draining until empty. */ async flush() { let iterations = 0; while (this.pendingSubscribers.size > 0) { await Promise.allSettled([...this.pendingSubscribers]); iterations++; if (iterations >= _BaseObservabilityEventBus.MAX_FLUSH_ITERATIONS) { this.logger.error( `[ObservabilityEventBus] flush() exceeded ${_BaseObservabilityEventBus.MAX_FLUSH_ITERATIONS} drain iterations \u2014 ${this.pendingSubscribers.size} promises still pending. Handlers may be re-emitting during flush.` ); break; } } } /** Flush pending promises, then clear all subscribers. */ async shutdown() { await this.flush(); this.subscribers.clear(); } }; // src/spans/serialization.ts var DEFAULT_KEYS_TO_STRIP = /* @__PURE__ */ new Set([ "logger", "experimental_providerMetadata", "providerMetadata", "steps", "tracingContext", "execute", // Tool execute functions "validate" // Schema validate functions ]); var DEFAULT_DEEP_CLEAN_OPTIONS = Object.freeze({ keysToStrip: DEFAULT_KEYS_TO_STRIP, maxDepth: 8, maxStringLength: 128 * 1024, // 128KB - sufficient for large LLM prompts/responses maxArrayLength: 50, maxObjectKeys: 50 }); function mergeSerializationOptions(userOptions) { if (!userOptions) { return DEFAULT_DEEP_CLEAN_OPTIONS; } return { keysToStrip: DEFAULT_KEYS_TO_STRIP, maxDepth: userOptions.maxDepth ?? DEFAULT_DEEP_CLEAN_OPTIONS.maxDepth, maxStringLength: userOptions.maxStringLength ?? DEFAULT_DEEP_CLEAN_OPTIONS.maxStringLength, maxArrayLength: userOptions.maxArrayLength ?? DEFAULT_DEEP_CLEAN_OPTIONS.maxArrayLength, maxObjectKeys: userOptions.maxObjectKeys ?? DEFAULT_DEEP_CLEAN_OPTIONS.maxObjectKeys }; } function truncateString(s, maxChars) { if (s.length <= maxChars) { return s; } return s.slice(0, maxChars) + "\u2026[truncated]"; } function formatSerializationError(error) { return `[${error instanceof Error ? truncateString(error.message, 256) : "unknown error"}]`; } function getMapKeyType(key) { if (key === null) { return "null"; } if (key instanceof Date) { return "date"; } if (Array.isArray(key)) { return "array"; } if (key instanceof Map) { return "map"; } if (key instanceof Set) { return "set"; } if (key instanceof Error) { return "error"; } return typeof key; } function restoreSerializedMapKey(keyType, key) { switch (keyType) { case "undefined": return void 0; case "null": return null; case "bigint": return typeof key === "string" && key.endsWith("n") ? BigInt(key.slice(0, -1)) : key; case "date": return typeof key === "string" ? new Date(key) : key; default: return key; } } function isSerializedMap(value) { return typeof value === "object" && value !== null && value.__type === "Map" && Array.isArray(value.__map_entries); } function reconstructSerializedMap(value) { return new Map( value.__map_entries.map(([keyType, key, mapValue]) => [restoreSerializedMapKey(keyType, key), mapValue]) ); } function isJsonSchema(val) { if (typeof val !== "object" || val === null) return false; if (val.$schema && typeof val.$schema === "string" && val.$schema.includes("json-schema")) { return true; } if (val.type === "object" && val.properties && typeof val.properties === "object") { return true; } return false; } function deepClean(value, options = DEFAULT_DEEP_CLEAN_OPTIONS) { const { keysToStrip, maxDepth, maxStringLength, maxArrayLength, maxObjectKeys } = options; const stripSet = keysToStrip instanceof Set ? keysToStrip : new Set(Array.isArray(keysToStrip) ? keysToStrip : Object.keys(keysToStrip)); const ancestors = /* @__PURE__ */ new WeakSet(); function helper(val, depth) { if (depth > maxDepth) { return "[MaxDepth]"; } if (val === null || val === void 0) { return val; } if (typeof val === "string") { return truncateString(val, maxStringLength); } if (typeof val === "number" || typeof val === "boolean") { return val; } if (typeof val === "bigint") { return `${val}n`; } if (typeof val === "function") { return "[Function]"; } if (typeof val === "symbol") { return val.description ? `[Symbol(${val.description})]` : "[Symbol]"; } if (val instanceof Date) { return val; } if (typeof val === "object") { if (ancestors.has(val)) { return "[Circular]"; } ancestors.add(val); } try { if (val instanceof Error) { let errorName; let errorMessage; let errorStack; let rawCause; let causeReadFailed = false; try { errorName = val.name; } catch (error) { errorName = formatSerializationError(error); } try { errorMessage = val.message; } catch (error) { errorMessage = formatSerializationError(error); } try { errorStack = val.stack; } catch (error) { errorStack = formatSerializationError(error); } try { rawCause = val.cause; } catch (error) { causeReadFailed = true; rawCause = formatSerializationError(error); } const cleanedError = { name: typeof errorName === "string" ? truncateString(errorName, maxStringLength) : errorName, message: typeof errorMessage === "string" ? truncateString(errorMessage, maxStringLength) : errorMessage }; if (typeof errorStack === "string") { cleanedError.stack = truncateString(errorStack, maxStringLength); } else if (errorStack !== void 0) { cleanedError.stack = errorStack; } if (causeReadFailed) { cleanedError.cause = rawCause; } else if (rawCause !== void 0) { try { cleanedError.cause = helper(rawCause, depth + 1); } catch (error) { cleanedError.cause = formatSerializationError(error); } } return cleanedError; } if (val instanceof Map) { const cleanedMap = { __type: "Map", __map_entries: [] }; let mapKeyCount = 0; let omittedMapEntries = 0; for (const [mapKey, mapVal] of val) { if (typeof mapKey === "string" && stripSet.has(mapKey)) { continue; } if (mapKeyCount >= maxObjectKeys) { omittedMapEntries++; continue; } const mapKeyType = getMapKeyType(mapKey); let cleanedMapKey; let cleanedMapValue; try { cleanedMapKey = helper(mapKey, depth + 1); } catch (error) { cleanedMapKey = formatSerializationError(error); } try { cleanedMapValue = helper(mapVal, depth + 1); } catch (error) { cleanedMapValue = formatSerializationError(error); } cleanedMap.__map_entries.push([mapKeyType, cleanedMapKey, cleanedMapValue]); mapKeyCount++; } if (omittedMapEntries > 0) { cleanedMap.__truncated = `${omittedMapEntries} more keys omitted`; } return cleanedMap; } if (val instanceof Set) { const cleanedSet = []; let i = 0; const totalSetSize = val.size; for (const item of val) { if (i >= maxArrayLength) break; try { cleanedSet.push(helper(item, depth + 1)); } catch (error) { cleanedSet.push(formatSerializationError(error)); } i++; } if (totalSetSize > maxArrayLength) { cleanedSet.push(`[\u2026${totalSetSize - maxArrayLength} more items]`); } return cleanedSet; } if (Array.isArray(val)) { const cleaned2 = []; for (let i = 0; i < Math.min(val.length, maxArrayLength); i++) { try { cleaned2.push(helper(val[i], depth + 1)); } catch (error) { cleaned2.push(formatSerializationError(error)); } } if (val.length > maxArrayLength) { cleaned2.push(`[\u2026${val.length - maxArrayLength} more items]`); } return cleaned2; } if (typeof Buffer !== "undefined" && Buffer.isBuffer(val)) { return `[Buffer length=${val.length}]`; } if (ArrayBuffer.isView(val)) { const ctor = val.constructor?.name ?? "TypedArray"; const byteLength = val.byteLength ?? "?"; return `[${ctor} byteLength=${byteLength}]`; } if (val instanceof ArrayBuffer) { return `[ArrayBuffer byteLength=${val.byteLength}]`; } let serializeForSpan; try { serializeForSpan = val.serializeForSpan; } catch (error) { return `[serializeForSpan failed: ${error instanceof Error ? truncateString(error.message, 256) : "unknown error"}]`; } if (typeof serializeForSpan === "function") { try { return helper(serializeForSpan.call(val), depth); } catch (error) { return `[serializeForSpan failed: ${error instanceof Error ? truncateString(error.message, 256) : "unknown error"}]`; } } let looksLikeJsonSchema = false; try { looksLikeJsonSchema = isJsonSchema(val); } catch { looksLikeJsonSchema = false; } if (looksLikeJsonSchema) { return val; } const cleaned = {}; const keys = Object.keys(val).filter((key) => !stripSet.has(key)); let keyCount = 0; for (const key of keys) { if (keyCount >= maxObjectKeys) { cleaned["__truncated"] = `${keys.length - keyCount} more keys omitted`; break; } try { cleaned[key] = helper(val[key], depth + 1); keyCount++; } catch (error) { cleaned[key] = formatSerializationError(error); keyCount++; } } return cleaned; } finally { if (typeof val === "object" && val !== null) { ancestors.delete(val); } } } return helper(value, 0); } // src/bus/observability-bus.ts function cleanEvent(event, options) { switch (event.type) { case "log": return { type: "log", log: deepClean(event.log, options) }; case "metric": return { type: "metric", metric: deepClean(event.metric, options) }; case "score": return { type: "score", score: deepClean(event.score, options) }; case "feedback": return { type: "feedback", feedback: deepClean(event.feedback, options) }; default: return event; } } var MAX_FLUSH_ITERATIONS = 3; var ObservabilityBus = class extends BaseObservabilityEventBus { exporters = []; bridge; /** In-flight handler promises from routeToHandler. Self-cleaning via .finally(). */ pendingHandlers = /* @__PURE__ */ new Set(); handlerBufferFlushDepth = 0; dropEventsEmittedDuringHandlerFlush = 0; /** Resolved deepClean options applied to non-tracing events before fan-out. */ deepCleanOptions; constructor(opts) { super({ name: "ObservabilityBus" }); this.deepCleanOptions = mergeSerializationOptions(opts?.serializationOptions); } /** * Register an exporter to receive routed events. * Duplicate registrations (same instance) are silently ignored. * * @param exporter - The exporter to register. */ registerExporter(exporter) { if (this.exporters.includes(exporter)) { return; } this.exporters.push(exporter); } /** * Unregister an exporter. * * @param exporter - The exporter instance to remove. * @returns `true` if the exporter was found and removed, `false` otherwise. */ unregisterExporter(exporter) { const index = this.exporters.indexOf(exporter); if (index !== -1) { this.exporters.splice(index, 1); return true; } return false; } /** * Get registered exporters (read-only snapshot). */ getExporters() { return [...this.exporters]; } /** * Register a bridge to receive all routed events alongside exporters. * Only one bridge can be registered at a time; replacing an existing bridge * logs a warning. * * @param bridge - The bridge to register. */ registerBridge(bridge) { if (this.bridge) { this.logger.warn(`[ObservabilityBus] Replacing existing bridge with new bridge`); } this.bridge = bridge; } /** * Unregister the bridge. * * @returns `true` if a bridge was registered and removed, `false` otherwise. */ unregisterBridge() { if (this.bridge) { this.bridge = void 0; return true; } return false; } /** * Get the registered bridge, if any. */ getBridge() { return this.bridge; } /** * Emit an event: route to exporter/bridge handlers, then forward to base * class for subscriber delivery. * * emit() is synchronous — async handler promises are tracked internally * and can be drained via flush(). */ emit(event) { const cleaned = cleanEvent(event, this.deepCleanOptions); for (const exporter of this.exporters) { this.trackPromise(routeToHandler(exporter, cleaned, this.logger)); } if (this.bridge) { this.trackPromise(routeToHandler(this.bridge, cleaned, this.logger)); } super.emit(cleaned); } /** * Emit exporter pipeline drop events to exporters and the bridge. * * Drop events describe exporter health, not user observability data, so they * are intentionally not delivered to generic event-bus subscribers. */ emitDropEvent(event) { if (this.handlerBufferFlushDepth > 0) { this.dropEventsEmittedDuringHandlerFlush++; } for (const exporter of this.exporters) { this.trackPromise(routeDropToHandler(exporter, event, this.logger)); } if (this.bridge) { this.trackPromise(routeDropToHandler(this.bridge, event, this.logger)); } } /** * Track an async handler promise so flush() can await it. * No-ops for sync (void) results. */ trackPromise(result) { if (result && typeof result.then === "function") { const promise = result; this.pendingHandlers.add(promise); void promise.finally(() => this.pendingHandlers.delete(promise)); } } /** Await in-flight routed handler promises, draining until empty. */ async drainPendingHandlers() { let iterations = 0; while (this.pendingHandlers.size > 0) { await Promise.allSettled([...this.pendingHandlers]); iterations++; if (iterations >= MAX_FLUSH_ITERATIONS) { this.logger.error( `[ObservabilityBus] flush() exceeded ${MAX_FLUSH_ITERATIONS} drain iterations \u2014 ${this.pendingHandlers.size} promises still pending. Handlers may be re-emitting during flush.` ); if (this.pendingHandlers.size > 0) { await Promise.allSettled([...this.pendingHandlers]); } break; } } } /** Drain exporter and bridge SDK-internal buffers. */ async flushHandlerBuffers() { const initialDropCount = this.dropEventsEmittedDuringHandlerFlush; this.handlerBufferFlushDepth++; try { const bufferFlushPromises = this.exporters.map((e) => e.flush()); if (this.bridge) { bufferFlushPromises.push(this.bridge.flush()); } if (bufferFlushPromises.length > 0) { await Promise.allSettled(bufferFlushPromises); } return this.dropEventsEmittedDuringHandlerFlush > initialDropCount; } finally { this.handlerBufferFlushDepth--; } } /** * Multi-phase flush to ensure all observability data is fully exported. * * **Phase 1 — Delivery:** Await all in-flight handler promises (exporters, * bridge, and base-class subscribers). After this resolves, all event data * has been delivered to handler methods. * * **Phase 2 — Buffer drain:** Call flush() on each exporter and bridge to * drain their SDK-internal buffers (e.g., OTEL BatchSpanProcessor, Langfuse * client queue). Phases are sequential — buffer drains must not start until * delivery completes, otherwise exporters would flush empty buffers. * * Exporter flushes can emit drop events. When that happens, flush loops * through delivery and buffer drain again so alerting integrations that buffer * drop notifications are drained before returning. */ async flush() { await this.drainPendingHandlers(); await super.flush(); for (let iterations = 0; iterations < MAX_FLUSH_ITERATIONS; iterations++) { const emittedDropEvents = await this.flushHandlerBuffers(); if (!emittedDropEvents && this.pendingHandlers.size === 0) { return; } await this.drainPendingHandlers(); await super.flush(); } this.logger.error( `[ObservabilityBus] flush() exceeded ${MAX_FLUSH_ITERATIONS} buffer drain iterations. Handlers may be emitting drop events during every flush.` ); } /** Flush all pending events and exporter buffers, then clear subscribers. */ async shutdown() { await this.flush(); await super.shutdown(); } }; var SamplingStrategyType = /* @__PURE__ */ ((SamplingStrategyType2) => { SamplingStrategyType2["ALWAYS"] = "always"; SamplingStrategyType2["NEVER"] = "never"; SamplingStrategyType2["RATIO"] = "ratio"; SamplingStrategyType2["CUSTOM"] = "custom"; return SamplingStrategyType2; })(SamplingStrategyType || {}); var functionSchema = v4.z.custom((value) => typeof value === "function", { message: "Expected function" }); var samplingStrategySchema = v4.z.discriminatedUnion("type", [ v4.z.object({ type: v4.z.literal("always" /* ALWAYS */) }), v4.z.object({ type: v4.z.literal("never" /* NEVER */) }), v4.z.object({ type: v4.z.literal("ratio" /* RATIO */), probability: v4.z.number().min(0, "Probability must be between 0 and 1").max(1, "Probability must be between 0 and 1") }), v4.z.object({ type: v4.z.literal("custom" /* CUSTOM */), sampler: functionSchema }) ]); var serializationOptionsSchema = v4.z.object({ maxStringLength: v4.z.number().int().positive().optional(), maxDepth: v4.z.number().int().positive().optional(), maxArrayLength: v4.z.number().int().positive().optional(), maxObjectKeys: v4.z.number().int().positive().optional() }).optional(); var LOG_LEVELS = ["debug", "info", "warn", "error", "fatal"]; var cardinalityConfigSchema = v4.z.object({ blockedLabels: v4.z.array(v4.z.string()).optional(), blockUUIDs: v4.z.boolean().optional() }).optional(); var loggingConfigSchema = v4.z.object({ enabled: v4.z.boolean().optional(), level: v4.z.enum(LOG_LEVELS).optional() }).optional(); var spanFilterSchema = functionSchema.optional(); var observabilityInstanceConfigFields = { serviceName: v4.z.string().min(1, "Service name is required"), sampling: samplingStrategySchema.optional(), exporters: v4.z.array(v4.z.any()).optional(), bridge: v4.z.any().optional(), spanOutputProcessors: v4.z.array(v4.z.any()).optional(), includeInternalSpans: v4.z.boolean().optional(), excludeSpanTypes: v4.z.array(v4.z.nativeEnum(observability.SpanType)).optional(), spanFilter: spanFilterSchema, requestContextKeys: v4.z.array(v4.z.string()).optional(), serializationOptions: serializationOptionsSchema, cardinality: cardinalityConfigSchema, logging: loggingConfigSchema }; var observabilityInstanceConfigSchema = v4.z.object({ name: v4.z.string().min(1, "Name is required"), ...observabilityInstanceConfigFields }).refine( (data) => { const hasExporters = data.exporters && data.exporters.length > 0; const hasBridge = !!data.bridge; return hasExporters || hasBridge; }, { message: "At least one exporter or a bridge is required" } ); var observabilityConfigValueSchema = v4.z.object(observabilityInstanceConfigFields).refine( (data) => { const hasExporters = data.exporters && data.exporters.length > 0; const hasBridge = !!data.bridge; return hasExporters || hasBridge; }, { message: "At least one exporter or a bridge is required" } ); var sensitiveDataFilterOptionsSchema = v4.z.object({ sensitiveFields: v4.z.array(v4.z.string()).optional(), redactionToken: v4.z.string().optional(), redactionStyle: v4.z.enum(["full", "partial"]).optional() }).strict(); var observabilityRegistryConfigSchema = v4.z.object({ default: v4.z.object({ enabled: v4.z.boolean().optional() }).optional().nullable(), configs: v4.z.union([v4.z.record(v4.z.string(), v4.z.any()), v4.z.array(v4.z.any()), v4.z.null()]).optional(), configSelector: functionSchema.optional(), sensitiveDataFilter: v4.z.union([v4.z.boolean(), sensitiveDataFilterOptionsSchema]).optional() }).passthrough().refine( (data) => { const isDefaultEnabled = data.default?.enabled === true; const hasConfigs = data.configs && typeof data.configs === "object" && !Array.isArray(data.configs) ? Object.keys(data.configs).length > 0 : false; return !(isDefaultEnabled && hasConfigs); }, { message: 'Cannot specify both "default" (when enabled) and "configs". Use either default observability or custom configs, but not both.' } ).refine( (data) => { const configCount = data.configs && typeof data.configs === "object" && !Array.isArray(data.configs) ? Object.keys(data.configs).length : 0; if (configCount > 1 && !data.configSelector) { return false; } return true; }, { message: 'A "configSelector" function is required when multiple configs are specified to determine which config to use.' } ).refine( (data) => { if (data.configSelector) { const isDefaultEnabled = data.default?.enabled === true; const hasConfigs = data.configs && typeof data.configs === "object" && !Array.isArray(data.configs) ? Object.keys(data.configs).length > 0 : false; return isDefaultEnabled || hasConfigs; } return true; }, { message: 'A "configSelector" requires at least one config or default observability to be configured.' } ); var LOG_LEVEL_PRIORITY = { debug: 0, info: 1, warn: 2, error: 3, fatal: 4 }; var LoggerContextImpl = class { config; /** * Create a logger context. Context and metadata are defensively copied so * mutations after construction do not affect emitted logs. */ constructor(config) { const correlationContext = config.correlationContext ? { ...config.correlationContext } : void 0; this.config = { ...config, traceId: config.traceId ?? correlationContext?.traceId, spanId: config.spanId ?? correlationContext?.spanId, correlationContext, metadata: config.metadata ? structuredClone(config.metadata) : void 0 }; } /** Log at DEBUG level. */ debug(message, data) { this.log("debug", message, data); } /** Log at INFO level. */ info(message, data) { this.log("info", message, data); } /** Log at WARN level. */ warn(message, data) { this.log("warn", message, data); } /** Log at ERROR level. */ error(message, data) { this.log("error", message, data); } /** Log at FATAL level. */ fatal(message, data) { this.log("fatal", message, data); } /** * Build an ExportedLog, check against the minimum level, and emit it through the bus. */ log(level, message, data) { const minLevel = this.config.minLevel ?? "warn"; if (LOG_LEVEL_PRIORITY[level] < LOG_LEVEL_PRIORITY[minLevel]) { return; } const exportedLog = { logId: observability.generateSignalId(), timestamp: /* @__PURE__ */ new Date(), level, message, data, traceId: this.config.traceId, spanId: this.config.spanId, correlationContext: this.config.correlationContext, metadata: this.config.metadata }; const event = { type: "log", log: exportedLog }; this.config.observabilityBus.emit(event); } }; var MetricsContextImpl = class { traceId; spanId; correlationContext; metadata; cardinalityFilter; observabilityBus; /** * Create a metrics context. Correlation context and metadata are defensively * copied so mutations after construction do not affect emitted metrics. */ constructor(config) { this.correlationContext = config.correlationContext ? { ...config.correlationContext } : void 0; this.traceId = config.traceId ?? this.correlationContext?.traceId; this.spanId = config.spanId ?? this.correlationContext?.spanId; this.metadata = config.metadata ? structuredClone(config.metadata) : void 0; this.cardinalityFilter = config.cardinalityFilter; this.observabilityBus = config.observabilityBus; } /** Emit a metric observation. */ emit(name, value, labels, options) { if (!Number.isFinite(value) || value < 0) { return; } const filteredLabels = labels ? this.cardinalityFilter.filterLabels(labels) : {}; const costContext = options?.costContext ? cloneCostContext(options.costContext) : void 0; const exportedMetric = { metricId: observability.generateSignalId(), timestamp: /* @__PURE__ */ new Date(), traceId: this.traceId, spanId: this.spanId, name, value, labels: filteredLabels, correlationContext: this.correlationContext, costContext, metadata: this.metadata }; const event = { type: "metric", metric: exportedMetric }; this.observabilityBus.emit(event); } /** @deprecated Use `emit()` instead. */ counter(name) { return { add: (value, additionalLabels) => { this.emit(name, value, additionalLabels); } }; } /** @deprecated Use `emit()` instead. */ gauge(name) { return { set: (value, additionalLabels) => { this.emit(name, value, additionalLabels); } }; } /** @deprecated Use `emit()` instead. */ histogram(name) { return { record: (value, additionalLabels) => { this.emit(name, value, additionalLabels); } }; } }; function cloneCostContext(costContext) { return { provider: costContext.provider, model: costContext.model, estimatedCost: costContext.estimatedCost, costUnit: costContext.costUnit, costMetadata: costContext.costMetadata ? structuredClone(costContext.costMetadata) : void 0 }; } // src/metrics/pricing-model.ts var PricingTier = class { index; when; rates; constructor(args) { this.index = args.index; this.when = args.when; this.rates = args.rates; } matchesUsage(usage) { if (!this.when || this.when.length === 0) { return true; } return this.when.every((condition) => this.matchesCondition(condition, usage)); } hasMatchingMeterForUsage(meter) { return Boolean(meter && typeof this.rates[meter] === "number"); } matchesCondition(condition, usage) { const left = this.getConditionFieldValue(condition.field, usage); if (left == null) { return false; } switch (condition.op) { case "gt": return left > condition.value; case "gte": return left >= condition.value; case "lt": return left < condition.value; case "lte": return left <= condition.value; case "eq": return left === condition.value; case "neq": return left !== condition.value; default: return false; } } getConditionFieldValue(field, usage) { switch (field) { case "total_input_tokens": return typeof usage.inputTokens === "number" ? usage.inputTokens : null; default: return null; } } }; var PricingModel = class { id; provider; model; schema; currency; tiers; constructor(args) { this.id = args.id; this.provider = args.provider; this.model = args.model; this.schema = args.schema; this.currency = args.currency; this.tiers = args.tiers; } getPricingTierForUsage(usage) { for (const tier of this.tiers) { if (tier.when && tier.when.length > 0 && tier.matchesUsage(usage)) { return tier; } } return this.getBasePricingTier(); } getBasePricingTier() { return this.tiers[0] ?? null; } }; // src/metrics/pricing-registry.ts var DATA_FILE_NAME = "pricing-data.jsonl"; var MINIFIED_METER_TO_CANONICAL = { it: "input_tokens", ot: "output_tokens", icrt: "input_cache_read_tokens", icwt: "input_cache_write_tokens", iat: "input_audio_tokens", oat: "output_audio_tokens", ort: "output_reasoning_tokens" }; var MINIFIED_CONDITION_FIELD_TO_CANONICAL = { tit: "total_input_tokens" }; var cachedLoadError = null; var PricingRegistry = class _PricingRegistry { constructor(pricingModels) { this.pricingModels = pricingModels; } pricingModels; static globalRegistry = null; static fromText(pricingModelText) { return new _PricingRegistry(parsePricingModelText(pricingModelText)); } static getGlobal() { if (_PricingRegistry.globalRegistry) { return _PricingRegistry.globalRegistry; } const pricingModels = loadPricingModels(); if (!pricingModels) { return null; } _PricingRegistry.globalRegistry = new _PricingRegistry(pricingModels); return _PricingRegistry.globalRegistry; } get(args) { const variants = getModelVariants(args.model); for (const variant of variants) { const key = makePricingKey({ provider: args.provider, model: variant }); const match = this.pricingModels.get(key); if (match) return match; } return null; } }; function loadPricingModels() { if (cachedLoadError) { return null; } try { const content = fs__default.default.readFileSync(resolvePricingModelPath(), "utf-8"); return parsePricingModelText(content); } catch (error) { cachedLoadError = error instanceof Error ? error.message : String(error); return null; } } function parsePricingModelText(content) { const pricingModels = /* @__PURE__ */ new Map(); for (const line of content.split("\n")) { const trimmed = line.trim(); if (!trimmed) { continue; } const parsed = JSON.parse(trimmed); const pricingModel = expandPricingModelRow(parsed); pricingModels.set(makePricingKey(pricingModel), pricingModel); } return pricingModels; } function expandPricingModelRow(row) { return new PricingModel({ id: row.i, provider: row.p, model: row.m, schema: row.s.v, currency: row.s.d.u, tiers: row.s.d.t.map( (tier, index) => new PricingTier({ index, when: tier.w?.map((condition) => ({ field: MINIFIED_CONDITION_FIELD_TO_CANONICAL[condition.f], op: condition.op, value: condition.value })), rates: Object.fromEntries( Object.entries(tier.r).map(([meter, value]) => [ MINIFIED_METER_TO_CANONICAL[meter], value.c ]) ) }) ) }); } function resolvePricingModelPath() { const packageRoot = getPackageRoot(); const candidates = [ path__default.default.join(packageRoot, "dist", "metrics", DATA_FILE_NAME), path__default.default.join(packageRoot, "src", "metrics", DATA_FILE_NAME), path__default.default.join(process.cwd(), "observability", "mastra", "src", "metrics", DATA_FILE_NAME), path__default.default.join(process.cwd(), "src", "metrics", DATA_FILE_NAME) ]; for (const candidate of candidates) { if (fs__default.default.existsSync(candidate)) { return candidate; } } throw new Error(`Unable to locate pricing data JSONL at any known path: ${candidates.join(", ")}`); } function getPackageRoot() { try { const require2 = module$1.createRequire((typeof document === 'undefined' ? require('u' + 'rl').pathToFileURL(__filename).href : (_documentCurrentScript && _documentCurrentScript.tagName.toUpperCase() === 'SCRIPT' && _documentCurrentScript.src || new URL('index.cjs', document.baseURI).href)) || "file://"); const packageJsonPath = require2.resolve("@mastra/observability/package.json"); return path__default.default.dirname(packageJsonPath); } catch { return process.cwd(); } } function makePricingKey(args) { return `${normalizeProvider(args.provider)}::${normalizeKeyPart(args.model)}`; } function normalizeKeyPart(value) { return value.trim().toLowerCase(); } function normalizeProvider(provider) { const normalized = provider.trim().toLowerCase(); const dotIndex = normalized.indexOf("."); return dotIndex !== -1 ? normalized.substring(0, dotIndex) : normalized; } function getModelVariants(model) { const variants = /* @__PURE__ */ new Set(); const add = (v) => { variants.add(v); variants.add(stripDateSuffix(v)); }; add(model); add(model.replace(/\./g, "-")); add(model.replace(/[./]/g, "-")); const slashIndex = model.indexOf("/"); if (slashIndex !== -1) { const withoutVendor = model.substring(slashIndex + 1); add(withoutVendor); add(withoutVendor.replace(/\./g, "-")); } return [...variants]; } function stripDateSuffix(model) { let stripped = model.replace(/-20\d{2}-\d{2}-\d{2}$/, ""); if (stripped !== model) return stripped; stripped = model.replace(/-20\d{6}(-[a-z]+)?$/, "$1"); if (stripped !== model) return stripped; stripped = model.replace(/-\d{2}-20\d{2}$/, ""); return stripped; } // src/metrics/types.ts var PricingMeter = { INPUT_TOKENS: "input_tokens", INPUT_AUDIO_TOKENS: "input_audio_tokens", INPUT_CACHE_READ_TOKENS: "input_cache_read_tokens", INPUT_CACHE_WRITE_TOKENS: "input_cache_write_tokens", INPUT_IMAGE_TOKENS: "input_image_tokens", OUTPUT_TOKENS: "output_tokens", OUTPUT_AUDIO_TOKENS: "output_audio_tokens", OUTPUT_IMAGE_TOKENS: "output_image_tokens", OUTPUT_REASONING_TOKENS: "output_reasoning_tokens" }; var TokenMetrics = { TOTAL_INPUT: "mastra_model_total_input_tokens", TOTAL_OUTPUT: "mastra_model_total_output_tokens", INPUT_TEXT: "mastra_model_input_text_tokens", INPUT_CACHE_READ: "mastra_model_input_cache_read_tokens", INPUT_CACHE_WRITE: "mastra_model_input_cache_write_tokens", INPUT_AUDIO: "mastra_model_input_audio_tokens", INPUT_IMAGE: "mastra_model_input_image_tokens", OUTPUT_TEXT: "mastra_model_output_text_tokens", OUTPUT_REASONING: "mastra_model_output_reasoning_tokens", OUTPUT_AUDIO: "mastra_model_output_audio_tokens", OUTPUT_IMAGE: "mastra_model_output_image_tokens" }; // src/metrics/usage-metrics.ts function getTokenMetricSamples(usage) { const samples = []; const pushIfDefined = (name, value) => { if (value != null) { samples.push({ name, value }); } }; const pushIfPositive = (name, value) => { if (value != null && value > 0) { samples.push({ name, value }); } }; pushIfDefined(TokenMetrics.TOTAL_INPUT, usage.inputTokens); pushIfDefined(TokenMetrics.TOTAL_OUTPUT, usage.outputTokens); if (usage.inputDetails) { pushIfPositive(TokenMetrics.INPUT_TEXT, usage.inputDetails.text); pushIfPositive(TokenMetrics.INPUT_CACHE_READ, usage.inputDetails.cacheRead); pushIfPositive(TokenMetrics.INPUT_CACHE_WRITE, usage.inputDetails.cacheWrite); pushIfPositive(TokenMetrics.INPUT_AUDIO, usage.inputDetails.audio); pushIfPositive(TokenMetrics.INPUT_IMAGE, usage.inputDetails.image); } if (usage.outputDetails) { pushIfPositive(TokenMetrics.OUTPUT_TEXT, usage.outputDetails.text); pushIfPositive(TokenMetrics.OUTPUT_REASONING, usage.outputDetails.reasoning); pushIfPositive(TokenMetrics.OUTPUT_AUDIO, usage.outputDetails.audio); pushIfPositive(TokenMetrics.OUTPUT_IMAGE, usage.outputDetails.image); } return samples; } // src/metrics/estimator.ts function estimateCosts(args, pricingRegistry = PricingRegistry.getGlobal()) { const { provider, model, usage } = args; const results = /* @__PURE__ */ new Map(); const pricingModel = pricingRegistry?.get({ provider, model }); if (!pricingModel) { const errorContext = { costMetadata: { error: "no_matching_model" }, provider, model }; applyErrorContextForUsage(results, usage, errorContext); return results; } const costMetadata = { pricing_id: pricingModel.id }; const pricingTier = pricingModel.getPricingTierForUsage(usage); if (!pricingTier) { const errorContext = { costMetadata: { ...costMetadata, error: "no_matching_tier" }, provider, model }; applyErrorContextForUsage(results, usage, errorContext); return results; } costMetadata["tier_index"] = pricingTier.index; const estimateFields = { pricingModel, pricingTier, costMetadata }; const inputDetailResults = []; if (usage.inputDetails?.audio) { const result = estimateCostForMeter({ meter: PricingMeter.INPUT_AUDIO_TOKENS, tokenCount: usage.inputDetails.audio, ...estimateFields }); results.set(TokenMetrics.INPUT_AUDIO, result.costContext); inputDetailResults.push(result); } if (usage.inputDetails?.cacheRead) { const result = estimateCostForMeter({ meter: PricingMeter.INPUT_CACHE_READ_TOKENS, tokenCount: usage.inputDetails.cacheRead, ...estimateFields }); results.set(TokenMetrics.INPUT_CACHE_READ, result.costContext); inputDetailResults.push(result); } if (usage.inputDetails?.cacheWrite) { const result = estimateCostForMeter({ meter: PricingMeter.INPUT_CACHE_WRITE_TOKENS, tokenCount: usage.inputDetails.cacheWrite, ...estimateFields }); results.set(TokenMetrics.INPUT_CACHE_WRITE, result.costContext); inputDetailResults.push(result); } if (usage.inputDetails?.image) { const result = estimateCostForMeter({ meter: PricingMeter.INPUT_IMAGE_TOKENS, tokenCount: usage.inputDetails.image, ...estimateFields }); results.set(TokenMetrics.INPUT_IMAGE, result.costContext); inputDetailResults.push(result); } if (usage.inputDetails?.text) { const result = estimateCostForMeter({ meter: PricingMeter.INPUT_TOKENS, tokenCount: usage.inputDetails.text, ...estimateFields }); results.set(TokenMetrics.INPUT_TEXT, result.costContext); inputDetailResults.push(result); } setAggregateCostContext({ results, totalMetric: TokenMetrics.TOTAL_INPUT, fallbackMeter: PricingMeter.INPUT_TOKENS, totalTokenCount: usage.inputTokens, detailResults: inputDetailResults, ...estimateFields }); const outputDetailResults = []; if (usage.outputDetails?.audio) { const result = estimateCostForMeter({ meter: PricingMeter.OUTPUT_AUDIO_TOKENS, tokenCount: usage.outputDetails.audio, ...estimateFields }); results.set(TokenMetrics.OUTPUT_AUDIO, result.costContext); outputDetailResults.push(result); } if (usage.outputDetails?.image) { const result = estimateCostForMeter({ meter: PricingMeter.OUTPUT_IMAGE_TOKENS, tokenCount: usage.outputDetails.image, ...estimateFields }); results.set(TokenMetrics.OUTPUT_IMAGE, result.costContext); outputDetailResults.push(result); } if (usage.outputDetails?.reasoning) { const result = estimateCostForMeter({ meter: PricingMeter.OUTPUT_REASONING_TOKENS, tokenCount: usage.outputDetails.reasoning, ...estimateFields }); results.set(TokenMetrics.OUTPUT_REASONING, result.costContext); outputDetailResults.push(result); } if (usage.outputDetails?.text) { const result = estimateCostForMeter({ meter: PricingMeter.OUTPUT_TOKENS, tokenCount: usage.outputDetails.text, ...estimateFields }); results.set(TokenMetrics.OUTPUT_TEXT, result.costContext); outputDetailResults.push(result); } setAggregateCostContext({ results, totalMetric: TokenMetrics.TOTAL_OUTPUT, fallbackMeter: PricingMeter.OUTPUT_TOKENS, totalTokenCount: usage.outputTokens, detailResults: outputDetailResults, ...estimateFields }); return results; } function applyErrorContextForUsage(results, usage, errorContext) { for (const sample of getTokenMetricSamples(usage)) { results.set(sample.name, errorContext); } } function setAggregateCostContext(args) { const { results, totalMetric, fallbackMeter, totalTokenCount, detailResults, pricingModel, pricingTier, costMetadata } = args; if (totalTokenCount == null) { return; } const successfulDetailCosts = detailResults.filter((result) => result.success).map((result) => result.costContext.estimatedCost).filter((value) => typeof value === "number"); if (successfulDetailCosts.length > 0) { const hasFailedDetailCost = detailResults.some((result) => !result.success); results.set(totalMetric, { provider: pricingModel.provider, model: pricingModel.model, estimatedCost: successfulDetailCosts.reduce((sum, value) => sum + value, 0), costUnit: pricingModel.currency, costMetadata: hasFailedDetailCost ? { ...costMetadata, error: "partial_cost" } : { ...costMetadata } }); return; } const fallbackResult = estimateCostForMeter({ meter: fallbackMeter, tokenCount: totalTokenCount, pricingModel, pricingTier, costMetadata }); results.set(totalMetric, fallbackResult.costContext); } function estimateCostForMeter(args) { const { pricingModel, pricingTier, meter, tokenCount, costMetadata } = args; const costContext = { provider: pricingModel.provider, model: pricingModel.model }; const pricePerUnit = pricingTier.rates[meter]; if (typeof pricePerUnit !== "number") { return { success: false, costContext: { ...costContext, costMetadata: { ...costMetadata, error: "no_pricing_for_usage_type" } } }; } return { success: true, costContext: { ...costContext, estimatedCost: tokenCount * pricePerUnit, costUnit: pricingModel.currency, costMetadata: { ...costMetadata } } }; } // src/metrics/auto-extract.ts function emitDurationMetrics(span, metrics) { const durationMetricName = getDurationMetricName(span); if (!durationMetricName || !span.startTime || !span.endTime) { return; } const durationMs = span.endTime.getTime() - span.startTime.getTime(); metrics.emit(durationMetricName, durationMs, { status: span.errorInfo ? "error" : "ok" }); } function emitTokenMetrics(span, metrics) { if (span.type !== observability.SpanType.MODEL_GENERATION) { return; } const attrs = span.attributes; if (!attrs?.usage) { return; } emitUsageMetrics(attrs, attrs.usage, metrics); } function emitTokenMetricsForUsage(usage, provider, model, metrics) { emitUsageMetrics({ provider, model }, usage, metrics); } function emitAutoExtractedMetrics(span, metrics) { emitDurationMetrics(span, metrics); emitTokenMetrics(span, metrics); } function emitUsageMetrics(attrs, usage, metrics) { let metricCosts = /* @__PURE__ */ new Map(); const providedCostContext = getProvidedCostContext(attrs, usage); if (providedCostContext) { metricCosts = providedCostContext; } else { try { const provider = attrs.provider; const model = attrs.responseModel ?? attrs.model; if (provider && model) { metricCosts = estimateCosts({ provider, model, usage }); } } catch { metricCosts = /* @__PURE__ */ new Map(); } } const emit = (name, value) => { const costContext = metricCosts.get(name); if (!costContext) { metrics.emit(name, value); return; } metrics.emit(name, value, void 0, { costContext }); }; for (const sample of getTokenMetricSamples(usage)) { emit(sample.name, sample.value); } } function getProvidedCostContext(attrs, usage) { const costContext = attrs.costContext; if (typeof costContext?.estimatedCost !== "number") { return void 0; } const carrierMetric = usage.inputTokens !== void 0 ? TokenMetrics.TOTAL_INPUT : TokenMetrics.TOTAL_OUTPUT; const provider = costContext.provider ?? attrs.provider; const model = costContext.model ?? attrs.responseModel ?? attrs.model; const contexts = /* @__PURE__ */ new Map(); for (const sample of getTokenMetricSamples(usage)) { contexts.set(sample.name, { provider, model }); } contexts.set(carrierMetric, { ...costContext, provider, model, costMetadata: { ...costContext.costMetadata, allocation: "query_total" } }); return contexts; } function getDurationMetricName(span) { switch (span.type) { case observability.SpanType.AGENT_RUN: return "mastra_agent_duration_ms"; case observability.SpanType.TOOL_CALL: case observability.SpanType.MCP_TOOL_CALL: return "mastra_tool_duration_ms"; case observability.SpanType.CLIENT_TOOL_CALL: return null; case observability.SpanType.WORKFLOW_RUN: return "mastra_workflow_duration_ms"; case observability.SpanType.MODEL_GENERATION: return "mastra_model_duration_ms"; case observability.SpanType.PROCESSOR_RUN: return "mastra_processor_duration_ms"; default: return null; } } var UUID_REGEX = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; var CardinalityFilter = class { blockedLabels; blockUUIDs; /** * @param config - Optional configuration. When omitted, uses * {@link DEFAULT_BLOCKED_LABELS} and blocks UUID-valued labels. */ constructor(config) { const blocked = config?.blockedLabels ?? [...observability.DEFAULT_BLOCKED_LABELS]; this.blockedLabels = new Set(blocked.map((l) => l.toLowerCase())); this.blockUUIDs = config?.blockUUIDs ?? true; } /** * Return a copy of `labels` with blocked keys and UUID values removed. * * @param labels - Raw metric labels to filter. * @returns A new object containing only the allowed labels. */ filterLabels(labels) { const filtered = {}; for (const [key, value] of Object.entries(labels)) { if (this.blockedLabels.has(key.toLowerCase())) { continue; } if (this.blockUUIDs && UUID_REGEX.test(value)) { continue; } filtered[key] = value; } return filtered; } }; // src/usage.ts function isV3RawUsage(raw) { return typeof raw === "object" && raw !== null && "inputTokens" in raw; } function isDefined(value) { return value != null; } function extractUsageMetrics(usage, providerMetadata) { if (!usage) { return {}; } const inputDetails = {}; const outputDetails = {}; let inputTokens = usage.inputTokens; const outputTokens = usage.outputTokens; const aiSdkDetails = usage.inputTokenDetails; if (isDefined(aiSdkDetails?.cacheReadTokens)) { inputDetails.cacheRead = aiSdkDetails.cacheReadTokens; } if (isDefined(aiSdkDetails?.cacheWriteTokens)) { inputDetails.cacheWrite = aiSdkDetails.cacheWriteTokens; } if (!isDefined(inputDetails.cacheRead) && isDefined(usage.cachedInputTokens)) { inputDetails.cacheRead = usage.cachedInputTokens; } if (!isDefined(inputDetails.cacheWrite) && isDefined(usage.cacheCreationInputTokens)) { inputDetails.cacheWrite = usage.cacheCreationInputTokens; } if (isDefined(usage.reasoningTokens)) { outputDetails.reasoning = usage.reasoningTokens; } const anthropic = providerMetadata?.anthropic; if (anthropic) { const rawV3InputUsage = isV3RawUsage(usage.raw) ? usage.raw.inputTokens : void 0; const hasV3CachedTotals = rawV3InputUsage?.total !== void 0 && (rawV3InputUsage.cacheRead !== void 0 || rawV3InputUsage.cacheWrite !== void 0); if (!isDefined(inputDetails.cacheRead) && isDefined(anthropic.cacheReadInputTokens)) { inputDetails.cacheRead = anthropic.cacheReadInputTokens; } if (!isDefined(inputDetails.cacheWrite) && isDefined(anthropic.cacheCreationInputTokens)) { inputDetails.cacheWrite = anthropic.cacheCreationInputTokens; } const inputAlreadyIncludesCache = hasV3CachedTotals || isDefined(usage.cachedInputTokens) && usage.cachedInputTokens > 0 || isDefined(usage.cacheCreationInputTokens) && usage.cacheCreationInputTokens > 0; if (!inputAlreadyIncludesCache && (isDefined(inputDetails.cacheRead) || isDefined(inputDetails.cacheWrite))) { inputTokens = (usage.inputTokens ?? 0) + (inputDetails.cacheRead ?? 0) + (inputDetails.cacheWrite ?? 0); } } const google = providerMetadata?.google; if (google?.usageMetadata) { if (!isDefined(inputDetails.cacheRead) && isDefined(google.usageMetadata.cachedContentTokenCount)) { inputDetails.cacheRead = google.usageMetadata.cachedContentTokenCount; } if (isDefined(google.usageMetadata.thoughtsTokenCount)) { outputDetails.reasoning = google.usageMetadata.thoughtsTokenCount; } } if (isDefined(inputTokens)) { inputDetails.text = Math.max( 0, inputTokens - sumDefinedValues(inputDetails, ["cacheRead", "cacheWrite", "audio", "image"]) ); } if (isDefined(outputTokens)) { outputDetails.text = Math.max(0, outputTokens - sumDefinedValues(outputDetails, ["reasoning", "audio", "image"])); } const result = { inputTokens, outputTokens }; if (Object.keys(inputDetails).length > 0) { result.inputDetails = inputDetails; } if (Object.keys(outputDetails).length > 0) { result.outputDetails = outputDetails; } return result; } function sumDefinedValues(obj, keys) { return keys.reduce((sum, key) => sum + (obj[key] ?? 0), 0); } function addOptional(a, b) { if (a === void 0 && b === void 0) return void 0; return (a ?? 0) + (b ?? 0); } function mergeInputDetails(a, b) { if (!a) return b ? { ...b } : void 0; if (!b) return { ...a }; return { text: addOptional(a.text, b.text), cacheRead: addOptional(a.cacheRead, b.cacheRead), cacheWrite: addOptional(a.cacheWrite, b.cacheWrite), audio: addOptional(a.audio, b.audio), image: addOptional(a.image, b.image) }; } function mergeOutputDetails(a, b) { if (!a) return b ? { ...b } : void 0; if (!b) return { ...a }; return { text: addOptional(a.text, b.text), reasoning: addOptional(a.reasoning, b.reasoning), audio: addOptional(a.audio, b.audio), image: addOptional(a.image, b.image) }; } function addUsageStats(a, b) { if (!a) { return { ...b, inputDetails: b.inputDetails ? { ...b.inputDetails } : void 0, outputDetails: b.outputDetails ? { ...b.outputDetails } : void 0 }; } return { inputTokens: addOptional(a.inputTokens, b.inputTokens), outputTokens: addOptional(a.outputTokens, b.outputTokens), inputDetails: mergeInputDetails(a.inputDetails, b.inputDetails), outputDetails: mergeOutputDetails(a.outputDetails, b.outputDetails) }; } // src/model-tracing.ts function supportsModelInference() { return features.coreFeatures.has("model-inference-span"); } function formatPreviewLabel(label, fallback) { return typeof label === "string" && label.length > 0 ? label : fallback; } function summarizePart(part) { if (typeof part === "string") { return part; } if (!part || typeof part !== "object") { return ""; } if ("text" in part && typeof part.text === "string") { return part.text; } if ("parts" in part && Array.isArray(part.parts)) { return part.parts.map(summarizePart).filter(Boolean).join(""); } if ("inlineData" in part && part.inlineData && typeof part.inlineData === "object") { return `[${formatPreviewLabel(part.inlineData.mimeType, "binary")}]`; } if ("image_url" in part) { return "[image]"; } if ("functionCall" in part && part.functionCall && typeof part.functionCall === "object") { return `[tool: ${formatPreviewLabel(part.functionCall.name, "unknown")}]`; } if ("function_call" in part && part.function_call && typeof part.function_call === "object") { return `[tool: ${formatPreviewLabel(part.function_call.name, "unknown")}]`; } if ("function" in part && part.function && typeof part.function === "object") { return `[tool: ${formatPreviewLabel(part.function.name, "unknown")}]`; } if ("toolName" in part) { return `[tool: ${formatPreviewLabel(part.toolName, "unknown")}]`; } if ("type" in part && typeof part.type === "string") { switch (part.type) { case "image": return "[image]"; case "file": return "[file]"; case "reasoning": return "[reasoning]"; case "tool-call": return `[tool: ${formatPreviewLabel(part.toolName, "unknown")}]`; case "tool-result": return "[tool-result]"; default: return `[${part.type}]`; } } if ("content" in part && typeof part.content === "string") { return part.content; } return "[object]"; } function summarizeMessageContent(content) { if (typeof content === "string") { return content; } if (Array.isArray(content)) { return content.map(summarizePart).filter(Boolean).join(""); } if (content && typeof content === "object") { if ("parts" in content && Array.isArray(content.parts)) { return content.parts.map(summarizePart).filter(Boolean).join(""); } return summarizePart(content); } if (content == null) { return ""; } return String(content); } function appendToolPreview(preview, toolCalls) { if (!Array.isArray(toolCalls) || toolCalls.length === 0) { return preview; } const toolPreview = toolCalls.map((toolCall) => summarizePart(toolCall)).filter(Boolean).join(" "); if (!toolPreview) { return preview; } return preview ? `${preview} ${toolPreview}` : toolPreview; } function appendPreview(preview, addition) { if (!addition) { return preview; } return preview ? `${preview} ${addition}` : addition; } function normalizeMessages(messages) { return messages.map((message) => { if (!message || typeof message !== "object") { return { role: "user", content: summarizeMessageContent(message) }; } const role = typeof message.role === "string" ? message.role : "user"; const baseContent = summarizeMessageContent(message.content); const contentWithToolArrays = appendToolPreview( appendToolPreview(baseContent, message.toolCalls), message.tool_calls ); const functionCall = message.functionCall; const functionCallPreview = functionCall === void 0 ? "" : summarizePart({ functionCall }); const functionCallSnakeCase = message.function_call; const functionCallSnakeCasePreview = functionCallSnakeCase === void 0 ? "" : summarizePart({ function_call: functionCallSnakeCase }); const contentWithFunctionCall = appendPreview( appendPreview(contentWithToolArrays, functionCallPreview), functionCallSnakeCasePreview ); return { role, content: contentWithFunctionCall }; }); } function summarizeRequestBody(body) { if (body == null) { return void 0; } if (typeof body !== "object") { return typeof body === "string" ? body : String(body); } if (Array.isArray(body.messages)) { return normalizeMessages(body.messages); } if (Array.isArray(body.input)) { return normalizeMessages(body.input); } if (Array.isArray(body.contents)) { return body.contents.map((item) => ({ role: typeof item?.role === "string" ? item.role : "user", content: Array.isArray(item?.parts) ? item.parts.map(summarizePart).filter(Boolean).join("") : "" })); } const summary = {}; if (typeof body.model === "string") { summary.model = body.model; } const bodyKeys = Object.keys(body).filter((key) => key !== "body"); if (bodyKeys.length > 0) { summary.keys = bodyKeys; } return Object.keys(summary).length > 0 ? summary : "[request body]"; } function extractStepInput(payload) { if (Array.isArray(payload?.inputMessages)) { return normalizeMessages(payload.inputMessages); } const request = payload?.request; if (!request) return void 0; const { body } = request; if (body == null) return request; try { const parsed = typeof body === "string" ? JSON.parse(body) : body; return summarizeRequestBody(parsed); } catch { return request; } } var ModelSpanTracker = class { #modelSpan; #currentStepSpan; #currentInferenceSpan; #currentChunkSpan; #currentChunkType; #accumulator = {}; #stepIndex = 0; #chunkSequence = 0; #completionStartTime; #currentStepInputIsFinal = false; /** When true, step-finish chunks don't auto-close the step span (for durable execution) */ #deferStepClose = false; /** Stored step-finish payload when defer mode is enabled */ #pendingStepFinishPayload; /** Static request-side context applied to every MODEL_INFERENCE span */ #inferenceContext; constructor(modelSpan) { this.#modelSpan = modelSpan; } /** * Set request-side context applied to subsequent MODEL_INFERENCE spans. * No-op when paired with an older @mastra/core that lacks the feature flag. */ setInferenceContext(context) { this.#inferenceContext = context; } /** * Capture the completion start time (time to first token) when the first content chunk arrives. */ #captureCompletionStartTime() { if (this.#completionStartTime) { return; } this.#completionStartTime = /* @__PURE__ */ new Date(); } /** * Get the tracing context for creating child spans. * Returns the current step span if active, otherwise the model span. */ getTracingContext() { return { currentSpan: this.#currentStepSpan ?? this.#modelSpan }; } /** * Report an error on the generation span, also closing any still-open * MODEL_INFERENCE / MODEL_STEP children so a fatal error before step-finish * doesn't leave them dangling. No-op if they were already closed. */ reportGenerationError(options) { if (this.#currentInferenceSpan) { this.#currentInferenceSpan.error({ error: options.error, endSpan: true }); this.#currentInferenceSpan = void 0; } if (this.#currentStepSpan) { this.#currentStepSpan.error({ error: options.error, endSpan: true }); this.#currentStepSpan = void 0; } this.#modelSpan?.error(options); } /** * End the generation span with optional raw usage data. * If usage is provided, it will be converted to UsageStats with cache token details. */ endGeneration(options) { const { usage, providerMetadata, ...spanOptions } = options ?? {}; if (spanOptions.attributes) { spanOptions.attributes.completionStartTime = this.#completionStartTime; spanOptions.attributes.usage = extractUsageMetrics(usage, providerMetadata); } this.#modelSpan?.end(spanOptions); } /** * Update the generation span */ updateGeneration(options) { this.#modelSpan?.update(options); } /** * Enable or disable deferred step closing for durable execution. * When enabled, step-finish chunks won't automatically close the step span. * Use exportCurrentStep() to get the span data, then close it manually later. */ setDeferStepClose(defer) { this.#deferStepClose = defer; } /** * Export the current step span for later rebuilding (durable execution). * Returns undefined if no step span is active. */ exportCurrentStep() { return this.#currentStepSpan?.exportSpan(); } /** * Get the pending step finish payload (captured when defer mode is enabled). * This contains usage, finishReason, etc. for closing the step later. */ getPendingStepFinishPayload() { return this.#pendingStepFinishPayload; } /** * Set the starting step index for durable execution. * Used when resuming across agentic loop iterations to maintain step continuity. */ setStepIndex(index) { this.#stepIndex = index; } /** * Get the current step index. */ getStepIndex() { return this.#stepIndex; } /** * Start a new Model execution step. * This should be called at the beginning of LLM execution to capture accurate startTime. * The step-start chunk payload can be passed later via updateStep() if needed. * * Note: this only opens MODEL_STEP. The MODEL_INFERENCE child span is opened * separately via startInference() so its duration excludes input processor work. * Callers that don't call startInference() explicitly will get one auto-created * when the first model chunk arrives. */ startStep(payload) { if (this.#currentStepSpan) { return; } const input = extractStepInput(payload); this.#currentStepSpan = this.#modelSpan?.createChildSpan({ name: `step: ${this.#stepIndex}`, type: observability.SpanType.MODEL_STEP, attributes: { stepIndex: this.#stepIndex, ...payload?.messageId ? { messageId: payload.messageId } : {}, ...payload?.warnings?.length ? { warnings: payload.warnings } : {} }, input, tracingPolicy: this.#modelSpan?.tracingPolicy }); this.#currentStepInputIsFinal = Array.isArray(payload?.inputMessages); this.#chunkSequence = 0; } /** * End the current MODEL_INFERENCE span when the provider stream finishes. * Fields are duplicated onto MODEL_STEP (in #endStepSpan) so existing * integrations that read usage/finishReason from the step span continue * to work unchanged. * * Safe to call multiple times - no-ops if the span is already closed. */ #endInferenceSpan(payload) { if (!this.#currentInferenceSpan) return; const { usage: rawUsage, ...otherOutput } = payload.output; const usage = extractUsageMetrics(rawUsage, payload.metadata?.providerMetadata); this.#currentInferenceSpan.end({ output: otherOutput, attributes: { usage, finishReason: payload.stepResult.reason, warnings: payload.stepResult.warnings, completionStartTime: this.#completionStartTime } }); this.#currentInferenceSpan = void 0; } /** * Open the MODEL_INFERENCE span for the current step. Chunks (including tool-call * chunks emitted by the model) parent under this span so its duration reflects * pure model latency. * * Should be called immediately before invoking the model — after any input * processors / `prepareStep` work has completed — so the span's startTime * does not include processor time. The latest `#inferenceContext` (set via * setInferenceContext) is snapshotted onto the span at creation. * * No-ops when the installed @mastra/core lacks the `model-inference-span` * feature flag, or when called without an active step span. Auto-invoked from * chunk handlers as a safety net; explicit callers get the most accurate * start time. */ startInference(payload) { if (!supportsModelInference()) { return; } if (!this.#currentStepSpan || this.#currentInferenceSpan) { return; } const input = extractStepInput(payload); const generationAttrs = this.#modelSpan?.attributes; const ctx = this.#inferenceContext; this.#currentInferenceSpan = this.#currentStepSpan.createChildSpan({ name: `inference: ${this.#stepIndex}`, type: observability.SpanType.MODEL_INFERENCE, attributes: { stepIndex: this.#stepIndex, model: generationAttrs?.model, provider: generationAttrs?.provider, streaming: generationAttrs?.streaming, ...ctx?.parameters !== void 0 ? { parameters: ctx.parameters } : {}, ...ctx?.providerOptions !== void 0 ? { providerOptions: ctx.providerOptions } : {}, ...ctx?.availableTools !== void 0 ? { availableTools: ctx.availableTools } : {}, ...ctx?.toolChoice !== void 0 ? { toolChoice: ctx.toolChoice } : {}, ...ctx?.responseFormat !== void 0 ? { responseFormat: ctx.responseFormat } : {} }, input, tracingPolicy: this.#modelSpan?.tracingPolicy }); } /** * Update the current step span with additional payload data. * Called when step-start chunk arrives with request/warnings info. */ updateStep(payload) { if (!this.#currentStepSpan || !payload) { return; } const hasFinalInput = Array.isArray(payload.inputMessages); const input = hasFinalInput || !this.#currentStepInputIsFinal ? extractStepInput(payload) : void 0; this.#currentStepSpan.update({ ...input !== void 0 ? { input } : {}, attributes: { ...payload.messageId ? { messageId: payload.messageId } : {}, ...payload.warnings?.length ? { warnings: payload.warnings } : {} } }); if (hasFinalInput) { this.#currentStepInputIsFinal = true; } } /** * End the current Model execution step with token usage, finish reason, output, and metadata */ #endStepSpan(payload) { this.#endChunkSpan(); if (!this.#currentStepSpan) return; const output = payload.output; const { usage: rawUsage, ...otherOutput } = output; const stepResult = payload.stepResult; const metadata = payload.metadata; const usage = extractUsageMetrics(rawUsage, metadata?.providerMetadata); const cleanMetadata = metadata ? { ...metadata } : void 0; if (cleanMetadata) { for (const key of ["request", "id", "timestamp", "modelId", "modelVersion", "modelProvider"]) { delete cleanMetadata[key]; } } this.#endInferenceSpan(payload); this.#currentStepSpan.end({ output: otherOutput, attributes: { usage, isContinued: stepResult.isContinued, finishReason: stepResult.reason, warnings: stepResult.warnings }, metadata: { ...cleanMetadata } }); this.#currentStepSpan = void 0; this.#currentStepInputIsFinal = false; this.#stepIndex++; } /** * Returns the parent span for chunks. Chunks parent under MODEL_INFERENCE * (the provider call) when available, falling back to MODEL_STEP only if * startStep() was bypassed. */ #chunkParent() { return this.#currentInferenceSpan ?? this.#currentStepSpan; } /** * Safety-net invoked from chunk handlers: auto-create MODEL_STEP and * MODEL_INFERENCE if a chunk arrives before the loop has explicitly opened * them, so chunks parent under MODEL_INFERENCE rather than falling through * to MODEL_STEP. Idempotent — each public start* method is itself a no-op * when its span is already live. */ #ensureStepAndInference() { if (!this.#currentStepSpan) { this.startStep(); } if (!this.#currentInferenceSpan) { this.startInference(); } } /** * Create a new chunk span (for multi-part chunks like text-start/delta/end) */ #startChunkSpan(chunkType, initialData) { this.#endChunkSpan(); this.#ensureStepAndInference(); this.#currentChunkSpan = this.#chunkParent()?.createChildSpan({ name: `chunk: '${chunkType}'`, type: observability.SpanType.MODEL_CHUNK, attributes: { chunkType, sequenceNumber: this.#chunkSequence }, tracingPolicy: this.#modelSpan?.tracingPolicy }); this.#currentChunkType = chunkType; this.#accumulator = initialData || {}; } /** * Append string content to a specific field in the accumulator */ #appendToAccumulator(field, text) { if (this.#accumulator[field] === void 0) { this.#accumulator[field] = text; } else { this.#accumulator[field] += text; } } /** * End the current chunk span. * Safe to call multiple times - will no-op if span already ended. */ #endChunkSpan(output) { if (!this.#currentChunkSpan) return; this.#currentChunkSpan.end({ output: output !== void 0 ? output : this.#accumulator }); this.#currentChunkSpan = void 0; this.#currentChunkType = void 0; this.#accumulator = {}; this.#chunkSequence++; } /** * Create an event span (for single chunks like tool-call) */ #createEventSpan(chunkType, output, options) { this.#ensureStepAndInference(); const span = this.#chunkParent()?.createEventSpan({ name: `chunk: '${chunkType}'`, type: observability.SpanType.MODEL_CHUNK, attributes: { chunkType, sequenceNumber: this.#chunkSequence, ...options?.attributes }, metadata: options?.metadata, output, tracingPolicy: this.#modelSpan?.tracingPolicy }); if (span) { this.#chunkSequence++; } } /** * Check if there is currently an active chunk span */ #hasActiveChunkSpan() { return !!this.#currentChunkSpan; } /** * Get the current accumulator value */ #getAccumulator() { return this.#accumulator; } /** * Handle text chunk spans (text-start/delta/end) */ #handleTextChunk(chunk) { switch (chunk.type) { case "text-start": this.#startChunkSpan("text"); break; case "text-delta": if (this.#currentChunkType !== "text") { this.#startChunkSpan("text"); } this.#appendToAccumulator("text", chunk.payload.text); break; case "text-end": { this.#endChunkSpan(); break; } } } /** * Handle reasoning chunk spans (reasoning-start/delta/end) */ #handleReasoningChunk(chunk) { switch (chunk.type) { case "reasoning-start": this.#startChunkSpan("reasoning"); break; case "reasoning-delta": if (this.#currentChunkType !== "reasoning") { this.#startChunkSpan("reasoning"); } this.#appendToAccumulator("text", chunk.payload.text); break; case "reasoning-end": { this.#endChunkSpan(); break; } } } /** * Handle tool call chunk spans (tool-call-input-streaming-start/delta/end, tool-call) */ #handleToolCallChunk(chunk) { switch (chunk.type) { case "tool-call-input-streaming-start": this.#startChunkSpan("tool-call", { toolName: chunk.payload.toolName, toolCallId: chunk.payload.toolCallId }); break; case "tool-call-delta": this.#appendToAccumulator("toolInput", chunk.payload.argsTextDelta); break; case "tool-call-input-streaming-end": case "tool-call": { const acc = this.#getAccumulator(); let toolInput; try { toolInput = acc.toolInput ? JSON.parse(acc.toolInput) : {}; } catch { toolInput = acc.toolInput; } this.#endChunkSpan({ toolName: acc.toolName, toolCallId: acc.toolCallId, toolInput }); break; } } } /** * Handle object chunk spans (object, object-result) */ #handleObjectChunk(chunk) { switch (chunk.type) { case "object": if (this.#currentChunkType !== "object") { this.#startChunkSpan("object"); } break; case "object-result": this.#endChunkSpan(chunk.object); break; } } /** * Handle tool-call-approval chunks. * Creates a span for approval requests so they can be seen in traces for debugging. */ #handleToolApprovalChunk(chunk) { if (chunk.type !== "tool-call-approval") return; const payload = chunk.payload; this.#ensureStepAndInference(); const span = this.#chunkParent()?.createEventSpan({ name: `chunk: 'tool-call-approval'`, type: observability.SpanType.MODEL_CHUNK, attributes: { chunkType: "tool-call-approval", sequenceNumber: this.#chunkSequence }, output: payload, tracingPolicy: this.#modelSpan?.tracingPolicy }); if (span) { this.#chunkSequence++; } } /** * Wraps a stream with model tracing transform to track MODEL_STEP and MODEL_CHUNK spans. * * This should be added to the stream pipeline to automatically * create MODEL_STEP and MODEL_CHUNK spans for each semantic unit in the stream. */ wrapStream(stream) { return stream.pipeThrough( new web.TransformStream({ transform: (chunk, controller) => { switch (chunk.type) { case "text-delta": case "tool-call-delta": case "reasoning-delta": this.#captureCompletionStartTime(); break; } controller.enqueue(chunk); switch (chunk.type) { case "text-start": case "text-delta": case "text-end": this.#handleTextChunk(chunk); break; case "tool-call-input-streaming-start": case "tool-call-delta": case "tool-call-input-streaming-end": case "tool-call": this.#handleToolCallChunk(chunk); break; case "reasoning-start": case "reasoning-delta": case "reasoning-end": this.#handleReasoningChunk(chunk); break; case "object": case "object-result": this.#handleObjectChunk(chunk); break; case "step-start": if (this.#currentStepSpan) { this.updateStep(chunk.payload); } else { this.startStep(chunk.payload); } if (!this.#currentInferenceSpan) { this.startInference(chunk.payload); } break; case "step-finish": if (this.#deferStepClose) { this.#pendingStepFinishPayload = chunk.payload; this.#endChunkSpan(); this.#endInferenceSpan(chunk.payload); } else { this.#endStepSpan(chunk.payload); } break; // Infrastructure chunks - skip creating spans for these // They are either redundant, metadata-only, or error/control flow case "raw": // Redundant raw data case "start": // Stream start marker case "finish": // Stream finish marker (step-finish already captures this) case "response-metadata": // Response metadata (not semantic content) case "source": // Source references (metadata) case "file": // Binary file data (too large/not semantic) case "error": // Error handling case "abort": // Abort signal case "tripwire": // Processor rejection case "watch": // Internal watch event case "tool-error": // Tool error handling case "tool-call-suspended": // Suspension (not content) case "reasoning-signature": // Signature metadata case "redacted-reasoning": // Redacted content metadata case "step-output": break; case "tool-call-approval": this.#handleToolApprovalChunk(chunk); break; case "tool-output": break; case "tool-result": { const { // Metadata - tool call context (unique to tool-result chunks) toolCallId, toolName, isError, dynamic, providerExecuted, providerMetadata, // Keep provider-executed results on MODEL_CHUNK because they come // from the model/provider stream and may not have a sibling TOOL_CALL span. // For locally executed tools, the canonical payload lives on TOOL_CALL. result} = chunk.payload || {}; const metadata = { toolCallId, toolName }; if (isError !== void 0) metadata.isError = isError; if (dynamic !== void 0) metadata.dynamic = dynamic; if (providerExecuted !== void 0) metadata.providerExecuted = providerExecuted; if (providerMetadata !== void 0) metadata.providerMetadata = providerMetadata; this.#createEventSpan(chunk.type, providerExecuted ? result : void 0, { metadata }); break; } } } }) ); } }; // src/spans/base.ts function isSpanInternal(spanType, flags) { if (flags === void 0 || flags === observability.InternalSpans.NONE) { return false; } switch (spanType) { // Workflow-related spans case observability.SpanType.WORKFLOW_RUN: case observability.SpanType.WORKFLOW_STEP: case observability.SpanType.WORKFLOW_CONDITIONAL: case observability.SpanType.WORKFLOW_CONDITIONAL_EVAL: case observability.SpanType.WORKFLOW_PARALLEL: case observability.SpanType.WORKFLOW_LOOP: case observability.SpanType.WORKFLOW_SLEEP: case observability.SpanType.WORKFLOW_WAIT_EVENT: return (flags & observability.InternalSpans.WORKFLOW) !== 0; // Agent-related spans case observability.SpanType.AGENT_RUN: return (flags & observability.InternalSpans.AGENT) !== 0; // Tool-related spans case observability.SpanType.TOOL_CALL: case observability.SpanType.MCP_TOOL_CALL: return (flags & observability.InternalSpans.TOOL) !== 0; // Model-related spans case observability.SpanType.MODEL_GENERATION: case observability.SpanType.MODEL_STEP: case observability.SpanType.MODEL_INFERENCE: case observability.SpanType.MODEL_CHUNK: return (flags & observability.InternalSpans.MODEL) !== 0; // Default: never internal default: return false; } } function getExternalParentId(options) { if (!options.parent) { return void 0; } if (options.parent.isInternal) { return options.parent.getParentSpanId(false); } else { return options.parent.id; } } var BaseSpan = class { name; type; attributes; parent; startTime; endTime; isEvent; isInternal; tracingPolicy; observabilityInstance; input; output; errorInfo; metadata; requestContext; tags; traceState; /** Entity type that created the span (e.g., agent, workflow) */ entityType; /** Entity ID that created the span */ entityId; /** Entity name that created the span */ entityName; /** Parent span ID (for root spans that are children of external spans) */ parentSpanId; /** Deep clean options for serialization */ deepCleanOptions; /** * Whether this span is filtered out before export. When true, BaseSpan/ * DefaultSpan skip attaching attributes/input/output/errorInfo/requestContext * entirely -- they are never read on excluded spans, and skipping avoids * both the deepClean cost and holding references to large payloads for * the lifetime of the span. Set when excludeSpanTypes drops the type, * when the span is internal and includeInternalSpans is false, or when * the subclass is always excluded (e.g., NoOpSpan). * * Note: metadata is still attached and deepCleaned because it is read in * process by getCorrelationContext() and by getLoggerContext() / * getMetricsContext() (which structuredClone it). */ isExcluded; /** Cached canonical correlation context for this live span */ correlationContext; /** * Subclasses can override to unconditionally mark the span as excluded. * NoOpSpan uses this because it is never exported regardless of config. */ get alwaysExcluded() { return false; } constructor(options, observabilityInstance) { const observabilityConfig = observabilityInstance.getConfig(); this.deepCleanOptions = mergeSerializationOptions(observabilityConfig.serializationOptions); this.name = options.name; this.type = options.type; this.isInternal = isSpanInternal(this.type, options.tracingPolicy?.internal); this.isExcluded = this.alwaysExcluded || observabilityConfig.excludeSpanTypes?.includes(this.type) === true || this.isInternal && !observabilityConfig.includeInternalSpans; this.metadata = deepClean( options.parent?.metadata || options.metadata ? { ...options.parent?.metadata, ...options.metadata } : void 0, this.deepCleanOptions ); if (options.requestContext && options.requestContext.size() > 0) { this.requestContext = deepClean(options.requestContext.all, this.deepCleanOptions); } this.parent = options.parent; this.startTime = options.startTime ?? /* @__PURE__ */ new Date(); this.observabilityInstance = observabilityInstance; this.isEvent = options.isEvent ?? false; this.tracingPolicy = options.tracingPolicy; this.traceState = options.traceState; this.tags = !options.parent && options.tags?.length ? options.tags : void 0; const entityParent = this.getParentSpan(false); this.entityType = options.entityType ?? entityParent?.entityType; this.entityId = options.entityId ?? entityParent?.entityId; this.entityName = options.entityName ?? entityParent?.entityName; if (this.isExcluded) { this.attributes = {}; return; } this.attributes = deepClean(options.attributes, this.deepCleanOptions) || {}; if (options.requestContext && options.requestContext.size() > 0) { this.requestContext = deepClean(options.requestContext.all, this.deepCleanOptions); } if (this.isEvent) { this.output = deepClean(options.output, this.deepCleanOptions); } else { this.input = deepClean(options.input, this.deepCleanOptions); } } createChildSpan(options) { return this.observabilityInstance.startSpan({ ...options, parent: this, isEvent: false }); } createEventSpan(options) { return this.observabilityInstance.startSpan({ ...options, parent: this, isEvent: true }); } /** * Create a ModelSpanTracker for this span (only works if this is a MODEL_GENERATION span) * Returns undefined for non-MODEL_GENERATION spans */ createTracker() { if (this.type !== observability.SpanType.MODEL_GENERATION) { return void 0; } return new ModelSpanTracker(this); } /** Returns `TRUE` if the span is the root span of a trace */ get isRootSpan() { return !this.parent; } /** Get the closest parent span, optionally skipping internal spans */ getParentSpan(includeInternalSpans) { if (!this.parent) { return void 0; } if (includeInternalSpans) return this.parent; if (this.parent.isInternal) return this.parent.getParentSpan(includeInternalSpans); return this.parent; } /** Get the closest parent spanId that isn't an internal span */ getParentSpanId(includeInternalSpans) { if (!this.parent) { return this.parentSpanId; } const parentSpan = this.getParentSpan(includeInternalSpans); if (parentSpan) { return parentSpan.id; } return this.parent.getParentSpanId(includeInternalSpans); } /** Find the closest parent span of a specific type by walking up the parent chain */ findParent(spanType) { let current = this.parent; while (current) { if (current.type === spanType) { return current; } current = current.parent; } return void 0; } /** Build and cache the canonical correlation context for this live span. */ getCorrelationContext() { if (this.correlationContext) { return this.correlationContext; } const metadata = this.metadata ?? {}; const getMetadataString = (key) => typeof metadata[key] === "string" ? metadata[key] : void 0; const getSpanMetadataString = (span, key) => { const m = span?.metadata; return m && typeof m[key] === "string" ? m[key] : void 0; }; const parentSpan = this.getParentSpan(false); let rootSpan = this; while (rootSpan.parent) { rootSpan = rootSpan.parent; } const rootTags = rootSpan.tags?.length ? [...rootSpan.tags] : void 0; this.correlationContext = { traceId: this.traceId, spanId: this.id, tags: rootTags, entityType: this.entityType, entityId: this.entityId, entityName: this.entityName, entityVersionId: getMetadataString("entityVersionId"), parentEntityType: parentSpan?.entityType, parentEntityId: parentSpan?.entityId, parentEntityName: parentSpan?.entityName, parentEntityVersionId: getSpanMetadataString(parentSpan, "entityVersionId"), rootEntityType: rootSpan.entityType, rootEntityId: rootSpan.entityId, rootEntityName: rootSpan.entityName, rootEntityVersionId: getSpanMetadataString(rootSpan, "entityVersionId"), userId: getMetadataString("userId"), organizationId: getMetadataString("organizationId"), resourceId: getMetadataString("resourceId"), runId: getMetadataString("runId"), sessionId: getMetadataString("sessionId"), threadId: getMetadataString("threadId"), requestId: getMetadataString("requestId"), environment: getMetadataString("environment") ?? this.observabilityInstance.getMastraEnvironment?.(), source: getMetadataString("source"), serviceName: getMetadataString("serviceName") ?? this.observabilityInstance.getConfig().serviceName, experimentId: getMetadataString("experimentId") }; return this.correlationContext; } /** Returns a lightweight span ready for export */ exportSpan(includeInternalSpans) { const hideInput = this.traceState?.hideInput ?? false; const hideOutput = this.traceState?.hideOutput ?? false; return { id: this.id, traceId: this.traceId, name: this.name, type: this.type, entityType: this.entityType, entityId: this.entityId, entityName: this.entityName, attributes: this.attributes, metadata: this.metadata, startTime: this.startTime, endTime: this.endTime, input: hideInput ? void 0 : this.input, output: hideOutput ? void 0 : this.output, errorInfo: this.errorInfo, requestContext: this.requestContext, isEvent: this.isEvent, isRootSpan: this.isRootSpan, parentSpanId: this.getParentSpanId(includeInternalSpans), // Tags are only included for root spans ...this.isRootSpan && this.tags?.length ? { tags: this.tags } : {} }; } get externalTraceId() { return this.isValid ? this.traceId : void 0; } /** * Execute an async function within this span's tracing context. * Delegates to the bridge if available. */ async executeInContext(fn) { const bridge = this.observabilityInstance.getBridge(); if (bridge?.executeInContext) { const bridgeContextSpan = this.isInternal ? this.getParentSpan(false) : this; return bridge.executeInContext(bridgeContextSpan?.id ?? this.id, fn); } return fn(); } /** * Execute a synchronous function within this span's tracing context. * Delegates to the bridge if available. */ executeInContextSync(fn) { const bridge = this.observabilityInstance.getBridge(); if (bridge?.executeInContextSync) { const bridgeContextSpan = this.isInternal ? this.getParentSpan(false) : this; return bridge.executeInContextSync(bridgeContextSpan?.id ?? this.id, fn); } return fn(); } }; var DefaultSpan = class extends BaseSpan { id; traceId; constructor(options, observabilityInstance) { super(options, observabilityInstance); if (options.spanId && options.traceId) { this.id = options.spanId; this.traceId = options.traceId; if (options.parentSpanId) { this.parentSpanId = options.parentSpanId; } return; } const bridge = observabilityInstance.getBridge(); if (bridge && !this.isInternal) { const bridgeIds = bridge.createSpan(options); if (bridgeIds) { this.id = bridgeIds.spanId; this.traceId = bridgeIds.traceId; this.parentSpanId = bridgeIds.parentSpanId; return; } } if (options.parent) { this.traceId = options.parent.traceId; this.parentSpanId = options.parent.id; this.id = generateSpanId(); return; } this.traceId = getOrCreateTraceId(options); this.id = generateSpanId(); if (options.parentSpanId) { if (isValidSpanId(options.parentSpanId)) { this.parentSpanId = options.parentSpanId; } else { console.error( `[Mastra Tracing] Invalid parentSpanId: must be 1-16 hexadecimal characters, got "${options.parentSpanId}". Ignoring.` ); } } } end(options) { if (this.isEvent) { return; } this.endTime = /* @__PURE__ */ new Date(); if (options?.metadata) { this.metadata = { ...this.metadata, ...deepClean(options.metadata, this.deepCleanOptions) }; } if (this.isExcluded) { return; } if (options?.output !== void 0) { this.output = deepClean(options.output, this.deepCleanOptions); } if (options?.attributes) { this.attributes = { ...this.attributes, ...deepClean(options.attributes, this.deepCleanOptions) }; } } error(options) { if (this.isEvent) { return; } const { error: error$1, endSpan = true, attributes, metadata } = options; if (metadata) { this.metadata = { ...this.metadata, ...deepClean(metadata, this.deepCleanOptions) }; } if (!this.isExcluded) { this.errorInfo = deepClean( error$1 instanceof error.MastraError ? { id: error$1.id, details: error$1.details, category: error$1.category, domain: error$1.domain, message: error$1.message, name: error$1.name, // Prefer the original cause's stack when available. MastraError wraps // thrown errors, so its own stack points to the wrapping site rather // than where the underlying error was thrown. stack: error$1.cause instanceof Error && error$1.cause.stack || error$1.stack } : { message: error$1.message, name: error$1.name, stack: error$1.stack }, this.deepCleanOptions ); if (attributes) { this.attributes = { ...this.attributes, ...deepClean(attributes, this.deepCleanOptions) }; } } if (endSpan) { this.end(); } else { this.update({}); } } update(options) { if (this.isEvent) { return; } if (options.name !== void 0) { this.name = options.name; } if (options.metadata) { this.metadata = { ...this.metadata, ...deepClean(options.metadata, this.deepCleanOptions) }; } if (this.isExcluded) { return; } if (options.input !== void 0) { this.input = deepClean(options.input, this.deepCleanOptions); } if (options.output !== void 0) { this.output = deepClean(options.output, this.deepCleanOptions); } if (options.attributes) { this.attributes = { ...this.attributes, ...deepClean(options.attributes, this.deepCleanOptions) }; } } get isValid() { return true; } async export() { return JSON.stringify({ spanId: this.id, traceId: this.traceId, startTime: this.startTime, endTime: this.endTime, attributes: this.attributes, metadata: this.metadata }); } }; function fillRandomBytes(bytes) { try { const webCrypto = globalThis.crypto; if (webCrypto?.getRandomValues) { webCrypto.getRandomValues.call(webCrypto, bytes); return; } } catch { } for (let i = 0; i < bytes.length; i++) { bytes[i] = Math.floor(Math.random() * 256); } } function generateSpanId() { const bytes = new Uint8Array(8); fillRandomBytes(bytes); return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(""); } function generateTraceId() { const bytes = new Uint8Array(16); fillRandomBytes(bytes); return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(""); } function isValidTraceId(traceId) { return /^[0-9a-f]{1,32}$/i.test(traceId); } function isValidSpanId(spanId) { return /^[0-9a-f]{1,16}$/i.test(spanId); } function getOrCreateTraceId(options) { if (options.traceId) { if (isValidTraceId(options.traceId)) { return options.traceId; } else { console.error( `[Mastra Tracing] Invalid traceId: must be 1-32 hexadecimal characters, got "${options.traceId}". Generating new trace ID.` ); } } return generateTraceId(); } // src/spans/no-op.ts var NoOpSpan = class extends BaseSpan { id; traceId; constructor(options, observabilityInstance) { super(options, observabilityInstance); this.id = "no-op"; this.traceId = "no-op-trace"; } end(_options) { } error(_options) { } update(_options) { } get isValid() { return false; } // NoOpSpan is never exported, so treat it as always excluded. get alwaysExcluded() { return true; } }; // src/instances/base.ts var BaseObservabilityInstance = class extends base.MastraBase { config; /** * Unified event bus for all observability signals. * Routes events to registered exporters based on event type. */ observabilityBus; /** * Cardinality filter for metrics label protection. */ cardinalityFilter; /** * Deployment environment propagated from the parent Mastra instance. * Set by `Observability.setMastraContext`, read by spans as a fallback when * a span's `metadata.environment` isn't set. */ #mastraEnvironment; constructor(config) { super({ component: logger.RegisteredLogger.OBSERVABILITY, name: config.serviceName }); this.config = { serviceName: config.serviceName, name: config.name, sampling: config.sampling ?? { type: "always" /* ALWAYS */ }, exporters: config.exporters ?? [], spanOutputProcessors: config.spanOutputProcessors ?? [], bridge: config.bridge ?? void 0, includeInternalSpans: config.includeInternalSpans ?? false, excludeSpanTypes: config.excludeSpanTypes, spanFilter: config.spanFilter, requestContextKeys: config.requestContextKeys ?? [], serializationOptions: config.serializationOptions, logging: config.logging }; this.cardinalityFilter = new CardinalityFilter(config.cardinality); this.observabilityBus = new ObservabilityBus({ serializationOptions: this.config.serializationOptions }); for (const exporter of this.exporters) { this.observabilityBus.registerExporter(exporter); } if (this.config.bridge) { this.observabilityBus.registerBridge(this.config.bridge); } if (this.config.bridge?.init) { this.config.bridge.init({ config: this.config }); } } /** * Override setLogger to add Observability specific initialization log * and propagate logger to exporters and bridge */ __setLogger(logger) { super.__setLogger(logger); this.exporters.forEach((exporter) => { if (typeof exporter.__setLogger === "function") { exporter.__setLogger(logger); } }); if (this.config.bridge?.__setLogger) { this.config.bridge.__setLogger(logger); } this.logger.debug( `[Observability] Initialized [service=${this.config.serviceName}] [instance=${this.config.name}] [sampling=${this.config.sampling?.type}] [bridge=${!!this.config.bridge}]` ); } // ============================================================================ // Protected getters for clean config access // ============================================================================ get exporters() { return this.config.exporters || []; } get spanOutputProcessors() { return this.config.spanOutputProcessors || []; } // ============================================================================ // Public API - Single type-safe span creation method // ============================================================================ /** * Start a new span of a specific SpanType * * Sampling Decision: * - For root spans (no parent): Perform sampling check using the configured strategy * - For child spans: Inherit the sampling decision from the parent * - If parent is a NoOpSpan (not sampled), child is also a NoOpSpan * - If parent is a valid span (sampled), child is also sampled * * This ensures trace-level sampling: either all spans in a trace are sampled or none are. * See: https://github.com/mastra-ai/mastra/issues/11504 */ startSpan(options) { const { customSamplerOptions, requestContext, metadata, tracingOptions, ...rest } = options; if (options.parent) { if (!options.parent.isValid) { return new NoOpSpan({ ...rest, metadata }, this); } } else { if (!this.shouldSample(customSamplerOptions)) { return new NoOpSpan({ ...rest, metadata }, this); } } let traceState; if (options.parent) { traceState = options.parent.traceState; } else { traceState = this.computeTraceState(tracingOptions); } const tracingMetadata = !options.parent ? tracingOptions?.metadata : void 0; const mergedMetadata = metadata || tracingMetadata ? { ...metadata, ...tracingMetadata } : void 0; const enrichedMetadata = this.extractMetadataFromRequestContext(requestContext, mergedMetadata, traceState); const finalMetadata = !options.parent && this.#mastraEnvironment !== void 0 && (enrichedMetadata === void 0 || enrichedMetadata.environment === void 0) ? { ...enrichedMetadata ?? {}, environment: this.#mastraEnvironment } : enrichedMetadata; const tags = !options.parent ? tracingOptions?.tags : void 0; const traceId = !options.parent ? options.traceId ?? tracingOptions?.traceId : options.traceId; const parentSpanId = !options.parent ? options.parentSpanId ?? tracingOptions?.parentSpanId : options.parentSpanId; const span = this.createSpan({ ...rest, traceId, parentSpanId, metadata: finalMetadata, traceState, tags, requestContext }); if (span.isEvent) { this.emitSpanEnded(span); } else { this.wireSpanLifecycle(span); this.emitSpanStarted(span); } return span; } /** * Rebuild a span from exported data for lifecycle operations. * Used by durable execution engines (e.g., Inngest) to end/update spans * that were created in a previous durable operation. * * The rebuilt span: * - Does NOT emit SPAN_STARTED (assumes original span already did) * - Can have end(), update(), error() called on it * - Will emit SPAN_ENDED or SPAN_UPDATED when those methods are called * * @param cached - The exported span data to rebuild from * @returns A span that can have lifecycle methods called on it */ rebuildSpan(cached) { const span = this.createSpan({ name: cached.name, type: cached.type, traceId: cached.traceId, spanId: cached.id, parentSpanId: cached.parentSpanId, startTime: cached.startTime instanceof Date ? cached.startTime : new Date(cached.startTime), input: cached.input, attributes: cached.attributes, metadata: cached.metadata, entityType: cached.entityType, entityId: cached.entityId, entityName: cached.entityName }); this.wireSpanLifecycle(span); return span; } // ============================================================================ // Configuration Management // ============================================================================ /** * Get current configuration */ getConfig() { return { ...this.config }; } /** * Returns the deployment environment propagated from the parent Mastra instance. * Spans use this as a fallback when `metadata.environment` isn't set. */ getMastraEnvironment() { return this.#mastraEnvironment; } /** * Internal hook used by `Observability.setMastraContext` to push the * resolved Mastra-level environment into this instance. */ __setMastraEnvironment(environment) { this.#mastraEnvironment = environment; } // ============================================================================ // Plugin Access // ============================================================================ /** * Get all exporters */ getExporters() { return [...this.exporters]; } /** * Register an additional exporter at runtime. * Adds to both the bus (for event routing) and the config (for getExporters). */ registerExporter(exporter) { this.observabilityBus.registerExporter(exporter); this.config.exporters ??= []; if (this.config.exporters.includes(exporter)) { return; } this.config.exporters.push(exporter); if (typeof exporter.__setLogger === "function") { exporter.__setLogger(this.logger); } } /** * Get all span output processors */ getSpanOutputProcessors() { return [...this.spanOutputProcessors]; } /** * Get the bridge instance if configured */ getBridge() { return this.config.bridge; } /** * Get the logger instance (for exporters and other components) */ getLogger() { return this.logger; } /** * Get the ObservabilityBus for this instance. * The bus routes all observability events (tracing, logs, metrics, scores, feedback) * to registered exporters based on event type. */ getObservabilityBus() { return this.observabilityBus; } // ============================================================================ // Context-factory bridge methods // ============================================================================ /** * Get a LoggerContext correlated to a span. * Called by the context-factory in core (deriveLoggerContext) so that * `observabilityContext.loggerVNext` is a real logger instead of no-op. */ getLoggerContext(span) { if (this.config.logging?.enabled === false) { return observability.noOpLoggerContext; } const correlationContext = span?.getCorrelationContext?.(); const metadata = span?.metadata ? structuredClone(span.metadata) : void 0; return new LoggerContextImpl({ traceId: span?.traceId, spanId: span?.id, correlationContext, metadata, observabilityBus: this.observabilityBus, minLevel: this.config.logging?.level }); } /** * Get a MetricsContext correlated to a span. * Called by the context-factory in core (deriveMetricsContext) so that * `observabilityContext.metrics` is a real metrics context instead of no-op. */ getMetricsContext(span) { const correlationContext = span?.getCorrelationContext?.(); const metadata = span?.metadata ? structuredClone(span.metadata) : void 0; return new MetricsContextImpl({ traceId: span?.traceId, spanId: span?.id, correlationContext, metadata, cardinalityFilter: this.cardinalityFilter, observabilityBus: this.observabilityBus }); } /** * Emit any observability event through the bus. * The bus routes the event to the appropriate handler on each registered exporter, * and for tracing events triggers auto-extracted metrics. */ emitObservabilityEvent(event) { this.observabilityBus.emit(event); } /** * Internal hook used by RecordedTrace/RecordedSpan hydration to route * non-tracing annotation events back through the normal exporter pipeline. */ __emitRecordedEvent(event) { this.emitObservabilityEvent(event); } /** * Internal hook used by the client observability proxy (`client/`) * to route already-validated events through the normal bus without * going through the live span lifecycle. The caller is responsible * for constructing well-formed `ExportedSpan`s/`ExportedLog`s and * for any validation needed. */ __receiveExternalEvent(event) { this.emitObservabilityEvent(event); } // ============================================================================ // Span Lifecycle Management // ============================================================================ /** * Automatically wires up Observability lifecycle events for any span * This ensures all spans emit events regardless of implementation */ wireSpanLifecycle(span) { if (!this.config.includeInternalSpans && span.isInternal && span.type !== observability.SpanType.MODEL_GENERATION) { return; } const originalEnd = span.end.bind(span); const originalUpdate = span.update.bind(span); span.end = (options) => { if (span.isEvent) { this.logger.warn(`End event is not available on event spans`); return; } const rollupTarget = this.captureModelUsageRollup(span, options); originalEnd(options); if (rollupTarget) { this.applyUsageRollup(rollupTarget); } this.emitSpanEnded(span); }; span.update = (options) => { if (span.isEvent) { this.logger.warn(`Update() is not available on event spans`); return; } originalUpdate(options); this.emitSpanUpdated(span); }; } // ============================================================================ // Utility Methods // ============================================================================ /** * Check if a trace should be sampled */ shouldSample(options) { const { sampling } = this.config; switch (sampling?.type) { case void 0: return true; case "always" /* ALWAYS */: return true; case "never" /* NEVER */: return false; case "ratio" /* RATIO */: if (sampling.probability === void 0 || sampling.probability < 0 || sampling.probability > 1) { this.logger.warn( `Invalid sampling probability: ${sampling.probability}. Expected value between 0 and 1. Defaulting to no sampling.` ); return false; } return Math.random() < sampling.probability; case "custom" /* CUSTOM */: return sampling.sampler(options); default: throw new Error(`Sampling strategy type not implemented: ${sampling.type}`); } } /** * Compute TraceState for a new trace based on configured and per-request keys */ computeTraceState(tracingOptions) { const configuredKeys = this.config.requestContextKeys ?? []; const additionalKeys = tracingOptions?.requestContextKeys ?? []; const allKeys = [...configuredKeys, ...additionalKeys]; const hideInput = tracingOptions?.hideInput; const hideOutput = tracingOptions?.hideOutput; if (allKeys.length === 0 && !hideInput && !hideOutput) { return void 0; } return { requestContextKeys: allKeys, ...hideInput !== void 0 && { hideInput }, ...hideOutput !== void 0 && { hideOutput } }; } /** * Extract metadata from RequestContext using TraceState */ extractMetadataFromRequestContext(requestContext, explicitMetadata, traceState) { if (!requestContext || !traceState || traceState.requestContextKeys.length === 0) { return explicitMetadata; } const extracted = this.extractKeys(requestContext, traceState.requestContextKeys); if (Object.keys(extracted).length === 0 && !explicitMetadata) { return void 0; } return { ...extracted, ...explicitMetadata // Explicit metadata always wins }; } /** * Extract specific keys from RequestContext */ extractKeys(requestContext, keys) { const result = {}; for (const key of keys) { const parts = key.split("."); const rootKey = parts[0]; const value = requestContext.get(rootKey); if (value !== void 0) { if (parts.length > 1) { const nestedPath = parts.slice(1).join("."); const nestedValue = utils.getNestedValue(value, nestedPath); if (nestedValue !== void 0) { utils.setNestedValue(result, key, nestedValue); } } else { utils.setNestedValue(result, key, value); } } } return result; } /** * Process a span through all output processors */ processSpan(span) { for (const processor of this.spanOutputProcessors) { if (!span) { break; } try { span = processor.process(span); } catch (error) { this.logger.error(`[Observability] Processor error [name=${processor.name}]`, error); } } return span; } // ============================================================================ // Event-driven Export Methods // ============================================================================ /** Process a span through output processors and export it, returning undefined if filtered out. */ getSpanForExport(span) { if (!span.isValid) return void 0; if (span.isInternal && !this.config.includeInternalSpans) return void 0; if (this.config.excludeSpanTypes?.includes(span.type)) return void 0; const processedSpan = this.processSpan(span); const exportedSpan = processedSpan?.exportSpan(this.config.includeInternalSpans); if (!exportedSpan) return void 0; if (this.config.spanFilter) { try { if (!this.config.spanFilter(exportedSpan)) return void 0; } catch (error) { this.logger.error(`[Observability] spanFilter error`, error); } } return exportedSpan; } /** * Emit a span started event. * Routes through the ObservabilityBus so exporters receive it via onTracingEvent. */ emitSpanStarted(span) { const exportedSpan = this.getSpanForExport(span); if (exportedSpan) { const event = { type: observability.TracingEventType.SPAN_STARTED, exportedSpan }; this.emitTracingEvent(event); } } /** * Emit a span ended event (called automatically when spans end). * Emits any auto-extracted metrics while the live span tree is still available, * then routes the exported tracing event through the ObservabilityBus. */ emitSpanEnded(span) { const exportedSpan = this.getSpanForExport(span); if (exportedSpan) { try { emitAutoExtractedMetrics(span, this.getMetricsContext(span)); } catch (err) { this.logger.error("[Observability] Auto-extraction error:", err); } const event = { type: observability.TracingEventType.SPAN_ENDED, exportedSpan }; this.emitTracingEvent(event); } } /** * Emit a span updated event. * Routes through the ObservabilityBus so exporters receive it via onTracingEvent. */ emitSpanUpdated(span) { const exportedSpan = this.getSpanForExport(span); if (exportedSpan) { const event = { type: observability.TracingEventType.SPAN_UPDATED, exportedSpan }; this.emitTracingEvent(event); } } /** * When an internal MODEL_GENERATION span ends, capture the rollup payload * (usage, provider, model, target ancestor) needed to attribute its cost * to the closest exported ancestor span. Returns undefined when no rollup * applies — non-MODEL_GENERATION spans, spans that will be exported, or * spans whose usage isn't available at end time. */ captureModelUsageRollup(span, endOptions) { if (span.type !== observability.SpanType.MODEL_GENERATION) return void 0; if (!span.isInternal || this.config.includeInternalSpans) return void 0; const endAttrs = endOptions?.attributes ?? void 0; const liveAttrs = span.attributes; const usage = endAttrs?.usage ?? liveAttrs?.usage; if (!usage) return void 0; const ancestor = this.findExportedAncestor(span); if (!ancestor) return void 0; const provider = endAttrs?.provider ?? liveAttrs?.provider; const model = endAttrs?.responseModel ?? endAttrs?.model ?? liveAttrs?.responseModel ?? liveAttrs?.model; return { ancestor, usage, provider, model }; } /** * Accumulate usage onto the ancestor's `internalUsage` attribute (for trace * UI visibility) and emit auto-extracted token metrics now, using the * ancestor's metrics context so cost / token labels point at the visible * span instead of the hidden agent that incurred them. */ applyUsageRollup(target) { const { ancestor, usage, provider, model } = target; const attrs = ancestor.attributes; attrs.internalUsage = addUsageStats(attrs.internalUsage, usage); try { emitTokenMetricsForUsage(usage, provider, model, this.getMetricsContext(ancestor)); } catch (err) { this.logger.error("[Observability] Usage rollup metric emission error:", err); } } /** * Walk up the parent chain to find the closest ancestor that will actually * reach exporters. Skips both internal-filtered ancestors and ancestors * whose type matches `excludeSpanTypes`, so the rollup target is one whose * mutated `internalUsage` attribute is visible in exported traces. * * Note: this does not preemptively run `spanFilter` — that filter can be * async and have side effects, so the rare case of a `spanFilter`-dropped * ancestor falls through. */ findExportedAncestor(span) { let ancestor = span.parent; while (ancestor && this.isFilteredFromExport(ancestor)) { ancestor = ancestor.parent; } return ancestor; } /** * Returns true when a span would be dropped by `getSpanForExport` for a * reason cheap to check up-front (internal-span filtering or * `excludeSpanTypes`). Used by `findExportedAncestor` to skip rollup * targets that would silently lose their `internalUsage` attribute. */ isFilteredFromExport(span) { if (span.isInternal && !this.config.includeInternalSpans) return true; if (this.config.excludeSpanTypes?.includes(span.type)) return true; return false; } /** * Emit a tracing event through the bus. * * The bus routes the event to each registered exporter's and bridge's * onTracingEvent handler. */ emitTracingEvent(event) { this.observabilityBus.emit(event); } /** * Export tracing event through all exporters and bridge. * * @deprecated Prefer emitTracingEvent() which routes through the bus. * Kept for backward compatibility with subclasses that may override it. */ async exportTracingEvent(event) { const targets = [ ...this.exporters ]; if (this.config.bridge) { targets.push(this.config.bridge); } const exportPromises = targets.map(async (target) => { try { await target.exportTracingEvent(event); this.logger.debug(`[Observability] Event exported [target=${target.name}] [type=${event.type}]`); } catch (error) { this.logger.error(`[Observability] Export error [target=${target.name}]`, error); } }); await Promise.allSettled(exportPromises); } // ============================================================================ // Lifecycle Management // ============================================================================ /** * Initialize Observability (called by Mastra during component registration) */ init() { this.logger.debug(`[Observability] Initialization started [name=${this.name}]`); this.logger.info(`[Observability] Initialized successfully [name=${this.name}]`); } /** * Flush all observability data: awaits in-flight handler promises, then * drains exporter and bridge SDK-internal buffers. * * Delegates to ObservabilityBus.flush() which owns the two-phase logic. * * This is critical for durable execution engines (e.g., Inngest) where * the process may be interrupted after a step completes. Calling flush() * outside the durable step ensures all span data reaches external systems. */ async flush() { this.logger.debug(`[Observability] Flush started [name=${this.name}]`); await this.observabilityBus.flush(); this.logger.debug(`[Observability] Flush completed [name=${this.name}]`); } /** * Shutdown Observability and clean up resources */ async shutdown() { this.logger.debug(`[Observability] Shutdown started [name=${this.name}]`); await this.observabilityBus.shutdown(); const shutdownPromises = [ ...this.exporters.map((e) => e.shutdown()), ...this.spanOutputProcessors.map((p) => p.shutdown()) ]; if (this.config.bridge) { shutdownPromises.push(this.config.bridge.shutdown()); } if (shutdownPromises.length > 0) { const results = await Promise.allSettled(shutdownPromises); for (const result of results) { if (result.status === "rejected") { this.logger.error(`[Observability] Component shutdown failed [name=${this.name}]:`, result.reason); } } } this.logger.info(`[Observability] Shutdown completed [name=${this.name}]`); } }; // src/client/id.ts function generateClientSignalId() { return globalThis.crypto.randomUUID(); } function nanosToDate(value) { if (typeof value === "number") { return new Date(Math.round(value / 1e6)); } if (typeof value === "string") { if (value.length <= 6) return /* @__PURE__ */ new Date(0); const ms = Number(value.slice(0, -6)); return Number.isFinite(ms) ? new Date(ms) : /* @__PURE__ */ new Date(0); } return /* @__PURE__ */ new Date(0); } function flattenAttributes(otlpAttrs) { if (!Array.isArray(otlpAttrs)) return void 0; const out = {}; for (const attr of otlpAttrs) { if (!attr || typeof attr !== "object") continue; const key = attr.key; if (typeof key !== "string") continue; const v = attr.value; if (!v || typeof v !== "object") continue; const value = v; if ("stringValue" in value) out[key] = value.stringValue; else if ("intValue" in value) out[key] = Number(value.intValue); else if ("doubleValue" in value) out[key] = value.doubleValue; else if ("boolValue" in value) out[key] = value.boolValue; else if ("arrayValue" in value || "kvlistValue" in value) { out[key] = value; } } return Object.keys(out).length > 0 ? out : void 0; } function decodeResourceSpans(payload) { const out = []; const resourceSpans = payload?.resourceSpans; if (!Array.isArray(resourceSpans)) return out; for (const rs of resourceSpans) { const scopeSpans = rs?.scopeSpans; if (!Array.isArray(scopeSpans)) continue; for (const ss of scopeSpans) { const spans = ss?.spans; if (!Array.isArray(spans)) continue; for (const span of spans) { if (!span || typeof span !== "object") continue; const s = span; const traceId = typeof s.traceId === "string" ? s.traceId : void 0; const spanId = typeof s.spanId === "string" ? s.spanId : void 0; const name = typeof s.name === "string" ? s.name : void 0; if (!traceId || !spanId || !name) continue; const decoded = { traceId, spanId, parentSpanId: typeof s.parentSpanId === "string" && s.parentSpanId ? s.parentSpanId : void 0, name, startTime: nanosToDate(s.startTimeUnixNano), endTime: s.endTimeUnixNano !== void 0 ? nanosToDate(s.endTimeUnixNano) : void 0, attributes: flattenAttributes(s.attributes) }; const status = s.status; if (status && typeof status === "object") { if (typeof status.code === "number") decoded.statusCode = status.code; if (typeof status.message === "string") decoded.statusMessage = status.message; } out.push(decoded); } } } return out; } function decodeResourceLogs(payload) { const out = []; const resourceLogs = payload?.resourceLogs; if (!Array.isArray(resourceLogs)) return out; for (const rl of resourceLogs) { const scopeLogs = rl?.scopeLogs; if (!Array.isArray(scopeLogs)) continue; for (const sl of scopeLogs) { const logRecords = sl?.logRecords; if (!Array.isArray(logRecords)) continue; for (const record of logRecords) { if (!record || typeof record !== "object") continue; const r = record; const traceId = typeof r.traceId === "string" ? r.traceId : void 0; if (!traceId) continue; let body = r.body; if (body && typeof body === "object" && "stringValue" in body) { body = body.stringValue; } out.push({ traceId, spanId: typeof r.spanId === "string" && r.spanId ? r.spanId : void 0, timestamp: r.timeUnixNano !== void 0 ? nanosToDate(r.timeUnixNano) : nanosToDate(r.observedTimeUnixNano), severityText: typeof r.severityText === "string" ? r.severityText : void 0, severityNumber: typeof r.severityNumber === "number" ? r.severityNumber : void 0, body, attributes: flattenAttributes(r.attributes) }); } } } return out; } function otlpSeverityToLogLevel(text, num) { if (text) { const lc = text.toLowerCase(); if (lc.startsWith("trace") || lc.startsWith("debug")) return "debug"; if (lc.startsWith("info")) return "info"; if (lc.startsWith("warn")) return "warn"; if (lc.startsWith("error")) return "error"; if (lc.startsWith("fatal")) return "fatal"; } if (typeof num === "number") { if (num <= 8) return "debug"; if (num <= 12) return "info"; if (num <= 16) return "warn"; if (num <= 20) return "error"; return "fatal"; } return "info"; } function buildExportedSpan(decoded, options = {}) { const errorInfo = decoded.statusCode === 2 ? { message: decoded.statusMessage ?? "Client tool span reported error status" // SpanErrorInfo allows extra fields; keep the minimum. } : void 0; return { id: decoded.spanId, traceId: decoded.traceId, name: decoded.name, type: observability.SpanType.GENERIC, parentSpanId: decoded.parentSpanId, isRootSpan: !decoded.parentSpanId, startTime: decoded.startTime, endTime: decoded.endTime, attributes: decoded.attributes, entityType: options.entityType, entityName: options.entityName, isEvent: false, ...errorInfo ? { errorInfo } : {} }; } function buildExportedLog(decoded) { return { logId: generateClientSignalId(), timestamp: decoded.timestamp, traceId: decoded.traceId, spanId: decoded.spanId, level: otlpSeverityToLogLevel(decoded.severityText, decoded.severityNumber), message: typeof decoded.body === "string" ? decoded.body : decoded.body !== void 0 ? String(decoded.body) : "", data: decoded.attributes }; } // src/client/w3c.ts var TRACEPARENT_RE = /^([0-9a-f]{2})-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$/; function parseTraceparent(value) { if (!value) return null; const m = TRACEPARENT_RE.exec(value.trim()); if (!m) return null; if (m[1] === "ff") return null; if (m[2] === "00000000000000000000000000000000") return null; if (m[3] === "0000000000000000") return null; return { version: m[1], traceId: m[2], spanId: m[3], flags: m[4] }; } function formatTraceparent(traceId, spanId, sampled) { return `00-${traceId}-${spanId}-${sampled ? "01" : "00"}`; } function parseBaggage(value) { const out = /* @__PURE__ */ new Map(); if (!value) return out; for (const entry of value.split(",")) { const trimmed = entry.trim(); if (!trimmed) continue; const semi = trimmed.indexOf(";"); const head = semi === -1 ? trimmed : trimmed.slice(0, semi); const eq = head.indexOf("="); if (eq === -1) continue; const key = head.slice(0, eq).trim(); const rawValue = head.slice(eq + 1).trim(); if (!key) continue; let decoded; try { decoded = decodeURIComponent(rawValue); } catch { decoded = rawValue; } out.set(key, decoded); } return out; } function formatBaggage(entries) { const iter = entries instanceof Map ? entries.entries() : Object.entries(entries); const parts = []; for (const [key, value] of iter) { if (!key) continue; parts.push(`${key}=${encodeURIComponent(value)}`); } return parts.join(","); } // src/client/proxy.ts var DEFAULT_LIMITS = { maxSpans: 10, maxLogs: 10, maxPayloadBytes: 1024 * 1024 }; var ClientObservabilityProxyImpl = class { #resolveInstance; #logger; #limits; constructor(options) { this.#resolveInstance = options.resolveInstance; this.#logger = options.logger; this.#limits = { ...DEFAULT_LIMITS, ...options.limits }; } inject(parentSpan) { return { // Mastra spans use OTel-compatible 32-hex traceIds and 16-hex // spanIds, so they drop straight into the W3C format. Sampled // flag is always 1: if the parent span exists at all, the trace // is being recorded server-side, so the client should record too. traceparent: formatTraceparent(parentSpan.traceId, parentSpan.id, true) }; } receive(payload, parentContext) { if (!payload || !payload.spans && !payload.logs && payload.executionDurationMs === void 0) { return; } let payloadBytes = 0; try { payloadBytes = JSON.stringify(payload).length; } catch { this.#warn("Client observability payload is not JSON-serializable; dropping."); return; } if (payloadBytes > this.#limits.maxPayloadBytes) { this.#warn("Client observability payload exceeds size limit; dropping.", { bytes: payloadBytes, limit: this.#limits.maxPayloadBytes }); return; } const parent = parseTraceparent(parentContext.traceparent); if (!parent) { this.#warn("Client observability parentContext.traceparent is malformed; dropping payload."); return; } const instance = this.#resolveInstance(); if (!instance || !(instance instanceof BaseObservabilityInstance)) { return; } const decodedSpans = payload.spans ? decodeResourceSpans(payload.spans) : []; const decodedLogs = payload.logs ? decodeResourceLogs(payload.logs) : []; if (decodedSpans.length > this.#limits.maxSpans) { this.#warn("Client observability payload exceeds span count limit; dropping.", { spans: decodedSpans.length, limit: this.#limits.maxSpans }); return; } if (decodedLogs.length > this.#limits.maxLogs) { this.#warn("Client observability payload exceeds log count limit; dropping.", { logs: decodedLogs.length, limit: this.#limits.maxLogs }); return; } if (!validateTraceIds(decodedSpans, decodedLogs, parent.traceId)) { this.#warn("Client observability payload contains spans or logs from a foreign trace; dropping."); return; } if (!validateParentLinks(decodedSpans, parent.spanId)) { this.#warn("Client observability payload contains spans with orphan parents; dropping."); return; } for (const decoded of decodedSpans) { const exported = buildExportedSpan(decoded); const startedEvent = { type: observability.TracingEventType.SPAN_STARTED, exportedSpan: exported }; instance.__receiveExternalEvent(startedEvent); if (exported.endTime) { const endedEvent = { type: observability.TracingEventType.SPAN_ENDED, exportedSpan: exported }; instance.__receiveExternalEvent(endedEvent); } } for (const decoded of decodedLogs) { const log = buildExportedLog(decoded); const event = { type: "log", log }; instance.__receiveExternalEvent(event); } if (typeof payload.executionDurationMs === "number") { const hasError = decodedSpans.some((s) => s.statusCode === 2); const metricEvent = { type: "metric", metric: { metricId: generateClientSignalId(), timestamp: /* @__PURE__ */ new Date(), traceId: parent.traceId, spanId: parent.spanId, name: "mastra_tool_duration_ms", value: payload.executionDurationMs, labels: { status: hasError ? "error" : "ok", toolType: "client" }, correlationContext: { traceId: parent.traceId, spanId: parent.spanId, entityType: observability.EntityType.TOOL, ...payload.toolName ? { entityName: payload.toolName } : {} } } }; instance.__receiveExternalEvent(metricEvent); } } #warn(message, data) { if (this.#logger) { this.#logger.warn(`[ClientObservabilityProxy] ${message}`, data); } } }; function validateTraceIds(spans, logs, expected) { for (const s of spans) if (s.traceId !== expected) return false; for (const l of logs) if (l.traceId !== expected) return false; return true; } function validateParentLinks(spans, rootParent) { if (spans.length === 0) return true; const known = /* @__PURE__ */ new Set([rootParent]); for (const s of spans) known.add(s.spanId); for (const s of spans) { if (!s.parentSpanId) { return false; } if (!known.has(s.parentSpanId)) { return false; } } return true; } function createClientObservabilityProxy(options) { return new ClientObservabilityProxyImpl(options); } var BaseExporter = class { /** Mastra logger instance */ logger; /** Base configuration (accessible by subclasses) */ baseConfig; /** Whether this exporter is disabled */ #disabled = false; /** Public getter for disabled state */ get isDisabled() { return this.#disabled; } /** * Initialize the base exporter with logger */ constructor(config = {}) { this.baseConfig = config; const logLevel = this.resolveLogLevel(config.logLevel); this.logger = config.logger ?? new logger.ConsoleLogger({ level: logLevel, name: this.constructor.name }); } /** * Set the logger for the exporter (called by Mastra/ObservabilityInstance during initialization) */ __setLogger(logger) { this.logger = logger; this.logger.debug(`Logger updated for exporter [name=${this.name}]`); } /** * Convert string log level to LogLevel enum */ resolveLogLevel(logLevel) { if (!logLevel) { return logger.LogLevel.INFO; } if (typeof logLevel === "number") { return logLevel; } const logLevelMap = { debug: logger.LogLevel.DEBUG, info: logger.LogLevel.INFO, warn: logger.LogLevel.WARN, error: logger.LogLevel.ERROR }; return logLevelMap[logLevel] ?? logger.LogLevel.INFO; } /** * Mark the exporter as disabled and log a message * * @param reason - Reason why the exporter is disabled */ setDisabled(reason, level = "warn") { this.#disabled = true; this.logger[level](`${this.name} disabled: ${reason}`); } /** * Apply the customSpanFormatter if configured. * This is called automatically by exportTracingEvent before _exportTracingEvent. * * Supports both synchronous and asynchronous formatters. If the formatter * returns a Promise, it will be awaited. * * @param event - The incoming tracing event * @returns The (possibly modified) event to process */ async applySpanFormatter(event) { if (this.baseConfig.customSpanFormatter) { try { const formattedSpan = await this.baseConfig.customSpanFormatter(event.exportedSpan); return { ...event, exportedSpan: formattedSpan }; } catch (error) { this.logger.error(`${this.name}: Error in customSpanFormatter`, { error, spanId: event.exportedSpan.id, traceId: event.exportedSpan.traceId }); } } return event; } /** * Default onTracingEvent handler that delegates to exportTracingEvent. * * This provides backward compatibility: existing exporters that only implement * _exportTracingEvent will automatically receive tracing events routed through * the ObservabilityBus. Subclasses can override this if they need different * routing behavior for bus-delivered events. * * Handler presence on ObservabilityExporter = signal support. */ onTracingEvent(event) { return this.exportTracingEvent(event); } /** * Export a tracing event * * This method checks if the exporter is disabled, applies the customSpanFormatter, * then calls _exportTracingEvent. * Subclasses should implement _exportTracingEvent instead of overriding this method. */ async exportTracingEvent(event) { if (this.isDisabled) { return; } const processedEvent = await this.applySpanFormatter(event); await this._exportTracingEvent(processedEvent); } /** * Force flush any buffered/queued spans without shutting down the exporter. * * This is useful in serverless environments where you need to ensure spans * are exported before the runtime instance is terminated, while keeping * the exporter active for future requests. * * Default implementation is a no-op. Override to add flush logic. */ async flush() { this.logger.debug(`${this.name} flush called (no-op in base class)`); } /** * Shutdown the exporter and clean up resources * * Default implementation just logs. Override to add custom cleanup. */ async shutdown() { this.logger.info(`${this.name} shutdown complete`); } }; var TraceData = class { /** The vendor-specific root/trace object */ #rootSpan; /** The span ID of the root span */ #rootSpanId; /** Whether a span with isRootSpan=true has been successfully processed */ #rootSpanProcessed; /** Maps eventId to vendor-specific event objects */ #events; /** Maps spanId to vendor-specific span objects */ #spans; /** Maps spanId to parentSpanId, representing the span hierarchy */ #tree; /** Set of span IDs that have started but not yet ended */ #activeSpanIds; /** Maps spanId to vendor-specific metadata */ #metadata; /** Arbitrary key-value storage for per-trace data */ #extraData; /** Events waiting for the root span to be processed */ #waitingForRoot; /** Events waiting for specific parent spans, keyed by parentSpanId */ #waitingForParent; /** When this trace data was created, used for cap enforcement */ createdAt; constructor() { this.#events = /* @__PURE__ */ new Map(); this.#spans = /* @__PURE__ */ new Map(); this.#activeSpanIds = /* @__PURE__ */ new Set(); this.#tree = /* @__PURE__ */ new Map(); this.#metadata = /* @__PURE__ */ new Map(); this.#extraData = /* @__PURE__ */ new Map(); this.#rootSpanProcessed = false; this.#waitingForRoot = []; this.#waitingForParent = /* @__PURE__ */ new Map(); this.createdAt = /* @__PURE__ */ new Date(); } /** * Check if this trace has a root span registered. * @returns True if addRoot() has been called */ hasRoot() { return !!this.#rootSpanId; } /** * Register the root span for this trace. * @param args.rootId - The span ID of the root span * @param args.rootData - The vendor-specific root object */ addRoot(args) { this.#rootSpanId = args.rootId; this.#rootSpan = args.rootData; this.#rootSpanProcessed = true; } /** * Get the vendor-specific root object. * @returns The root object, or undefined if not yet set */ getRoot() { return this.#rootSpan; } /** * Check if a span with isRootSpan=true has been successfully processed. * Set via addRoot() or markRootSpanProcessed(). * @returns True if the root span has been processed */ isRootProcessed() { return this.#rootSpanProcessed; } /** * Mark that the root span has been processed. * Used by exporters with skipBuildRootTask=true where root goes through _buildSpan * instead of _buildRoot. */ markRootSpanProcessed() { this.#rootSpanProcessed = true; } /** * Store an arbitrary value in per-trace storage. * @param key - Storage key * @param value - Value to store */ setExtraValue(key, value) { this.#extraData.set(key, value); } /** * Check if a key exists in per-trace storage. * @param key - Storage key * @returns True if the key exists */ hasExtraValue(key) { return this.#extraData.has(key); } /** * Get a value from per-trace storage. * @param key - Storage key * @returns The stored value, or undefined if not found */ getExtraValue(key) { return this.#extraData.get(key); } // ============================================================================ // Early Queue Methods // ============================================================================ /** * Add an event to the waiting queue. * @param args.event - The tracing event to queue * @param args.waitingFor - 'root' or a specific parentSpanId * @param args.attempts - Optional: preserve attempts count when re-queuing * @param args.queuedAt - Optional: preserve original queue time when re-queuing */ addToWaitingQueue(args) { const queuedEvent = { event: args.event, waitingFor: args.waitingFor, attempts: args.attempts ?? 0, queuedAt: args.queuedAt ?? /* @__PURE__ */ new Date() }; if (args.waitingFor === "root") { this.#waitingForRoot.push(queuedEvent); } else { const queue = this.#waitingForParent.get(args.waitingFor) ?? []; queue.push(queuedEvent); this.#waitingForParent.set(args.waitingFor, queue); } } /** * Get all events waiting for the root span. * Returns a copy of the internal array. */ getEventsWaitingForRoot() { return [...this.#waitingForRoot]; } /** * Get all events waiting for a specific parent span. * Returns a copy of the internal array. */ getEventsWaitingFor(args) { return [...this.#waitingForParent.get(args.spanId) ?? []]; } /** * Clear the waiting-for-root queue. */ clearWaitingForRoot() { this.#waitingForRoot = []; } /** * Clear the waiting queue for a specific parent span. */ clearWaitingFor(args) { this.#waitingForParent.delete(args.spanId); } /** * Get total count of events in all waiting queues. */ waitingQueueSize() { let count = this.#waitingForRoot.length; for (const queue of this.#waitingForParent.values()) { count += queue.length; } return count; } /** * Get all queued events across all waiting queues. * Used for cleanup and logging orphaned events. * @returns Array of all queued events */ getAllQueuedEvents() { const all = [...this.#waitingForRoot]; for (const queue of this.#waitingForParent.values()) { all.push(...queue); } return all; } // ============================================================================ // Span Tree Methods // ============================================================================ /** * Record the parent-child relationship for a span. * @param args.spanId - The child span ID * @param args.parentSpanId - The parent span ID, or undefined for root spans */ addBranch(args) { this.#tree.set(args.spanId, args.parentSpanId); } /** * Get the parent span ID for a given span. * @param args.spanId - The span ID to look up * @returns The parent span ID, or undefined if root or not found */ getParentId(args) { return this.#tree.get(args.spanId); } // ============================================================================ // Span Management Methods // ============================================================================ /** * Register a span and mark it as active. * @param args.spanId - The span ID * @param args.spanData - The vendor-specific span object */ addSpan(args) { this.#spans.set(args.spanId, args.spanData); this.#activeSpanIds.add(args.spanId); } /** * Check if a span exists (regardless of active state). * @param args.spanId - The span ID to check * @returns True if the span exists */ hasSpan(args) { return this.#spans.has(args.spanId); } /** * Get a span by ID. * @param args.spanId - The span ID to look up * @returns The vendor-specific span object, or undefined if not found */ getSpan(args) { return this.#spans.get(args.spanId); } /** * Mark a span as ended (no longer active). * @param args.spanId - The span ID to mark as ended */ endSpan(args) { this.#activeSpanIds.delete(args.spanId); } /** * Check if a span is currently active (started but not ended). * @param args.spanId - The span ID to check * @returns True if the span is active */ isActiveSpan(args) { return this.#activeSpanIds.has(args.spanId); } /** * Get the count of currently active spans. * @returns Number of active spans */ activeSpanCount() { return this.#activeSpanIds.size; } /** * Get all active span IDs. * @returns Array of active span IDs */ get activeSpanIds() { return [...this.#activeSpanIds]; } // ============================================================================ // Event Management Methods // ============================================================================ /** * Register an event. * @param args.eventId - The event ID * @param args.eventData - The vendor-specific event object */ addEvent(args) { this.#events.set(args.eventId, args.eventData); } // ============================================================================ // Metadata Methods // ============================================================================ /** * Store vendor-specific metadata for a span. * Note: This overwrites any existing metadata for the span. * @param args.spanId - The span ID * @param args.metadata - The vendor-specific metadata */ addMetadata(args) { this.#metadata.set(args.spanId, args.metadata); } /** * Get vendor-specific metadata for a span. * @param args.spanId - The span ID * @returns The metadata, or undefined if not found */ getMetadata(args) { return this.#metadata.get(args.spanId); } // ============================================================================ // Parent Lookup Methods // ============================================================================ /** * Get the parent span or event for a given span. * Looks up in both spans and events maps. * @param args.span - The span to find the parent for * @returns The parent span/event object, or undefined if root or not found */ getParent(args) { const parentId = args.span.parentSpanId; if (parentId) { if (this.#spans.has(parentId)) { return this.#spans.get(parentId); } if (this.#events.has(parentId)) { return this.#events.get(parentId); } } return void 0; } /** * Get the parent span/event or fall back to the root object. * Useful for vendors that attach child spans to either parent spans or the trace root. * @param args.span - The span to find the parent for * @returns The parent span/event, the root object, or undefined */ getParentOrRoot(args) { return this.getParent(args) ?? this.getRoot(); } }; var DEFAULT_EARLY_QUEUE_MAX_ATTEMPTS = 5; var DEFAULT_EARLY_QUEUE_TTL_MS = 3e4; var DEFAULT_TRACE_CLEANUP_DELAY_MS = 3e4; var DEFAULT_MAX_PENDING_CLEANUP_TRACES = 100; var DEFAULT_MAX_TOTAL_TRACES = 500; var TrackingExporter = class extends BaseExporter { /** Map of traceId to per-trace data container */ #traceMap = /* @__PURE__ */ new Map(); /** Flag to prevent processing during shutdown */ #shutdownStarted = false; /** Flag to prevent concurrent hard cap enforcement */ #hardCapEnforcementInProgress = false; /** Map of traceId to scheduled cleanup timeout */ #pendingCleanups = /* @__PURE__ */ new Map(); // Note: #traceMap maintains insertion order (JS Map spec), so we use // #traceMap.keys() to iterate traces oldest-first for cap enforcement. /** Subclass configuration with resolved values */ config; /** Maximum attempts to process a queued event before dropping */ #earlyQueueMaxAttempts; /** TTL in milliseconds for queued events */ #earlyQueueTTLMs; /** Delay before cleaning up completed traces */ #traceCleanupDelayMs; /** Soft cap on traces awaiting cleanup */ #maxPendingCleanupTraces; /** Hard cap on total traces (will abort active spans if exceeded) */ #maxTotalTraces; constructor(config) { super(config); this.config = config; this.#earlyQueueMaxAttempts = config.earlyQueueMaxAttempts ?? DEFAULT_EARLY_QUEUE_MAX_ATTEMPTS; this.#earlyQueueTTLMs = config.earlyQueueTTLMs ?? DEFAULT_EARLY_QUEUE_TTL_MS; this.#traceCleanupDelayMs = config.traceCleanupDelayMs ?? DEFAULT_TRACE_CLEANUP_DELAY_MS; this.#maxPendingCleanupTraces = config.maxPendingCleanupTraces ?? DEFAULT_MAX_PENDING_CLEANUP_TRACES; this.#maxTotalTraces = config.maxTotalTraces ?? DEFAULT_MAX_TOTAL_TRACES; } // ============================================================================ // Early Queue Processing // ============================================================================ /** * Schedule async processing of events waiting for root span. * Called after root span is successfully processed. */ #scheduleProcessWaitingForRoot(traceId) { setImmediate(() => { this.#processWaitingForRoot(traceId).catch((error) => { this.logger.error(`${this.name}: Error processing waiting-for-root queue`, { error, traceId }); }); }); } /** * Schedule async processing of events waiting for a specific parent span. * Called after a span/event is successfully created. */ #scheduleProcessWaitingFor(traceId, spanId) { setImmediate(() => { this.#processWaitingFor(traceId, spanId).catch((error) => { this.logger.error(`${this.name}: Error processing waiting queue`, { error, traceId, spanId }); }); }); } /** * Process all events waiting for root span. */ async #processWaitingForRoot(traceId) { if (this.#shutdownStarted) return; const traceData = this.#traceMap.get(traceId); if (!traceData) return; const queue = traceData.getEventsWaitingForRoot(); if (queue.length === 0) return; this.logger.debug(`${this.name}: Processing ${queue.length} events waiting for root`, { traceId }); const toKeep = []; const now = Date.now(); for (const queuedEvent of queue) { if (now - queuedEvent.queuedAt.getTime() > this.#earlyQueueTTLMs) { this.logger.warn(`${this.name}: Dropping event due to TTL expiry`, { traceId, spanId: queuedEvent.event.exportedSpan.id, waitingFor: queuedEvent.waitingFor, queuedAt: queuedEvent.queuedAt, attempts: queuedEvent.attempts }); continue; } if (queuedEvent.attempts >= this.#earlyQueueMaxAttempts) { this.logger.warn(`${this.name}: Dropping event due to max attempts`, { traceId, spanId: queuedEvent.event.exportedSpan.id, waitingFor: queuedEvent.waitingFor, attempts: queuedEvent.attempts }); continue; } queuedEvent.attempts++; const processed = await this.#tryProcessQueuedEvent(queuedEvent, traceData); if (!processed) { const parentId = queuedEvent.event.exportedSpan.parentSpanId; if (parentId && traceData.isRootProcessed()) { traceData.addToWaitingQueue({ event: queuedEvent.event, waitingFor: parentId, attempts: queuedEvent.attempts, queuedAt: queuedEvent.queuedAt }); } else { toKeep.push(queuedEvent); } } } traceData.clearWaitingForRoot(); for (const event of toKeep) { traceData.addToWaitingQueue({ event: event.event, waitingFor: "root", attempts: event.attempts, queuedAt: event.queuedAt }); } } /** * Process events waiting for a specific parent span. */ async #processWaitingFor(traceId, spanId) { if (this.#shutdownStarted) return; const traceData = this.#traceMap.get(traceId); if (!traceData) return; const queue = traceData.getEventsWaitingFor({ spanId }); if (queue.length === 0) return; this.logger.debug(`${this.name}: Processing ${queue.length} events waiting for span`, { traceId, spanId }); const toKeep = []; const now = Date.now(); for (const queuedEvent of queue) { if (now - queuedEvent.queuedAt.getTime() > this.#earlyQueueTTLMs) { this.logger.warn(`${this.name}: Dropping event due to TTL expiry`, { traceId, spanId: queuedEvent.event.exportedSpan.id, waitingFor: queuedEvent.waitingFor, queuedAt: queuedEvent.queuedAt, attempts: queuedEvent.attempts }); continue; } if (queuedEvent.attempts >= this.#earlyQueueMaxAttempts) { this.logger.warn(`${this.name}: Dropping event due to max attempts`, { traceId, spanId: queuedEvent.event.exportedSpan.id, waitingFor: queuedEvent.waitingFor, attempts: queuedEvent.attempts }); continue; } queuedEvent.attempts++; const processed = await this.#tryProcessQueuedEvent(queuedEvent, traceData); if (!processed) { toKeep.push(queuedEvent); } } traceData.clearWaitingFor({ spanId }); for (const event of toKeep) { traceData.addToWaitingQueue({ event: event.event, waitingFor: spanId, attempts: event.attempts, queuedAt: event.queuedAt }); } } /** * Try to process a queued event. * Returns true if successfully processed, false if still waiting for dependencies. */ async #tryProcessQueuedEvent(queuedEvent, traceData) { const { event } = queuedEvent; const { exportedSpan } = event; const method = this.getMethod(event); try { switch (method) { case "handleEventSpan": { traceData.addBranch({ spanId: exportedSpan.id, parentSpanId: exportedSpan.parentSpanId }); const eventData = await this._buildEvent({ span: exportedSpan, traceData }); if (eventData) { if (!this.skipCachingEventSpans) { traceData.addEvent({ eventId: exportedSpan.id, eventData }); } this.#scheduleProcessWaitingFor(exportedSpan.traceId, exportedSpan.id); return true; } return false; } case "handleSpanStart": { traceData.addBranch({ spanId: exportedSpan.id, parentSpanId: exportedSpan.parentSpanId }); const spanData = await this._buildSpan({ span: exportedSpan, traceData }); if (spanData) { traceData.addSpan({ spanId: exportedSpan.id, spanData }); if (exportedSpan.isRootSpan) { traceData.markRootSpanProcessed(); } this.#scheduleProcessWaitingFor(exportedSpan.traceId, exportedSpan.id); return true; } return false; } case "handleSpanUpdate": { await this._updateSpan({ span: exportedSpan, traceData }); return true; } case "handleSpanEnd": { traceData.endSpan({ spanId: exportedSpan.id }); await this._finishSpan({ span: exportedSpan, traceData }); if (traceData.activeSpanCount() === 0) { this.#scheduleCleanup(exportedSpan.traceId); } return true; } default: return false; } } catch (error) { this.logger.error(`${this.name}: Error processing queued event`, { error, event, method }); return false; } } // ============================================================================ // Delayed Cleanup // ============================================================================ /** * Schedule cleanup of trace data after a delay. * Allows late-arriving data to still be processed. */ #scheduleCleanup(traceId) { this.#cancelScheduledCleanup(traceId); this.logger.debug(`${this.name}: Scheduling cleanup in ${this.#traceCleanupDelayMs}ms`, { traceId }); const timeout = setTimeout(() => { this.#pendingCleanups.delete(traceId); this.#performCleanup(traceId); }, this.#traceCleanupDelayMs); this.#pendingCleanups.set(traceId, timeout); this.#enforcePendingCleanupCap(); } /** * Cancel a scheduled cleanup for a trace. */ #cancelScheduledCleanup(traceId) { const existingTimeout = this.#pendingCleanups.get(traceId); if (existingTimeout) { clearTimeout(existingTimeout); this.#pendingCleanups.delete(traceId); this.logger.debug(`${this.name}: Cancelled scheduled cleanup`, { traceId }); } } /** * Perform the actual cleanup of trace data. */ #performCleanup(traceId) { const traceData = this.#traceMap.get(traceId); if (!traceData) return; const orphanedEvents = traceData.getAllQueuedEvents(); if (orphanedEvents.length > 0) { this.logger.warn(`${this.name}: Dropping ${orphanedEvents.length} orphaned events on cleanup`, { traceId, orphanedEvents: orphanedEvents.map((e) => ({ spanId: e.event.exportedSpan.id, waitingFor: e.waitingFor, attempts: e.attempts, queuedAt: e.queuedAt })) }); } this.#traceMap.delete(traceId); this.logger.debug(`${this.name}: Cleaned up trace data`, { traceId }); } // ============================================================================ // Cap Enforcement // ============================================================================ /** * Enforce soft cap on pending cleanup traces. * Only removes traces with activeSpanCount == 0. */ #enforcePendingCleanupCap() { if (this.#pendingCleanups.size <= this.#maxPendingCleanupTraces) { return; } const toRemove = this.#pendingCleanups.size - this.#maxPendingCleanupTraces; this.logger.warn(`${this.name}: Pending cleanup cap exceeded, force-cleaning ${toRemove} traces`, { pendingCount: this.#pendingCleanups.size, cap: this.#maxPendingCleanupTraces }); let removed = 0; for (const traceId of this.#traceMap.keys()) { if (removed >= toRemove) break; if (this.#pendingCleanups.has(traceId)) { this.#cancelScheduledCleanup(traceId); this.#performCleanup(traceId); removed++; } } } /** * Enforce hard cap on total traces. * Will kill even active traces if necessary. * Uses a flag to prevent concurrent executions when called fire-and-forget. */ async #enforceHardCap() { if (this.#traceMap.size <= this.#maxTotalTraces || this.#hardCapEnforcementInProgress) { return; } this.#hardCapEnforcementInProgress = true; try { if (this.#traceMap.size <= this.#maxTotalTraces) { return; } const toRemove = this.#traceMap.size - this.#maxTotalTraces; this.logger.warn(`${this.name}: Total trace cap exceeded, killing ${toRemove} oldest traces`, { traceCount: this.#traceMap.size, cap: this.#maxTotalTraces }); const reason = { id: "TRACE_CAP_EXCEEDED", message: "Trace killed due to memory cap enforcement.", domain: "MASTRA_OBSERVABILITY", category: "SYSTEM" }; let removed = 0; for (const traceId of [...this.#traceMap.keys()]) { if (removed >= toRemove) break; const traceData = this.#traceMap.get(traceId); if (traceData) { for (const spanId of traceData.activeSpanIds) { const span = traceData.getSpan({ spanId }); if (span) { await this._abortSpan({ span, traceData, reason }); } } this.#cancelScheduledCleanup(traceId); this.#performCleanup(traceId); removed++; } } } finally { this.#hardCapEnforcementInProgress = false; } } // ============================================================================ // Lifecycle Hooks (Override in subclass) // ============================================================================ /** * Hook called before processing each tracing event. * Override to transform or enrich the event before processing. * * Note: The customSpanFormatter is applied at the BaseExporter level before this hook. * Subclasses can override this to add additional pre-processing logic. * * @param event - The incoming tracing event * @returns The (possibly modified) event to process */ async _preExportTracingEvent(event) { return event; } /** * Hook called after processing each tracing event. * Override to perform post-processing actions like flushing. */ async _postExportTracingEvent() { } // ============================================================================ // Behavior Flags (Override in subclass as needed) // ============================================================================ /** * If true, skip calling _buildRoot and let root spans go through _buildSpan. * Use when the vendor doesn't have a separate trace/root concept. * @default false */ skipBuildRootTask = false; /** * If true, skip processing span_updated events entirely. * Use when the vendor doesn't support incremental span updates. * @default false */ skipSpanUpdateEvents = false; /** * If true, don't cache event spans in TraceData. * Use when events can't be parents of other spans. * @default false */ skipCachingEventSpans = false; getMethod(event) { if (event.exportedSpan.isEvent) { return "handleEventSpan"; } const eventType = event.type; switch (eventType) { case observability.TracingEventType.SPAN_STARTED: return "handleSpanStart"; case observability.TracingEventType.SPAN_UPDATED: return "handleSpanUpdate"; case observability.TracingEventType.SPAN_ENDED: return "handleSpanEnd"; default: { const _exhaustiveCheck = eventType; throw new Error(`Unhandled event type: ${_exhaustiveCheck}`); } } } async _exportTracingEvent(event) { if (this.#shutdownStarted) { return; } const method = this.getMethod(event); if (method == "handleSpanUpdate" && this.skipSpanUpdateEvents) { return; } const traceId = event.exportedSpan.traceId; const traceData = this.getTraceData({ traceId, method }); const { exportedSpan } = await this._preExportTracingEvent(event); if (!this.skipBuildRootTask && !traceData.hasRoot()) { if (exportedSpan.isRootSpan) { this.logger.debug(`${this.name}: Building root`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id }); const rootData = await this._buildRoot({ span: exportedSpan, traceData }); if (rootData) { this.logger.debug(`${this.name}: Adding root`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id }); traceData.addRoot({ rootId: exportedSpan.id, rootData }); this.#scheduleProcessWaitingForRoot(traceId); } } else { this.logger.debug(`${this.name}: Root does not exist, adding span to waiting queue.`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id }); traceData.addToWaitingQueue({ event, waitingFor: "root" }); return; } } if (exportedSpan.metadata && this.name in exportedSpan.metadata) { const metadata = exportedSpan.metadata[this.name]; this.logger.debug(`${this.name}: Found provider metadata in span`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id, metadata }); traceData.addMetadata({ spanId: exportedSpan.id, metadata }); } try { switch (method) { case "handleEventSpan": { this.logger.debug(`${this.name}: handling event`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id }); traceData.addBranch({ spanId: exportedSpan.id, parentSpanId: exportedSpan.parentSpanId }); const eventData = await this._buildEvent({ span: exportedSpan, traceData }); if (eventData) { if (!this.skipCachingEventSpans) { this.logger.debug(`${this.name}: adding event to traceData`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id }); traceData.addEvent({ eventId: exportedSpan.id, eventData }); } this.#scheduleProcessWaitingFor(traceId, exportedSpan.id); } else { const parentId = exportedSpan.parentSpanId; this.logger.debug(`${this.name}: adding event to waiting queue`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id, waitingFor: parentId ?? "root" }); traceData.addToWaitingQueue({ event, waitingFor: parentId ?? "root" }); } break; } case "handleSpanStart": { this.logger.debug(`${this.name}: handling span start`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id }); traceData.addBranch({ spanId: exportedSpan.id, parentSpanId: exportedSpan.parentSpanId }); const spanData = await this._buildSpan({ span: exportedSpan, traceData }); if (spanData) { this.logger.debug(`${this.name}: adding span to traceData`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id }); traceData.addSpan({ spanId: exportedSpan.id, spanData }); if (exportedSpan.isRootSpan) { traceData.markRootSpanProcessed(); this.#scheduleProcessWaitingForRoot(traceId); } this.#scheduleProcessWaitingFor(traceId, exportedSpan.id); } else { const parentId = exportedSpan.parentSpanId; this.logger.debug(`${this.name}: adding span to waiting queue`, { traceId: exportedSpan.traceId, waitingFor: parentId ?? "root" }); traceData.addToWaitingQueue({ event, waitingFor: parentId ?? "root" }); } break; } case "handleSpanUpdate": this.logger.debug(`${this.name}: handling span update`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id }); await this._updateSpan({ span: exportedSpan, traceData }); break; case "handleSpanEnd": this.logger.debug(`${this.name}: handling span end`, { traceId: exportedSpan.traceId, spanId: exportedSpan.id }); traceData.endSpan({ spanId: exportedSpan.id }); await this._finishSpan({ span: exportedSpan, traceData }); if (traceData.activeSpanCount() === 0) { this.#scheduleCleanup(traceId); } break; } } catch (error) { this.logger.error(`${this.name}: exporter error`, { error, event, method }); } if (traceData.activeSpanCount() === 0) { this.#scheduleCleanup(traceId); } await this._postExportTracingEvent(); } // ============================================================================ // Protected Helpers // ============================================================================ /** * Get or create the TraceData container for a trace. * Also cancels any pending cleanup since new data has arrived. * * @param args.traceId - The trace ID * @param args.method - The calling method name (for logging) * @returns The TraceData container for this trace */ getTraceData(args) { const { traceId, method } = args; this.#cancelScheduledCleanup(traceId); if (!this.#traceMap.has(traceId)) { this.#traceMap.set(traceId, new TraceData()); this.logger.debug(`${this.name}: Created new trace data cache`, { traceId, method }); this.#enforceHardCap().catch((error) => { this.logger.error(`${this.name}: Error enforcing hard cap`, { error }); }); } return this.#traceMap.get(traceId); } /** * Get the current number of traces being tracked. * @returns The trace count */ traceMapSize() { return this.#traceMap.size; } // ============================================================================ // Flush and Shutdown Hooks (Override in subclass as needed) // ============================================================================ /** * Hook called by flush() to perform vendor-specific flush logic. * Override to send buffered data to the vendor's API. * * Unlike _postShutdown(), this method should NOT release resources, * as the exporter will continue to be used after flushing. */ async _flush() { } /** * Force flush any buffered data without shutting down the exporter. * This is useful in serverless environments where you need to ensure spans * are exported before the runtime instance is terminated. * * Subclasses should override _flush() to implement vendor-specific flush logic. */ async flush() { if (this.isDisabled) { return; } this.logger.debug(`${this.name}: Flushing`); await this._flush(); } /** * Hook called at the start of shutdown, before cancelling timers and aborting spans. * Override to perform vendor-specific pre-shutdown tasks. */ async _preShutdown() { } /** * Hook called at the end of shutdown, after all spans are aborted. * Override to perform vendor-specific cleanup (e.g., flushing). */ async _postShutdown() { } /** * Gracefully shut down the exporter. * Cancels all pending cleanup timers, aborts all active spans, and clears state. */ async shutdown() { if (this.isDisabled) { return; } this.#shutdownStarted = true; await this._preShutdown(); for (const [traceId, timeout] of this.#pendingCleanups) { clearTimeout(timeout); this.logger.debug(`${this.name}: Cancelled pending cleanup on shutdown`, { traceId }); } this.#pendingCleanups.clear(); const reason = { id: "SHUTDOWN", message: "Observability is shutting down.", domain: "MASTRA_OBSERVABILITY", category: "SYSTEM" }; for (const [traceId, traceData] of this.#traceMap) { const orphanedEvents = traceData.getAllQueuedEvents(); if (orphanedEvents.length > 0) { this.logger.warn(`${this.name}: Dropping ${orphanedEvents.length} orphaned events on shutdown`, { traceId, orphanedEvents: orphanedEvents.map((e) => ({ spanId: e.event.exportedSpan.id, waitingFor: e.waitingFor, attempts: e.attempts })) }); } for (const spanId of traceData.activeSpanIds) { const span = traceData.getSpan({ spanId }); if (span) { await this._abortSpan({ span, traceData, reason }); } } } this.#traceMap.clear(); await this._postShutdown(); await super.shutdown(); } }; // src/exporters/span-formatters.ts function chainFormatters(formatters) { return async (span) => { let currentSpan = span; for (const formatter of formatters) { currentSpan = await formatter(currentSpan); } return currentSpan; }; } var AUTH_FAILURE_STATUSES = /* @__PURE__ */ new Set([401, 403]); var AUTH_COOLDOWN_BASE_MS = 6e4; var AUTH_COOLDOWN_MAX_MS = 15 * 6e4; var AUTH_COOLDOWN_JITTER_RATIO = 0.1; var AuthFailureError = class extends Error { status; constructor(status, cause) { super(`Request failed with authentication status: ${status}`, { cause }); this.name = "AuthFailureError"; this.status = status; } }; function isAuthFailureError(error) { return error instanceof AuthFailureError; } function isAuthFailureStatus(status) { return AUTH_FAILURE_STATUSES.has(status); } async function fetchWithAuthFailureHandling(url, options, maxRetries) { let authFailureStatus; try { await utils.fetchWithRetry(url, options, maxRetries, { shouldRetryResponse: (response) => { if (isAuthFailureStatus(response.status)) { authFailureStatus = response.status; return false; } return true; } }); } catch (error) { if (authFailureStatus !== void 0) { throw new AuthFailureError(authFailureStatus, error); } throw error; } } var AuthFailureCooldown = class { constructor(exporterName, getLogger) { this.exporterName = exporterName; this.getLogger = getLogger; } exporterName; getLogger; failureCount = 0; cooldownUntilMs = 0; droppedEventsDuringCooldown = 0; shouldDropEvents() { return Date.now() < this.cooldownUntilMs; } dropEventIfCoolingDown() { return this.dropEventsIfCoolingDown(1); } dropEventsIfCoolingDown(count) { if (!this.shouldDropEvents()) { return false; } this.droppedEventsDuringCooldown += count; return true; } reset() { const droppedEventsDuringCooldown = this.droppedEventsDuringCooldown; this.failureCount = 0; this.cooldownUntilMs = 0; this.droppedEventsDuringCooldown = 0; return droppedEventsDuringCooldown; } recordFailure(args) { this.failureCount++; const droppedEventsDuringCooldown = this.droppedEventsDuringCooldown; this.droppedEventsDuringCooldown = 0; const targetCooldownMs = Math.min(AUTH_COOLDOWN_BASE_MS * Math.pow(2, this.failureCount - 1), AUTH_COOLDOWN_MAX_MS); const jitterMs = Math.floor(targetCooldownMs * AUTH_COOLDOWN_JITTER_RATIO * (Math.random() - 0.5) * 2); const cooldownMs = Math.max(AUTH_COOLDOWN_BASE_MS, targetCooldownMs + jitterMs); const cooldownSeconds = Math.ceil(cooldownMs / 1e3); this.cooldownUntilMs = Date.now() + cooldownMs; this.getLogger().warn( `${this.exporterName} received an authentication failure; pausing uploads for ${cooldownSeconds}s`, { status: args.status, failedSignals: args.failedSignals, droppedBatchSize: args.droppedBatchSize, droppedEventsDuringCooldown, authFailureCount: this.failureCount, cooldownMs } ); } }; // src/exporters/cloud.ts var SIGNAL_PUBLISH_SUFFIXES = { traces: "/spans/publish", logs: "/logs/publish", metrics: "/metrics/publish", scores: "/scores/publish", feedback: "/feedback/publish" }; var DEFAULT_CLOUD_SPAN_FILTER = (span) => span.type !== observability.SpanType.MODEL_CHUNK; var SIGNAL_PUBLISH_SEGMENTS = { traces: "spans", logs: "logs", metrics: "metrics", scores: "scores", feedback: "feedback" }; function trimTrailingSlashes(value) { let end = value.length; while (end > 0 && value.charCodeAt(end - 1) === 47) { end--; } return end === value.length ? value : value.slice(0, end); } function createInvalidEndpointError(endpoint, text, cause) { return new error.MastraError( { id: `CLOUD_EXPORTER_INVALID_ENDPOINT`, text, domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER, details: { endpoint } }, cause ); } var VALID_PROJECT_ID = /^[a-zA-Z0-9_-]+$/; function createInvalidProjectIdError(projectId) { return new error.MastraError({ id: `CLOUD_EXPORTER_INVALID_PROJECT_ID`, text: "CloudExporter projectId must only contain letters, numbers, hyphens, and underscores.", domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER, details: { projectId } }); } function resolveBaseEndpoint(baseEndpoint) { const normalizedEndpoint = trimTrailingSlashes(baseEndpoint); const invalidText = 'CloudExporter endpoint must be a base origin like "https://collector.example.com" with no path, search, or hash.'; try { const parsedEndpoint = new URL(normalizedEndpoint); if (parsedEndpoint.pathname !== "/" || parsedEndpoint.search || parsedEndpoint.hash) { throw createInvalidEndpointError(baseEndpoint, invalidText); } return trimTrailingSlashes(parsedEndpoint.origin); } catch (error$1) { if (error$1 instanceof error.MastraError) { throw error$1; } throw createInvalidEndpointError(baseEndpoint, invalidText, error$1); } } function buildSignalPath(signal, projectId) { const signalSegment = SIGNAL_PUBLISH_SEGMENTS[signal]; if (!projectId) { return `/ai/${signalSegment}/publish`; } return `/projects/${projectId}/ai/${signalSegment}/publish`; } function buildSignalEndpoint(baseEndpoint, signal, projectId) { return `${baseEndpoint}${buildSignalPath(signal, projectId)}`; } function resolveExplicitSignalEndpoint(signal, endpoint, projectId) { const normalizedEndpoint = trimTrailingSlashes(endpoint); const invalidText = `CloudExporter ${signal}Endpoint must be a base origin like "https://collector.example.com" or a full ${signal} publish URL ending in "${SIGNAL_PUBLISH_SUFFIXES[signal]}".`; try { const parsedEndpoint = new URL(normalizedEndpoint); if (parsedEndpoint.search || parsedEndpoint.hash) { throw createInvalidEndpointError(endpoint, invalidText); } const normalizedOrigin = trimTrailingSlashes(parsedEndpoint.origin); const normalizedPathname = trimTrailingSlashes(parsedEndpoint.pathname); if (!normalizedPathname || normalizedPathname === "/") { return buildSignalEndpoint(normalizedOrigin, signal, projectId); } if (normalizedPathname.endsWith(SIGNAL_PUBLISH_SUFFIXES[signal])) { return `${normalizedOrigin}${normalizedPathname}`; } throw createInvalidEndpointError(endpoint, invalidText); } catch (error$1) { if (error$1 instanceof error.MastraError) { throw error$1; } throw createInvalidEndpointError(endpoint, invalidText, error$1); } } function deriveSignalEndpointFromTracesEndpoint(signal, tracesEndpoint) { if (signal === "traces") { return tracesEndpoint; } const normalizedTracesEndpoint = trimTrailingSlashes(tracesEndpoint); const invalidText = 'CloudExporter tracesEndpoint must be a base origin like "https://collector.example.com" or a full traces publish URL ending in "/spans/publish".'; try { const parsedEndpoint = new URL(normalizedTracesEndpoint); const normalizedOrigin = trimTrailingSlashes(parsedEndpoint.origin); const normalizedPathname = trimTrailingSlashes(parsedEndpoint.pathname); if (!normalizedPathname.endsWith(SIGNAL_PUBLISH_SUFFIXES.traces)) { throw createInvalidEndpointError(tracesEndpoint, invalidText); } const basePath = normalizedPathname.slice(0, -SIGNAL_PUBLISH_SUFFIXES.traces.length); return `${normalizedOrigin}${basePath}${SIGNAL_PUBLISH_SUFFIXES[signal]}`; } catch (error$1) { if (error$1 instanceof error.MastraError) { throw error$1; } throw createInvalidEndpointError(tracesEndpoint, invalidText, error$1); } } var CloudExporter = class extends BaseExporter { name = "mastra-cloud-observability-exporter"; cloudConfig; authFailureCooldown; buffer; flushTimer = null; inFlightFlushes = /* @__PURE__ */ new Set(); constructor(config = {}) { super(config); if (config.projectId !== void 0 && !VALID_PROJECT_ID.test(config.projectId)) { throw createInvalidProjectIdError(config.projectId); } const accessToken = config.accessToken ?? process.env.MASTRA_CLOUD_ACCESS_TOKEN; const rawProjectId = config.projectId ?? process.env.MASTRA_PROJECT_ID; const projectId = rawProjectId && VALID_PROJECT_ID.test(rawProjectId) ? rawProjectId : void 0; if (!accessToken) { this.setDisabled("MASTRA_CLOUD_ACCESS_TOKEN environment variable not set.", "debug"); } const tracesEndpointOverride = config.tracesEndpoint ?? process.env.MASTRA_CLOUD_TRACES_ENDPOINT; let baseEndpoint; let tracesEndpoint; if (tracesEndpointOverride) { tracesEndpoint = resolveExplicitSignalEndpoint("traces", tracesEndpointOverride, projectId); } else { baseEndpoint = resolveBaseEndpoint(config.endpoint ?? "https://observability.mastra.ai"); tracesEndpoint = buildSignalEndpoint(baseEndpoint, "traces", projectId); } const resolveConfiguredSignalEndpoint = (signal, explicitEndpoint) => { if (explicitEndpoint) { return resolveExplicitSignalEndpoint(signal, explicitEndpoint, projectId); } if (tracesEndpointOverride) { return deriveSignalEndpointFromTracesEndpoint(signal, tracesEndpoint); } return buildSignalEndpoint(baseEndpoint, signal, projectId); }; this.cloudConfig = { logger: this.logger, logLevel: config.logLevel ?? logger.LogLevel.INFO, maxBatchSize: config.maxBatchSize ?? 1e3, maxBatchWaitMs: config.maxBatchWaitMs ?? 5e3, maxRetries: config.maxRetries ?? 3, accessToken: accessToken || "", tracesEndpoint, logsEndpoint: resolveConfiguredSignalEndpoint("logs", config.logsEndpoint), metricsEndpoint: resolveConfiguredSignalEndpoint("metrics", config.metricsEndpoint), scoresEndpoint: resolveConfiguredSignalEndpoint("scores", config.scoresEndpoint), feedbackEndpoint: resolveConfiguredSignalEndpoint("feedback", config.feedbackEndpoint) }; this.authFailureCooldown = new AuthFailureCooldown("CloudExporter", () => this.logger); this.buffer = { spans: [], logs: [], metrics: [], scores: [], feedback: [], totalSize: 0 }; } async _exportTracingEvent(event) { if (event.type !== observability.TracingEventType.SPAN_ENDED) { return; } if (!DEFAULT_CLOUD_SPAN_FILTER(event.exportedSpan)) { return; } if (this.authFailureCooldown.dropEventIfCoolingDown()) { return; } this.addToBuffer(event); await this.handleBufferedEvent(); } async onLogEvent(event) { if (this.isDisabled) { return; } if (this.authFailureCooldown.dropEventIfCoolingDown()) { return; } this.addLogToBuffer(event); await this.handleBufferedEvent(); } async onMetricEvent(event) { if (this.isDisabled) { return; } if (this.authFailureCooldown.dropEventIfCoolingDown()) { return; } this.addMetricToBuffer(event); await this.handleBufferedEvent(); } async onScoreEvent(event) { if (this.isDisabled) { return; } if (this.authFailureCooldown.dropEventIfCoolingDown()) { return; } this.addScoreToBuffer(event); await this.handleBufferedEvent(); } async onFeedbackEvent(event) { if (this.isDisabled) { return; } if (this.authFailureCooldown.dropEventIfCoolingDown()) { return; } this.addFeedbackToBuffer(event); await this.handleBufferedEvent(); } addToBuffer(event) { this.markBufferStart(); const spanRecord = this.formatSpan(event.exportedSpan); this.buffer.spans.push(spanRecord); this.buffer.totalSize++; } addLogToBuffer(event) { this.markBufferStart(); this.buffer.logs.push(this.formatLog(event.log)); this.buffer.totalSize++; } addMetricToBuffer(event) { this.markBufferStart(); this.buffer.metrics.push(this.formatMetric(event.metric)); this.buffer.totalSize++; } addScoreToBuffer(event) { this.markBufferStart(); this.buffer.scores.push(this.formatScore(event.score)); this.buffer.totalSize++; } addFeedbackToBuffer(event) { this.markBufferStart(); this.buffer.feedback.push(this.formatFeedback(event.feedback)); this.buffer.totalSize++; } markBufferStart() { if (this.buffer.totalSize === 0) { this.buffer.firstEventTime = /* @__PURE__ */ new Date(); } } formatSpan(span) { const spanRecord = { ...span, spanId: span.id, spanType: span.type, startedAt: span.startTime, endedAt: span.endTime ?? null, error: span.errorInfo ?? null, createdAt: /* @__PURE__ */ new Date(), updatedAt: null }; return spanRecord; } formatLog(log) { return { ...log }; } formatMetric(metric) { return { ...metric }; } formatScore(score) { return { ...score }; } formatFeedback(feedback) { return { ...feedback }; } async handleBufferedEvent() { if (this.shouldFlush()) { void this.flush().catch((error) => { this.logger.error("Batch flush failed", { error: error instanceof Error ? error.message : String(error) }); }); } else if (this.buffer.totalSize === 1) { this.scheduleFlush(); } } shouldFlush() { if (this.buffer.totalSize >= this.cloudConfig.maxBatchSize) { return true; } if (this.buffer.firstEventTime && this.buffer.totalSize > 0) { const elapsed = Date.now() - this.buffer.firstEventTime.getTime(); if (elapsed >= this.cloudConfig.maxBatchWaitMs) { return true; } } return false; } scheduleFlush() { if (this.flushTimer) { clearTimeout(this.flushTimer); } this.flushTimer = setTimeout(() => { void this.flush().catch((error$1) => { const mastraError = new error.MastraError( { id: `CLOUD_EXPORTER_FAILED_TO_SCHEDULE_FLUSH`, domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER }, error$1 ); this.logger.trackException(mastraError); this.logger.error("Scheduled flush failed", mastraError); }); }, this.cloudConfig.maxBatchWaitMs); } async flushBuffer() { if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; } if (this.buffer.totalSize === 0) { return; } if (this.authFailureCooldown.dropEventsIfCoolingDown(this.buffer.totalSize)) { this.resetBuffer(); return; } const startTime = Date.now(); const spansCopy = [...this.buffer.spans]; const logsCopy = [...this.buffer.logs]; const metricsCopy = [...this.buffer.metrics]; const scoresCopy = [...this.buffer.scores]; const feedbackCopy = [...this.buffer.feedback]; const batchSize = this.buffer.totalSize; const flushReason = this.buffer.totalSize >= this.cloudConfig.maxBatchSize ? "size" : "time"; this.resetBuffer(); const results = await Promise.all([ this.flushSignalBatch("traces", spansCopy), this.flushSignalBatch("logs", logsCopy), this.flushSignalBatch("metrics", metricsCopy), this.flushSignalBatch("scores", scoresCopy), this.flushSignalBatch("feedback", feedbackCopy) ]); const failedSignals = results.filter((result) => !result.succeeded).map((result) => result.signal); const authFailure = results.find((result) => result.authFailureStatus !== void 0); const elapsed = Date.now() - startTime; if (failedSignals.length === 0) { const droppedEventsDuringAuthCooldown = this.authFailureCooldown.reset(); const logData = { batchSize, flushReason, durationMs: elapsed }; if (droppedEventsDuringAuthCooldown > 0) { logData.droppedEventsDuringAuthCooldown = droppedEventsDuringAuthCooldown; } this.logger.debug("Batch flushed successfully", logData); return; } if (authFailure?.authFailureStatus !== void 0) { this.authFailureCooldown.recordFailure({ status: authFailure.authFailureStatus, failedSignals, droppedBatchSize: batchSize }); } this.logger.warn("Batch flush completed with dropped signal batches", { batchSize, flushReason, durationMs: elapsed, failedSignals }); } /** * Uploads a signal batch to the configured cloud API using fetchWithRetry. */ async batchUpload(signal, records) { const headers = { Authorization: `Bearer ${this.cloudConfig.accessToken}`, "Content-Type": "application/json" }; const endpointMap = { traces: this.cloudConfig.tracesEndpoint, logs: this.cloudConfig.logsEndpoint, metrics: this.cloudConfig.metricsEndpoint, scores: this.cloudConfig.scoresEndpoint, feedback: this.cloudConfig.feedbackEndpoint }; const options = { method: "POST", headers, body: JSON.stringify({ [SIGNAL_PUBLISH_SEGMENTS[signal]]: records }) }; await fetchWithAuthFailureHandling(endpointMap[signal], options, this.cloudConfig.maxRetries); } async flushSignalBatch(signal, records) { if (records.length === 0) { return { signal, succeeded: true }; } try { await this.batchUpload(signal, records); return { signal, succeeded: true }; } catch (error$1) { if (isAuthFailureError(error$1)) { return { signal, succeeded: false, authFailureStatus: error$1.status }; } const errorId = `CLOUD_EXPORTER_FAILED_TO_BATCH_UPLOAD_${signal.toUpperCase()}`; const mastraError = new error.MastraError( { id: errorId, domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER, details: { signal, droppedBatchSize: records.length } }, error$1 ); this.logger.trackException(mastraError); this.logger.error("Batch upload failed after all retries, dropping batch", mastraError); return { signal, succeeded: false }; } } resetBuffer() { this.buffer.spans = []; this.buffer.logs = []; this.buffer.metrics = []; this.buffer.scores = []; this.buffer.feedback = []; this.buffer.firstEventTime = void 0; this.buffer.totalSize = 0; } /** * Force flush any buffered spans without shutting down the exporter. * This is useful in serverless environments where you need to ensure spans * are exported before the runtime instance is terminated. */ async flush() { if (this.isDisabled) { return; } while (this.buffer.totalSize > 0 || this.inFlightFlushes.size > 0) { if (this.buffer.totalSize > 0) { this.logger.debug("Flushing buffered events", { bufferedEvents: this.buffer.totalSize }); const flushPromise = this.flushBuffer(); this.inFlightFlushes.add(flushPromise); try { await flushPromise; } finally { this.inFlightFlushes.delete(flushPromise); } continue; } await Promise.allSettled([...this.inFlightFlushes]); } } async shutdown() { if (this.isDisabled) { return; } if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; } try { await this.flush(); } catch (error$1) { const mastraError = new error.MastraError( { id: `CLOUD_EXPORTER_FAILED_TO_FLUSH_REMAINING_EVENTS_DURING_SHUTDOWN`, domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER, details: { remainingEvents: this.buffer.totalSize } }, error$1 ); this.logger.trackException(mastraError); this.logger.error("Failed to flush remaining events during shutdown", mastraError); } this.logger.info("CloudExporter shutdown complete"); } }; var ConsoleExporter = class extends BaseExporter { name = "tracing-console-exporter"; constructor(config = {}) { super(config); } async _exportTracingEvent(event) { const span = event.exportedSpan; const formatAttributes = (attributes) => { try { return JSON.stringify(attributes, null, 2); } catch (error) { const errMsg = error instanceof Error ? error.message : "Unknown formatting error"; return `[Unable to serialize attributes: ${errMsg}]`; } }; const formatDuration = (startTime, endTime) => { if (!endTime) return "N/A"; const duration = endTime.getTime() - startTime.getTime(); return `${duration}ms`; }; switch (event.type) { case observability.TracingEventType.SPAN_STARTED: this.logger.info(`\u{1F680} SPAN_STARTED`); this.logger.info(` Type: ${span.type}`); this.logger.info(` Name: ${span.name}`); this.logger.info(` ID: ${span.id}`); this.logger.info(` Trace ID: ${span.traceId}`); if (span.input !== void 0) { this.logger.info(` Input: ${formatAttributes(span.input)}`); } this.logger.info(` Attributes: ${formatAttributes(span.attributes)}`); this.logger.info("\u2500".repeat(80)); break; case observability.TracingEventType.SPAN_ENDED: const duration = formatDuration(span.startTime, span.endTime); this.logger.info(`\u2705 SPAN_ENDED`); this.logger.info(` Type: ${span.type}`); this.logger.info(` Name: ${span.name}`); this.logger.info(` ID: ${span.id}`); this.logger.info(` Duration: ${duration}`); this.logger.info(` Trace ID: ${span.traceId}`); if (span.input !== void 0) { this.logger.info(` Input: ${formatAttributes(span.input)}`); } if (span.output !== void 0) { this.logger.info(` Output: ${formatAttributes(span.output)}`); } if (span.errorInfo) { this.logger.info(` Error: ${formatAttributes(span.errorInfo)}`); } this.logger.info(` Attributes: ${formatAttributes(span.attributes)}`); this.logger.info("\u2500".repeat(80)); break; case observability.TracingEventType.SPAN_UPDATED: this.logger.info(`\u{1F4DD} SPAN_UPDATED`); this.logger.info(` Type: ${span.type}`); this.logger.info(` Name: ${span.name}`); this.logger.info(` ID: ${span.id}`); this.logger.info(` Trace ID: ${span.traceId}`); if (span.input !== void 0) { this.logger.info(` Input: ${formatAttributes(span.input)}`); } if (span.output !== void 0) { this.logger.info(` Output: ${formatAttributes(span.output)}`); } if (span.errorInfo) { this.logger.info(` Error: ${formatAttributes(span.errorInfo)}`); } this.logger.info(` Updated Attributes: ${formatAttributes(span.attributes)}`); this.logger.info("\u2500".repeat(80)); break; default: this.logger.warn(`Tracing event type not implemented: ${event.type}`); } } async shutdown() { this.logger.info("ConsoleExporter shutdown"); } }; var EventBuffer = class { #preInit = []; #creates = []; #updates = []; #allCreatedSpans = /* @__PURE__ */ new Set(); #firstEventTime; #storageStrategy; #maxRetries; constructor(args) { this.#maxRetries = args.maxRetries; } /** Initialize with a storage strategy and replay any pre-init events. */ init(args) { if (!this.#storageStrategy) { this.#storageStrategy = args.strategy; for (const event of this.#preInit) { this.addEvent(event); } this.#preInit = []; } } /** Clear the create and update buffers and reset the event timer. */ reset() { this.#creates = []; this.#updates = []; this.#firstEventTime = void 0; } setFirstEventTime() { if (!this.#firstEventTime) { this.#firstEventTime = /* @__PURE__ */ new Date(); } } pushCreate(event) { this.setFirstEventTime(); this.#creates.push({ ...event, retryCount: 0 }); } pushUpdate(event) { this.setFirstEventTime(); this.#updates.push({ ...event, retryCount: 0 }); } /** Route an event to the create or update buffer based on its type and the storage strategy. */ addEvent(event) { if (!this.#storageStrategy) { this.#preInit.push({ ...event, retryCount: 0 }); return; } switch (event.type) { case observability.TracingEventType.SPAN_STARTED: switch (this.#storageStrategy) { case "realtime": case "event-sourced": case "batch-with-updates": this.pushCreate(event); break; } break; case observability.TracingEventType.SPAN_UPDATED: switch (this.#storageStrategy) { case "realtime": case "batch-with-updates": this.pushUpdate(event); break; } break; case observability.TracingEventType.SPAN_ENDED: if (event.exportedSpan.isEvent) { this.pushCreate(event); } else { switch (this.#storageStrategy) { case "realtime": case "batch-with-updates": this.pushUpdate(event); break; default: this.pushCreate(event); break; } } break; default: this.pushCreate(event); break; } } /** Re-add failed create events to the buffer, returning events that exceed max retries. */ reAddCreates(events) { const retryable = []; const dropped = []; for (const e of events) { if (++e.retryCount <= this.#maxRetries) { retryable.push(e); } else { dropped.push(e); } } if (retryable.length > 0) { this.setFirstEventTime(); this.#creates.push(...retryable); } return dropped; } /** Re-add failed update events to the buffer, returning events that exceed max retries. */ reAddUpdates(events) { const retryable = []; const dropped = []; for (const e of events) { if (++e.retryCount <= this.#maxRetries) { retryable.push(e); } else { dropped.push(e); } } if (retryable.length > 0) { this.setFirstEventTime(); this.#updates.push(...retryable); } return dropped; } /** Snapshot of buffered create events. */ get creates() { return [...this.#creates]; } /** Snapshot of buffered update events. */ get updates() { return [...this.#updates]; } /** Total number of buffered events (creates + updates). */ get totalSize() { return this.#creates.length + this.#updates.length; } /** Milliseconds since the first event was buffered in the current batch. */ get elapsed() { if (!this.#firstEventTime) { return 0; } return Date.now() - this.#firstEventTime.getTime(); } /** * Builds a unique span key for tracking */ buildSpanKey(span) { return `${span.traceId}:${span.spanId}`; } /** Track successfully created spans so updates can verify span existence before flushing. */ addCreatedSpans(args) { if (this.#storageStrategy === "event-sourced" || this.#storageStrategy === "insert-only") { return; } for (const createRecord of args.records) { if (!createRecord.isEvent) { this.#allCreatedSpans.add(this.buildSpanKey(createRecord)); } } } /** Check whether a span's create record has already been flushed to storage. */ spanExists(span) { return this.#allCreatedSpans?.has(this.buildSpanKey({ traceId: span.traceId, spanId: span.id })); } /** Remove completed spans from tracking after their SPAN_ENDED updates are flushed. */ endFinishedSpans(args) { if (this.#storageStrategy === "event-sourced" || this.#storageStrategy === "insert-only") { return; } args.records.forEach((r) => { this.#allCreatedSpans.delete(this.buildSpanKey(r)); }); } }; // src/exporters/default.ts function resolveTracingStorageStrategy(config, observabilityStorage, storageName, logger) { const observabilityStrategy = observabilityStorage.observabilityStrategy; if (config.strategy && config.strategy !== "auto") { if (observabilityStrategy.supported.includes(config.strategy)) { return config.strategy; } logger.warn("User-specified tracing strategy not supported by storage adapter, falling back to auto-selection", { userStrategy: config.strategy, storageAdapter: storageName, supportedStrategies: observabilityStrategy.supported, fallbackStrategy: observabilityStrategy.preferred }); } return observabilityStrategy.preferred; } var DefaultExporter = class extends BaseExporter { name = "mastra-default-observability-exporter"; #config; #isInitializing = false; #initPromises = /* @__PURE__ */ new Set(); #eventBuffer; #storage; #observabilityStorage; #resolvedStrategy; #flushTimer; #emitDropEvent; // Signals whose storage methods threw "not implemented" — skip on future flushes #unsupportedSignals = /* @__PURE__ */ new Set(); constructor(config = {}) { super(config); this.#config = { ...config, maxBatchSize: config.maxBatchSize ?? 1e3, maxBufferSize: config.maxBufferSize ?? 1e4, maxBatchWaitMs: config.maxBatchWaitMs ?? 5e3, maxRetries: config.maxRetries ?? 4, retryDelayMs: config.retryDelayMs ?? 500, strategy: config.strategy ?? "auto" }; this.#eventBuffer = new EventBuffer({ maxRetries: this.#config.maxRetries ?? 4 }); } /** * Initialize the exporter (called after all dependencies are ready) */ async init(options) { try { this.#isInitializing = true; this.#emitDropEvent = options.emitDropEvent; this.#storage = options.mastra?.getStorage(); if (!this.#storage) { this.logger.warn("DefaultExporter disabled: Storage not available. Traces will not be persisted."); return; } this.#observabilityStorage = await this.#storage.getStore("observability"); if (!this.#observabilityStorage) { this.logger.warn( "DefaultExporter disabled: Observability storage not available. Traces will not be persisted." ); return; } if (!this.#resolvedStrategy) { this.#resolvedStrategy = resolveTracingStorageStrategy( this.#config, this.#observabilityStorage, this.#storage.constructor.name, this.logger ); this.logger.debug("tracing storage exporter initialized", { strategy: this.#resolvedStrategy, source: this.#config.strategy !== "auto" ? "user" : "auto", storageAdapter: this.#storage.constructor.name, maxBatchSize: this.#config.maxBatchSize, maxBatchWaitMs: this.#config.maxBatchWaitMs }); } if (this.#resolvedStrategy) { this.#eventBuffer.init({ strategy: this.#resolvedStrategy }); } } finally { this.#isInitializing = false; this.#initPromises.forEach((resolve) => { resolve(); }); this.#initPromises.clear(); } } /** * Checks if buffer should be flushed based on size or time triggers */ shouldFlush() { if (this.#resolvedStrategy === "realtime") { return true; } if (this.#eventBuffer.totalSize >= this.#config.maxBufferSize) { return true; } if (this.#eventBuffer.totalSize >= this.#config.maxBatchSize) { return true; } if (this.#eventBuffer.totalSize > 0) { if (this.#eventBuffer.elapsed >= this.#config.maxBatchWaitMs) { return true; } } return false; } /** * Schedules a flush using setTimeout */ scheduleFlush() { if (this.#flushTimer) { clearTimeout(this.#flushTimer); } this.#flushTimer = setTimeout(() => { this.flushBuffer().catch((error) => { this.logger.error("Scheduled flush failed", { error: error instanceof Error ? error.message : String(error) }); }); }, this.#config.maxBatchWaitMs); } /** * Checks flush triggers and schedules/triggers flush as needed. * Called after adding any event to the buffer. * Returns the flush promise when flushing so callers can await it. */ async handleBatchedFlush() { if (this.shouldFlush()) { await this.flushBuffer(); } else if (this.#eventBuffer.totalSize === 1) { this.scheduleFlush(); } } sanitizeDropError(error$1) { if (error$1 instanceof error.MastraError) { return { id: error$1.id, domain: String(error$1.domain), message: error$1.message }; } if (error$1 instanceof Error) { return { message: error$1.message }; } return { message: String(error$1) }; } emitDrop(signal, reason, count, error) { if (count === 0) return; const dropEvent = { type: "drop", signal, reason, count, timestamp: /* @__PURE__ */ new Date(), exporterName: this.name, ...this.#observabilityStorage ? { storageName: this.#observabilityStorage.constructor.name } : {}, ...error === void 0 ? {} : { error: this.sanitizeDropError(error) } }; this.#emitDropEvent?.(dropEvent); } /** * Flush a batch of create events for a single signal type. * On "not implemented" errors, disables the signal for future flushes. * On other errors, re-adds events to the buffer for retry. */ async flushCreates(signal, events, storageCall) { if (events.length === 0) return; if (this.#unsupportedSignals.has(signal)) { this.emitDrop(signal, "unsupported-storage", events.length); return; } try { await storageCall(events); } catch (error$1) { if (error$1 instanceof error.MastraError && error$1.domain === error.ErrorDomain.MASTRA_OBSERVABILITY && error$1.id.endsWith("_NOT_IMPLEMENTED")) { this.logger.warn(error$1.message); this.#unsupportedSignals.add(signal); this.emitDrop(signal, "unsupported-storage", events.length, error$1); } else { const dropped = this.#eventBuffer.reAddCreates(events); this.emitDrop(signal, "retry-exhausted", dropped.length, error$1); } } } /** * Flush span update/end events, deferring any whose span hasn't been created yet. * When `isEnd` is true, successfully flushed spans are removed from tracking. */ async flushSpanUpdates(events, deferredUpdates, isEnd) { const deferredCountAtEntry = deferredUpdates.length; if (events.length === 0) return; if (this.#unsupportedSignals.has("tracing")) { this.emitDrop("tracing", "unsupported-storage", events.length); return; } const partials = []; for (const event of events) { const span = event.exportedSpan; if (this.#eventBuffer.spanExists(span)) { partials.push({ traceId: span.traceId, spanId: span.id, updates: storage.buildUpdateSpanRecord(span) }); } else { deferredUpdates.push(event); } } if (partials.length === 0) return; try { await this.#observabilityStorage.batchUpdateSpans({ records: partials }); if (isEnd) { this.#eventBuffer.endFinishedSpans({ records: partials }); } } catch (error$1) { if (error$1 instanceof error.MastraError && error$1.domain === error.ErrorDomain.MASTRA_OBSERVABILITY && error$1.id.endsWith("_NOT_IMPLEMENTED")) { this.logger.warn(error$1.message); this.#unsupportedSignals.add("tracing"); deferredUpdates.length = 0; this.emitDrop("tracing", "unsupported-storage", events.length + deferredCountAtEntry, error$1); } else { const newlyDeferred = deferredUpdates.length - deferredCountAtEntry; if (newlyDeferred > 0) { deferredUpdates.splice(deferredUpdates.length - newlyDeferred, newlyDeferred); } const dropped = this.#eventBuffer.reAddUpdates(events); this.emitDrop("tracing", "retry-exhausted", dropped.length, error$1); } } } /** * Flushes the current buffer to storage. * * Creates are flushed first, then their span keys are added to allCreatedSpans. * Updates are checked against allCreatedSpans — those whose span hasn't been * created yet are re-inserted into the live buffer for the next flush. * Completed spans (SPAN_ENDED) are cleaned up from allCreatedSpans after success. */ async flushBuffer() { if (!this.#observabilityStorage) { this.logger.debug("Cannot flush. Observability storage is not initialized"); return; } if (!this.#resolvedStrategy) { this.logger.debug("Cannot flush. Observability strategy is not resolved"); return; } if (this.#flushTimer) { clearTimeout(this.#flushTimer); this.#flushTimer = void 0; } if (this.#eventBuffer.totalSize === 0) { return; } const startTime = Date.now(); const batchSize = this.#eventBuffer.totalSize; const creates = this.#eventBuffer.creates; const updates = this.#eventBuffer.updates; this.#eventBuffer.reset(); const createFeedbackEvents = []; const createLogEvents = []; const createMetricEvents = []; const createScoreEvents = []; const createSpanEvents = []; const updateSpanEvents = []; const endSpanEvents = []; for (const createEvent of creates) { switch (createEvent.type) { case "feedback": createFeedbackEvents.push(createEvent); break; case "log": createLogEvents.push(createEvent); break; case "metric": createMetricEvents.push(createEvent); break; case "score": createScoreEvents.push(createEvent); break; default: createSpanEvents.push(createEvent); break; } } for (const updateEvent of updates) { switch (updateEvent.type) { case observability.TracingEventType.SPAN_UPDATED: updateSpanEvents.push(updateEvent); break; case observability.TracingEventType.SPAN_ENDED: endSpanEvents.push(updateEvent); break; } } await Promise.all([ this.flushCreates( "feedback", createFeedbackEvents, (events) => this.#observabilityStorage.batchCreateFeedback({ feedbacks: events.map((f) => storage.buildFeedbackRecord(f)) }) ), this.flushCreates( "log", createLogEvents, (events) => this.#observabilityStorage.batchCreateLogs({ logs: events.map((l) => storage.buildLogRecord(l)) }) ), this.flushCreates( "metric", createMetricEvents, (events) => this.#observabilityStorage.batchCreateMetrics({ metrics: events.map((m) => storage.buildMetricRecord(m)) }) ), this.flushCreates( "score", createScoreEvents, (events) => this.#observabilityStorage.batchCreateScores({ scores: events.map((s) => storage.buildScoreRecord(s)) }) ), this.flushCreates("tracing", createSpanEvents, async (events) => { const records = events.map((t) => storage.buildCreateSpanRecord(t.exportedSpan)); await this.#observabilityStorage.batchCreateSpans({ records }); this.#eventBuffer.addCreatedSpans({ records }); }) ]); const deferredUpdates = []; await this.flushSpanUpdates(updateSpanEvents, deferredUpdates, false); await this.flushSpanUpdates(endSpanEvents, deferredUpdates, true); if (deferredUpdates.length > 0) { if (this.#unsupportedSignals.has("tracing")) { this.emitDrop("tracing", "unsupported-storage", deferredUpdates.length); deferredUpdates.length = 0; } else { const dropped = this.#eventBuffer.reAddUpdates(deferredUpdates); this.emitDrop("tracing", "retry-exhausted", dropped.length); } } const elapsed = Date.now() - startTime; this.logger.debug("Batch flushed", { strategy: this.#resolvedStrategy, batchSize, durationMs: elapsed, deferredUpdates: deferredUpdates.length > 0 ? deferredUpdates.length : void 0 }); return; } async _exportTracingEvent(event) { await this.waitForInit(); if (!this.#observabilityStorage) { this.logger.debug("Cannot store traces. Observability storage is not initialized"); return; } this.#eventBuffer.addEvent(event); await this.handleBatchedFlush(); } /** * Resolves when an ongoing init call is finished * Doesn't wait for the caller to call init * @returns */ async waitForInit() { if (!this.#isInitializing) return; return new Promise((resolve) => { this.#initPromises.add(resolve); }); } /** * Handle metric events — buffer for batch flush. */ async onMetricEvent(event) { await this.waitForInit(); if (!this.#observabilityStorage) return; this.#eventBuffer.addEvent(event); await this.handleBatchedFlush(); } /** * Handle log events — buffer for batch flush. */ async onLogEvent(event) { await this.waitForInit(); if (!this.#observabilityStorage) return; this.#eventBuffer.addEvent(event); await this.handleBatchedFlush(); } /** * Handle score events — buffer for batch flush. */ async onScoreEvent(event) { await this.waitForInit(); if (!this.#observabilityStorage) return; this.#eventBuffer.addEvent(event); await this.handleBatchedFlush(); } /** * Handle feedback events — buffer for batch flush. */ async onFeedbackEvent(event) { await this.waitForInit(); if (!this.#observabilityStorage) return; this.#eventBuffer.addEvent(event); await this.handleBatchedFlush(); } /** * Force flush any buffered spans without shutting down the exporter. * This is useful in serverless environments where you need to ensure spans * are exported before the runtime instance is terminated. */ async flush() { if (this.#eventBuffer.totalSize > 0) { this.logger.debug("Flushing buffered events", { bufferedEvents: this.#eventBuffer.totalSize }); await this.flushBuffer(); } } async shutdown() { if (this.#flushTimer) { clearTimeout(this.#flushTimer); this.#flushTimer = void 0; } await this.flush(); this.logger.info("DefaultExporter shutdown complete"); } }; var SIGNAL_PUBLISH_SUFFIXES2 = { traces: "/spans/publish", logs: "/logs/publish", metrics: "/metrics/publish", scores: "/scores/publish", feedback: "/feedback/publish" }; var DEFAULT_PLATFORM_SPAN_FILTER = (span) => span.type !== observability.SpanType.MODEL_CHUNK; var SIGNAL_PUBLISH_SEGMENTS2 = { traces: "spans", logs: "logs", metrics: "metrics", scores: "scores", feedback: "feedback" }; function trimTrailingSlashes2(value) { let end = value.length; while (end > 0 && value.charCodeAt(end - 1) === 47) { end--; } return end === value.length ? value : value.slice(0, end); } function createInvalidEndpointError2(endpoint, text, cause) { return new error.MastraError( { id: `MASTRA_PLATFORM_EXPORTER_INVALID_ENDPOINT`, text, domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER, details: { endpoint } }, cause ); } var VALID_PROJECT_ID2 = /^[a-zA-Z0-9_-]+$/; function createInvalidProjectIdError2(projectId) { return new error.MastraError({ id: `MASTRA_PLATFORM_EXPORTER_INVALID_PROJECT_ID`, text: "MastraPlatformExporter projectId must only contain letters, numbers, hyphens, and underscores.", domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER, details: { projectId } }); } function resolveBaseEndpoint2(baseEndpoint) { const normalizedEndpoint = trimTrailingSlashes2(baseEndpoint); const invalidText = 'MastraPlatformExporter endpoint must be a base origin like "https://collector.example.com" with no path, search, or hash.'; try { const parsedEndpoint = new URL(normalizedEndpoint); if (parsedEndpoint.pathname !== "/" || parsedEndpoint.search || parsedEndpoint.hash) { throw createInvalidEndpointError2(baseEndpoint, invalidText); } return trimTrailingSlashes2(parsedEndpoint.origin); } catch (error$1) { if (error$1 instanceof error.MastraError) { throw error$1; } throw createInvalidEndpointError2(baseEndpoint, invalidText, error$1); } } function buildSignalPath2(signal, projectId) { const signalSegment = SIGNAL_PUBLISH_SEGMENTS2[signal]; if (!projectId) { return `/ai/${signalSegment}/publish`; } return `/projects/${projectId}/ai/${signalSegment}/publish`; } function buildSignalEndpoint2(baseEndpoint, signal, projectId) { return `${baseEndpoint}${buildSignalPath2(signal, projectId)}`; } function resolveExplicitSignalEndpoint2(signal, endpoint, projectId) { const normalizedEndpoint = trimTrailingSlashes2(endpoint); const invalidText = `MastraPlatformExporter ${signal}Endpoint must be a base origin like "https://collector.example.com" or a full ${signal} publish URL ending in "${SIGNAL_PUBLISH_SUFFIXES2[signal]}".`; try { const parsedEndpoint = new URL(normalizedEndpoint); if (parsedEndpoint.search || parsedEndpoint.hash) { throw createInvalidEndpointError2(endpoint, invalidText); } const normalizedOrigin = trimTrailingSlashes2(parsedEndpoint.origin); const normalizedPathname = trimTrailingSlashes2(parsedEndpoint.pathname); if (!normalizedPathname || normalizedPathname === "/") { return buildSignalEndpoint2(normalizedOrigin, signal, projectId); } if (normalizedPathname.endsWith(SIGNAL_PUBLISH_SUFFIXES2[signal])) { return `${normalizedOrigin}${normalizedPathname}`; } throw createInvalidEndpointError2(endpoint, invalidText); } catch (error$1) { if (error$1 instanceof error.MastraError) { throw error$1; } throw createInvalidEndpointError2(endpoint, invalidText, error$1); } } function deriveSignalEndpointFromTracesEndpoint2(signal, tracesEndpoint) { if (signal === "traces") { return tracesEndpoint; } const normalizedTracesEndpoint = trimTrailingSlashes2(tracesEndpoint); const invalidText = 'MastraPlatformExporter tracesEndpoint must be a base origin like "https://collector.example.com" or a full traces publish URL ending in "/spans/publish".'; try { const parsedEndpoint = new URL(normalizedTracesEndpoint); const normalizedOrigin = trimTrailingSlashes2(parsedEndpoint.origin); const normalizedPathname = trimTrailingSlashes2(parsedEndpoint.pathname); if (!normalizedPathname.endsWith(SIGNAL_PUBLISH_SUFFIXES2.traces)) { throw createInvalidEndpointError2(tracesEndpoint, invalidText); } const basePath = normalizedPathname.slice(0, -SIGNAL_PUBLISH_SUFFIXES2.traces.length); return `${normalizedOrigin}${basePath}${SIGNAL_PUBLISH_SUFFIXES2[signal]}`; } catch (error$1) { if (error$1 instanceof error.MastraError) { throw error$1; } throw createInvalidEndpointError2(tracesEndpoint, invalidText, error$1); } } var MastraPlatformExporter = class extends BaseExporter { name = "mastra-platform-exporter"; platformConfig; authFailureCooldown; buffer; flushTimer = null; inFlightFlushes = /* @__PURE__ */ new Set(); constructor(config = {}) { super(config); if (config.projectId !== void 0 && !VALID_PROJECT_ID2.test(config.projectId)) { throw createInvalidProjectIdError2(config.projectId); } const accessToken = config.accessToken || process.env.MASTRA_PLATFORM_ACCESS_TOKEN || process.env.MASTRA_CLOUD_ACCESS_TOKEN; const envProjectId = process.env.MASTRA_PROJECT_ID === "" ? void 0 : process.env.MASTRA_PROJECT_ID; const rawProjectId = config.projectId ?? envProjectId; if (rawProjectId !== void 0 && !VALID_PROJECT_ID2.test(rawProjectId)) { throw createInvalidProjectIdError2(rawProjectId); } const projectId = rawProjectId; if (!accessToken) { this.setDisabled("MASTRA_PLATFORM_ACCESS_TOKEN environment variable not set.", "debug"); } const tracesEndpointOverride = config.tracesEndpoint ?? process.env.MASTRA_CLOUD_TRACES_ENDPOINT; let baseEndpoint; let tracesEndpoint; if (tracesEndpointOverride) { tracesEndpoint = resolveExplicitSignalEndpoint2("traces", tracesEndpointOverride, projectId); } else { baseEndpoint = resolveBaseEndpoint2(config.endpoint ?? "https://observability.mastra.ai"); tracesEndpoint = buildSignalEndpoint2(baseEndpoint, "traces", projectId); } const resolveConfiguredSignalEndpoint = (signal, explicitEndpoint) => { if (explicitEndpoint) { return resolveExplicitSignalEndpoint2(signal, explicitEndpoint, projectId); } if (tracesEndpointOverride) { return deriveSignalEndpointFromTracesEndpoint2(signal, tracesEndpoint); } return buildSignalEndpoint2(baseEndpoint, signal, projectId); }; this.platformConfig = { logger: this.logger, logLevel: config.logLevel ?? logger.LogLevel.INFO, maxBatchSize: config.maxBatchSize ?? 1e3, maxBatchWaitMs: config.maxBatchWaitMs ?? 5e3, maxRetries: config.maxRetries ?? 3, accessToken: accessToken || "", tracesEndpoint, logsEndpoint: resolveConfiguredSignalEndpoint("logs", config.logsEndpoint), metricsEndpoint: resolveConfiguredSignalEndpoint("metrics", config.metricsEndpoint), scoresEndpoint: resolveConfiguredSignalEndpoint("scores", config.scoresEndpoint), feedbackEndpoint: resolveConfiguredSignalEndpoint("feedback", config.feedbackEndpoint) }; this.authFailureCooldown = new AuthFailureCooldown("MastraPlatformExporter", () => this.logger); this.buffer = { spans: [], logs: [], metrics: [], scores: [], feedback: [], totalSize: 0 }; } async _exportTracingEvent(event) { if (event.type !== observability.TracingEventType.SPAN_ENDED) { return; } if (!DEFAULT_PLATFORM_SPAN_FILTER(event.exportedSpan)) { return; } if (this.authFailureCooldown.dropEventIfCoolingDown()) { return; } this.addToBuffer(event); await this.handleBufferedEvent(); } async onLogEvent(event) { if (this.isDisabled) { return; } if (this.authFailureCooldown.dropEventIfCoolingDown()) { return; } this.addLogToBuffer(event); await this.handleBufferedEvent(); } async onMetricEvent(event) { if (this.isDisabled) { return; } if (this.authFailureCooldown.dropEventIfCoolingDown()) { return; } this.addMetricToBuffer(event); await this.handleBufferedEvent(); } async onScoreEvent(event) { if (this.isDisabled) { return; } if (this.authFailureCooldown.dropEventIfCoolingDown()) { return; } this.addScoreToBuffer(event); await this.handleBufferedEvent(); } async onFeedbackEvent(event) { if (this.isDisabled) { return; } if (this.authFailureCooldown.dropEventIfCoolingDown()) { return; } this.addFeedbackToBuffer(event); await this.handleBufferedEvent(); } addToBuffer(event) { this.markBufferStart(); const spanRecord = this.formatSpan(event.exportedSpan); this.buffer.spans.push(spanRecord); this.buffer.totalSize++; } addLogToBuffer(event) { this.markBufferStart(); this.buffer.logs.push(this.formatLog(event.log)); this.buffer.totalSize++; } addMetricToBuffer(event) { this.markBufferStart(); this.buffer.metrics.push(this.formatMetric(event.metric)); this.buffer.totalSize++; } addScoreToBuffer(event) { this.markBufferStart(); this.buffer.scores.push(this.formatScore(event.score)); this.buffer.totalSize++; } addFeedbackToBuffer(event) { this.markBufferStart(); this.buffer.feedback.push(this.formatFeedback(event.feedback)); this.buffer.totalSize++; } markBufferStart() { if (this.buffer.totalSize === 0) { this.buffer.firstEventTime = /* @__PURE__ */ new Date(); } } formatSpan(span) { const spanRecord = { ...span, spanId: span.id, spanType: span.type, startedAt: span.startTime, endedAt: span.endTime ?? null, error: span.errorInfo ?? null, createdAt: /* @__PURE__ */ new Date(), updatedAt: null }; return spanRecord; } formatLog(log) { return { ...log }; } formatMetric(metric) { return { ...metric }; } formatScore(score) { return { ...score }; } formatFeedback(feedback) { return { ...feedback }; } async handleBufferedEvent() { if (this.shouldFlush()) { void this.flush().catch((error) => { this.logger.error("Batch flush failed", { error: error instanceof Error ? error.message : String(error) }); }); } else if (this.buffer.totalSize === 1) { this.scheduleFlush(); } } shouldFlush() { if (this.buffer.totalSize >= this.platformConfig.maxBatchSize) { return true; } if (this.buffer.firstEventTime && this.buffer.totalSize > 0) { const elapsed = Date.now() - this.buffer.firstEventTime.getTime(); if (elapsed >= this.platformConfig.maxBatchWaitMs) { return true; } } return false; } scheduleFlush() { if (this.flushTimer) { clearTimeout(this.flushTimer); } this.flushTimer = setTimeout(() => { void this.flush().catch((error$1) => { const mastraError = new error.MastraError( { id: `MASTRA_PLATFORM_EXPORTER_FAILED_TO_SCHEDULE_FLUSH`, domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER }, error$1 ); this.logger.trackException(mastraError); this.logger.error("Scheduled flush failed", mastraError); }); }, this.platformConfig.maxBatchWaitMs); } async flushBuffer() { if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; } if (this.buffer.totalSize === 0) { return; } if (this.authFailureCooldown.dropEventsIfCoolingDown(this.buffer.totalSize)) { this.resetBuffer(); return; } const startTime = Date.now(); const spansCopy = [...this.buffer.spans]; const logsCopy = [...this.buffer.logs]; const metricsCopy = [...this.buffer.metrics]; const scoresCopy = [...this.buffer.scores]; const feedbackCopy = [...this.buffer.feedback]; const batchSize = this.buffer.totalSize; const flushReason = this.buffer.totalSize >= this.platformConfig.maxBatchSize ? "size" : "time"; this.resetBuffer(); const results = await Promise.all([ this.flushSignalBatch("traces", spansCopy), this.flushSignalBatch("logs", logsCopy), this.flushSignalBatch("metrics", metricsCopy), this.flushSignalBatch("scores", scoresCopy), this.flushSignalBatch("feedback", feedbackCopy) ]); const failedSignals = results.filter((result) => !result.succeeded).map((result) => result.signal); const authFailure = results.find((result) => result.authFailureStatus !== void 0); const elapsed = Date.now() - startTime; if (failedSignals.length === 0) { const droppedEventsDuringAuthCooldown = this.authFailureCooldown.reset(); const logData = { batchSize, flushReason, durationMs: elapsed }; if (droppedEventsDuringAuthCooldown > 0) { logData.droppedEventsDuringAuthCooldown = droppedEventsDuringAuthCooldown; } this.logger.debug("Batch flushed successfully", logData); return; } if (authFailure?.authFailureStatus !== void 0) { this.authFailureCooldown.recordFailure({ status: authFailure.authFailureStatus, failedSignals, droppedBatchSize: batchSize }); } this.logger.warn("Batch flush completed with dropped signal batches", { batchSize, flushReason, durationMs: elapsed, failedSignals }); } /** * Uploads a signal batch to the configured Mastra Observability API using fetchWithRetry. */ async batchUpload(signal, records) { const headers = { Authorization: `Bearer ${this.platformConfig.accessToken}`, "Content-Type": "application/json" }; const endpointMap = { traces: this.platformConfig.tracesEndpoint, logs: this.platformConfig.logsEndpoint, metrics: this.platformConfig.metricsEndpoint, scores: this.platformConfig.scoresEndpoint, feedback: this.platformConfig.feedbackEndpoint }; const options = { method: "POST", headers, body: JSON.stringify({ [SIGNAL_PUBLISH_SEGMENTS2[signal]]: records }) }; await fetchWithAuthFailureHandling(endpointMap[signal], options, this.platformConfig.maxRetries); } async flushSignalBatch(signal, records) { if (records.length === 0) { return { signal, succeeded: true }; } try { await this.batchUpload(signal, records); return { signal, succeeded: true }; } catch (error$1) { if (isAuthFailureError(error$1)) { return { signal, succeeded: false, authFailureStatus: error$1.status }; } const errorId = `MASTRA_PLATFORM_EXPORTER_FAILED_TO_BATCH_UPLOAD_${signal.toUpperCase()}`; const mastraError = new error.MastraError( { id: errorId, domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER, details: { signal, droppedBatchSize: records.length } }, error$1 ); this.logger.trackException(mastraError); this.logger.error("Batch upload failed after all retries, dropping batch", mastraError); return { signal, succeeded: false }; } } resetBuffer() { this.buffer.spans = []; this.buffer.logs = []; this.buffer.metrics = []; this.buffer.scores = []; this.buffer.feedback = []; this.buffer.firstEventTime = void 0; this.buffer.totalSize = 0; } /** * Force flush any buffered events without shutting down the exporter. * This is useful in serverless environments where you need to ensure events * are exported before the runtime instance is terminated. */ async flush() { if (this.isDisabled) { return; } while (this.buffer.totalSize > 0 || this.inFlightFlushes.size > 0) { if (this.buffer.totalSize > 0) { this.logger.debug("Flushing buffered events", { bufferedEvents: this.buffer.totalSize }); const flushPromise = this.flushBuffer(); this.inFlightFlushes.add(flushPromise); try { await flushPromise; } finally { this.inFlightFlushes.delete(flushPromise); } continue; } await Promise.allSettled([...this.inFlightFlushes]); } } async shutdown() { if (this.isDisabled) { return; } if (this.flushTimer) { clearTimeout(this.flushTimer); this.flushTimer = null; } try { await this.flush(); } catch (error$1) { const mastraError = new error.MastraError( { id: `MASTRA_PLATFORM_EXPORTER_FAILED_TO_FLUSH_REMAINING_EVENTS_DURING_SHUTDOWN`, domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER, details: { remainingEvents: this.buffer.totalSize } }, error$1 ); this.logger.trackException(mastraError); this.logger.error("Failed to flush remaining events during shutdown", mastraError); } this.logger.info("MastraPlatformExporter shutdown complete"); } }; function resolveTracingStorageStrategy2(config, observabilityStorage, storageName, logger) { const observabilityStrategy = observabilityStorage.observabilityStrategy; if (config.strategy && config.strategy !== "auto") { if (observabilityStrategy.supported.includes(config.strategy)) { return config.strategy; } logger.warn("User-specified tracing strategy not supported by storage adapter, falling back to auto-selection", { userStrategy: config.strategy, storageAdapter: storageName, supportedStrategies: observabilityStrategy.supported, fallbackStrategy: observabilityStrategy.preferred }); } return observabilityStrategy.preferred; } var MastraStorageExporter = class extends BaseExporter { name = "mastra-storage-exporter"; #config; #isInitializing = false; #initPromises = /* @__PURE__ */ new Set(); #eventBuffer; #storage; #observabilityStorage; #resolvedStrategy; #flushTimer; #emitDropEvent; // Signals whose storage methods threw "not implemented" — skip on future flushes #unsupportedSignals = /* @__PURE__ */ new Set(); constructor(config = {}) { super(config); this.#config = { ...config, maxBatchSize: config.maxBatchSize ?? 1e3, maxBufferSize: config.maxBufferSize ?? 1e4, maxBatchWaitMs: config.maxBatchWaitMs ?? 5e3, maxRetries: config.maxRetries ?? 4, retryDelayMs: config.retryDelayMs ?? 500, strategy: config.strategy ?? "auto" }; this.#eventBuffer = new EventBuffer({ maxRetries: this.#config.maxRetries ?? 4 }); } /** * Initialize the exporter (called after all dependencies are ready) */ async init(options) { try { this.#isInitializing = true; this.#emitDropEvent = options.emitDropEvent; this.#storage = options.mastra?.getStorage(); if (!this.#storage) { this.logger.warn("MastraStorageExporter disabled: Storage not available. Traces will not be persisted."); return; } this.#observabilityStorage = await this.#storage.getStore("observability"); if (!this.#observabilityStorage) { this.logger.warn( "MastraStorageExporter disabled: Observability storage not available. Traces will not be persisted." ); return; } if (!this.#resolvedStrategy) { this.#resolvedStrategy = resolveTracingStorageStrategy2( this.#config, this.#observabilityStorage, this.#storage.constructor.name, this.logger ); this.logger.debug("tracing storage exporter initialized", { strategy: this.#resolvedStrategy, source: this.#config.strategy !== "auto" ? "user" : "auto", storageAdapter: this.#storage.constructor.name, maxBatchSize: this.#config.maxBatchSize, maxBatchWaitMs: this.#config.maxBatchWaitMs }); } if (this.#resolvedStrategy) { this.#eventBuffer.init({ strategy: this.#resolvedStrategy }); } } finally { this.#isInitializing = false; this.#initPromises.forEach((resolve) => { resolve(); }); this.#initPromises.clear(); } } /** * Checks if buffer should be flushed based on size or time triggers */ shouldFlush() { if (this.#resolvedStrategy === "realtime") { return true; } if (this.#eventBuffer.totalSize >= this.#config.maxBufferSize) { return true; } if (this.#eventBuffer.totalSize >= this.#config.maxBatchSize) { return true; } if (this.#eventBuffer.totalSize > 0) { if (this.#eventBuffer.elapsed >= this.#config.maxBatchWaitMs) { return true; } } return false; } /** * Schedules a flush using setTimeout */ scheduleFlush() { if (this.#flushTimer) { clearTimeout(this.#flushTimer); } this.#flushTimer = setTimeout(() => { this.flushBuffer().catch((error) => { this.logger.error("Scheduled flush failed", { error: error instanceof Error ? error.message : String(error) }); }); }, this.#config.maxBatchWaitMs); } /** * Checks flush triggers and schedules/triggers flush as needed. * Called after adding any event to the buffer. * Returns the flush promise when flushing so callers can await it. */ async handleBatchedFlush() { if (this.shouldFlush()) { await this.flushBuffer(); } else if (this.#eventBuffer.totalSize === 1) { this.scheduleFlush(); } } sanitizeDropError(error$1) { if (error$1 instanceof error.MastraError) { return { id: error$1.id, domain: String(error$1.domain), message: error$1.message }; } if (error$1 instanceof Error) { return { message: error$1.message }; } return { message: String(error$1) }; } emitDrop(signal, reason, count, error) { if (count === 0) return; const dropEvent = { type: "drop", signal, reason, count, timestamp: /* @__PURE__ */ new Date(), exporterName: this.name, ...this.#observabilityStorage ? { storageName: this.#observabilityStorage.constructor.name } : {}, ...error === void 0 ? {} : { error: this.sanitizeDropError(error) } }; this.#emitDropEvent?.(dropEvent); } /** * Flush a batch of create events for a single signal type. * On "not implemented" errors, disables the signal for future flushes. * On other errors, re-adds events to the buffer for retry. */ async flushCreates(signal, events, storageCall) { if (events.length === 0) return; if (this.#unsupportedSignals.has(signal)) { this.emitDrop(signal, "unsupported-storage", events.length); return; } try { await storageCall(events); } catch (error$1) { if (error$1 instanceof error.MastraError && error$1.domain === error.ErrorDomain.MASTRA_OBSERVABILITY && error$1.id.endsWith("_NOT_IMPLEMENTED")) { this.logger.warn(error$1.message); this.#unsupportedSignals.add(signal); this.emitDrop(signal, "unsupported-storage", events.length, error$1); } else { const dropped = this.#eventBuffer.reAddCreates(events); this.emitDrop(signal, "retry-exhausted", dropped.length, error$1); } } } /** * Flush span update/end events, deferring any whose span hasn't been created yet. * When `isEnd` is true, successfully flushed spans are removed from tracking. */ async flushSpanUpdates(events, deferredUpdates, isEnd) { const deferredCountAtEntry = deferredUpdates.length; if (events.length === 0) return; if (this.#unsupportedSignals.has("tracing")) { this.emitDrop("tracing", "unsupported-storage", events.length); return; } const partials = []; for (const event of events) { const span = event.exportedSpan; if (this.#eventBuffer.spanExists(span)) { partials.push({ traceId: span.traceId, spanId: span.id, updates: storage.buildUpdateSpanRecord(span) }); } else { deferredUpdates.push(event); } } if (partials.length === 0) return; try { await this.#observabilityStorage.batchUpdateSpans({ records: partials }); if (isEnd) { this.#eventBuffer.endFinishedSpans({ records: partials }); } } catch (error$1) { if (error$1 instanceof error.MastraError && error$1.domain === error.ErrorDomain.MASTRA_OBSERVABILITY && error$1.id.endsWith("_NOT_IMPLEMENTED")) { this.logger.warn(error$1.message); this.#unsupportedSignals.add("tracing"); deferredUpdates.length = 0; this.emitDrop("tracing", "unsupported-storage", events.length + deferredCountAtEntry, error$1); } else { const newlyDeferred = deferredUpdates.length - deferredCountAtEntry; if (newlyDeferred > 0) { deferredUpdates.splice(deferredUpdates.length - newlyDeferred, newlyDeferred); } const dropped = this.#eventBuffer.reAddUpdates(events); this.emitDrop("tracing", "retry-exhausted", dropped.length, error$1); } } } /** * Flushes the current buffer to storage. * * Creates are flushed first, then their span keys are added to allCreatedSpans. * Updates are checked against allCreatedSpans — those whose span hasn't been * created yet are re-inserted into the live buffer for the next flush. * Completed spans (SPAN_ENDED) are cleaned up from allCreatedSpans after success. */ async flushBuffer() { if (!this.#observabilityStorage) { this.logger.debug("Cannot flush. Observability storage is not initialized"); return; } if (!this.#resolvedStrategy) { this.logger.debug("Cannot flush. Observability strategy is not resolved"); return; } if (this.#flushTimer) { clearTimeout(this.#flushTimer); this.#flushTimer = void 0; } if (this.#eventBuffer.totalSize === 0) { return; } const startTime = Date.now(); const batchSize = this.#eventBuffer.totalSize; const creates = this.#eventBuffer.creates; const updates = this.#eventBuffer.updates; this.#eventBuffer.reset(); const createFeedbackEvents = []; const createLogEvents = []; const createMetricEvents = []; const createScoreEvents = []; const createSpanEvents = []; const updateSpanEvents = []; const endSpanEvents = []; for (const createEvent of creates) { switch (createEvent.type) { case "feedback": createFeedbackEvents.push(createEvent); break; case "log": createLogEvents.push(createEvent); break; case "metric": createMetricEvents.push(createEvent); break; case "score": createScoreEvents.push(createEvent); break; default: createSpanEvents.push(createEvent); break; } } for (const updateEvent of updates) { switch (updateEvent.type) { case observability.TracingEventType.SPAN_UPDATED: updateSpanEvents.push(updateEvent); break; case observability.TracingEventType.SPAN_ENDED: endSpanEvents.push(updateEvent); break; } } await Promise.all([ this.flushCreates( "feedback", createFeedbackEvents, (events) => this.#observabilityStorage.batchCreateFeedback({ feedbacks: events.map((f) => storage.buildFeedbackRecord(f)) }) ), this.flushCreates( "log", createLogEvents, (events) => this.#observabilityStorage.batchCreateLogs({ logs: events.map((l) => storage.buildLogRecord(l)) }) ), this.flushCreates( "metric", createMetricEvents, (events) => this.#observabilityStorage.batchCreateMetrics({ metrics: events.map((m) => storage.buildMetricRecord(m)) }) ), this.flushCreates( "score", createScoreEvents, (events) => this.#observabilityStorage.batchCreateScores({ scores: events.map((s) => storage.buildScoreRecord(s)) }) ), this.flushCreates("tracing", createSpanEvents, async (events) => { const records = events.map((t) => storage.buildCreateSpanRecord(t.exportedSpan)); await this.#observabilityStorage.batchCreateSpans({ records }); this.#eventBuffer.addCreatedSpans({ records }); }) ]); const deferredUpdates = []; await this.flushSpanUpdates(updateSpanEvents, deferredUpdates, false); await this.flushSpanUpdates(endSpanEvents, deferredUpdates, true); if (deferredUpdates.length > 0) { if (this.#unsupportedSignals.has("tracing")) { this.emitDrop("tracing", "unsupported-storage", deferredUpdates.length); deferredUpdates.length = 0; } else { const dropped = this.#eventBuffer.reAddUpdates(deferredUpdates); this.emitDrop("tracing", "retry-exhausted", dropped.length); } } const elapsed = Date.now() - startTime; this.logger.debug("Batch flushed", { strategy: this.#resolvedStrategy, batchSize, durationMs: elapsed, deferredUpdates: deferredUpdates.length > 0 ? deferredUpdates.length : void 0 }); return; } async _exportTracingEvent(event) { await this.waitForInit(); if (!this.#observabilityStorage) { this.logger.debug("Cannot store traces. Observability storage is not initialized"); return; } this.#eventBuffer.addEvent(event); await this.handleBatchedFlush(); } /** * Resolves when an ongoing init call is finished * Doesn't wait for the caller to call init * @returns */ async waitForInit() { if (!this.#isInitializing) return; return new Promise((resolve) => { this.#initPromises.add(resolve); }); } /** * Handle metric events — buffer for batch flush. */ async onMetricEvent(event) { await this.waitForInit(); if (!this.#observabilityStorage) return; this.#eventBuffer.addEvent(event); await this.handleBatchedFlush(); } /** * Handle log events — buffer for batch flush. */ async onLogEvent(event) { await this.waitForInit(); if (!this.#observabilityStorage) return; this.#eventBuffer.addEvent(event); await this.handleBatchedFlush(); } /** * Handle score events — buffer for batch flush. */ async onScoreEvent(event) { await this.waitForInit(); if (!this.#observabilityStorage) return; this.#eventBuffer.addEvent(event); await this.handleBatchedFlush(); } /** * Handle feedback events — buffer for batch flush. */ async onFeedbackEvent(event) { await this.waitForInit(); if (!this.#observabilityStorage) return; this.#eventBuffer.addEvent(event); await this.handleBatchedFlush(); } /** * Force flush any buffered spans without shutting down the exporter. * This is useful in serverless environments where you need to ensure spans * are exported before the runtime instance is terminated. */ async flush() { if (this.#eventBuffer.totalSize > 0) { this.logger.debug("Flushing buffered events", { bufferedEvents: this.#eventBuffer.totalSize }); await this.flushBuffer(); } } async shutdown() { if (this.#flushTimer) { clearTimeout(this.#flushTimer); this.#flushTimer = void 0; } await this.flush(); this.logger.info("MastraStorageExporter shutdown complete"); } }; var _snapshotsDir; async function getSnapshotsDir() { if (!_snapshotsDir) { if (typeof (typeof document === 'undefined' ? require('u' + 'rl').pathToFileURL(__filename).href : (_documentCurrentScript && _documentCurrentScript.tagName.toUpperCase() === 'SCRIPT' && _documentCurrentScript.src || new URL('index.cjs', document.baseURI).href)) !== "string") { throw new Error( "Snapshot functionality requires a Node.js environment. import.meta.url is not available in this runtime." ); } const { fileURLToPath } = await import('url'); const { dirname, join } = await import('path'); const __filename = fileURLToPath((typeof document === 'undefined' ? require('u' + 'rl').pathToFileURL(__filename).href : (_documentCurrentScript && _documentCurrentScript.tagName.toUpperCase() === 'SCRIPT' && _documentCurrentScript.src || new URL('index.cjs', document.baseURI).href))); const __dirname = dirname(__filename); _snapshotsDir = join(__dirname, "..", "__snapshots__"); } return _snapshotsDir; } var TestExporter = class extends BaseExporter { name = "test-exporter"; /** All collected tracing events */ #tracingEvents = []; /** Per-span state tracking */ #spanStates = /* @__PURE__ */ new Map(); /** All collected log events */ #logEvents = []; /** All collected metric events */ #metricEvents = []; /** All collected score events */ #scoreEvents = []; /** All collected feedback events */ #feedbackEvents = []; /** Debug logs for the exporter itself */ #debugLogs = []; /** Configuration */ #config; /** Internal metrics tracking */ #internalMetrics; constructor(config = {}) { super(config); this.#config = { validateLifecycle: true, storeLogs: true, jsonIndent: 2, logMetricsOnFlush: true, ...config }; this.#internalMetrics = { startedAt: /* @__PURE__ */ new Date(), lastEventAt: null, totalEventsReceived: 0, bySignal: { tracing: 0, log: 0, metric: 0, score: 0, feedback: 0 }, flushCount: 0 }; } /** * Process incoming tracing events with lifecycle tracking */ async _exportTracingEvent(event) { const span = event.exportedSpan; const spanId = span.id; this.#trackEvent("tracing"); const logMessage = `[TestExporter] ${event.type}: ${span.type} "${span.name}" (entity: ${span.entityName ?? span.entityId ?? "unknown"}, trace: ${span.traceId.slice(-8)}, span: ${spanId.slice(-8)})`; if (this.#config.storeLogs) { this.#debugLogs.push(logMessage); } const state = this.#spanStates.get(spanId) || { hasStart: false, hasEnd: false, hasUpdate: false, events: [] }; if (this.#config.validateLifecycle) { this.#validateLifecycle(event, state, spanId); } if (event.type === observability.TracingEventType.SPAN_STARTED) { state.hasStart = true; } else if (event.type === observability.TracingEventType.SPAN_ENDED) { state.hasEnd = true; if (span.isEvent) { state.isEventSpan = true; } } else if (event.type === observability.TracingEventType.SPAN_UPDATED) { state.hasUpdate = true; } state.events.push(event); this.#spanStates.set(spanId, state); this.#tracingEvents.push(event); } // ============================================================================ // Signal Handlers (Logs, Metrics, Scores, Feedback) // ============================================================================ /** * Process incoming log events */ async onLogEvent(event) { this.#trackEvent("log"); if (this.#config.storeLogs) { const log = event.log; const traceId = log.traceId; const logMessage = `[TestExporter] log.${log.level}: "${log.message}"${traceId ? ` (trace: ${traceId.slice(-8)})` : ""}`; this.#debugLogs.push(logMessage); } this.#logEvents.push(event); } /** * Process incoming metric events */ async onMetricEvent(event) { this.#trackEvent("metric"); if (this.#config.storeLogs) { const metric = event.metric; const labelsStr = Object.entries(metric.labels).map(([k, v]) => `${k}=${v}`).join(", "); const logMessage = `[TestExporter] metric: ${metric.name}=${metric.value}${labelsStr ? ` {${labelsStr}}` : ""}`; this.#debugLogs.push(logMessage); } this.#metricEvents.push(event); } /** * Process incoming score events */ async onScoreEvent(event) { this.#trackEvent("score"); if (this.#config.storeLogs) { const score = event.score; const traceLabel = score.traceId ? score.traceId.slice(-8) : "unanchored"; const logMessage = `[TestExporter] score: ${score.scorerId}=${score.score} (trace: ${traceLabel}${score.spanId ? `, span: ${score.spanId.slice(-8)}` : ""})`; this.#debugLogs.push(logMessage); } this.#scoreEvents.push(event); } /** * Process incoming feedback events */ async onFeedbackEvent(event) { this.#trackEvent("feedback"); if (this.#config.storeLogs) { const fb = event.feedback; const traceLabel = fb.traceId ? fb.traceId.slice(-8) : "unanchored"; const feedbackSource = fb.feedbackSource ?? fb.source; const logMessage = `[TestExporter] feedback: ${fb.feedbackType} from ${feedbackSource}=${fb.value} (trace: ${traceLabel}${fb.spanId ? `, span: ${fb.spanId.slice(-8)}` : ""})`; this.#debugLogs.push(logMessage); } this.#feedbackEvents.push(event); } /** * Track an event for internal metrics */ #trackEvent(signal) { this.#internalMetrics.lastEventAt = /* @__PURE__ */ new Date(); this.#internalMetrics.totalEventsReceived++; this.#internalMetrics.bySignal[signal]++; } /** * Validate span lifecycle rules */ #validateLifecycle(event, state, spanId) { const span = event.exportedSpan; if (event.type === observability.TracingEventType.SPAN_STARTED) { if (state.hasStart) { this.logger.warn(`Span ${spanId} (${span.type} "${span.name}") started twice`); } } else if (event.type === observability.TracingEventType.SPAN_ENDED) { if (span.isEvent) { if (state.hasStart) { this.logger.warn(`Event span ${spanId} (${span.type} "${span.name}") incorrectly received SPAN_STARTED`); } if (state.hasUpdate) { this.logger.warn(`Event span ${spanId} (${span.type} "${span.name}") incorrectly received SPAN_UPDATED`); } } else { if (!state.hasStart) { this.logger.warn(`Normal span ${spanId} (${span.type} "${span.name}") ended without starting`); } } } } // ============================================================================ // Tracing Query Methods // ============================================================================ /** * Get all collected tracing events */ get events() { return [...this.#tracingEvents]; } /** * Get completed spans by SpanType (e.g., 'agent_run', 'tool_call') * * @param type - The SpanType to filter by * @returns Array of completed exported spans of the specified type */ getSpansByType(type) { return Array.from(this.#spanStates.values()).filter((state) => { if (!state.hasEnd) return false; const endEvent = state.events.find((e) => e.type === observability.TracingEventType.SPAN_ENDED); return endEvent?.exportedSpan.type === type; }).map((state) => { const endEvent = state.events.find((e) => e.type === observability.TracingEventType.SPAN_ENDED); return endEvent?.exportedSpan; }).filter((span) => span !== void 0); } /** * Get events by TracingEventType (SPAN_STARTED, SPAN_UPDATED, SPAN_ENDED) * * @param type - The TracingEventType to filter by * @returns Array of events of the specified type */ getByEventType(type) { return this.#tracingEvents.filter((e) => e.type === type); } /** * Get all events and spans for a specific trace * * @param traceId - The trace ID to filter by * @returns Object containing tracing events, final spans, plus logs/scores/feedback for the trace */ getByTraceId(traceId) { const events = this.#tracingEvents.filter((e) => e.exportedSpan.traceId === traceId); const spans = this.#getUniqueSpansFromEvents(events); const logs = this.#logEvents.filter((e) => e.log.traceId === traceId).map((e) => e.log); const scores = this.#scoreEvents.filter((e) => e.score.traceId === traceId).map((e) => e.score); const feedback = this.#feedbackEvents.filter((e) => e.feedback.traceId === traceId).map((e) => e.feedback); return { events, spans, logs, scores, feedback }; } /** * Get all events for a specific span * * @param spanId - The span ID to filter by * @returns Object containing events and final span state */ getBySpanId(spanId) { const state = this.#spanStates.get(spanId); if (!state) { return { events: [], span: void 0, state: void 0 }; } const endEvent = state.events.find((e) => e.type === observability.TracingEventType.SPAN_ENDED); const span = endEvent?.exportedSpan ?? state.events[state.events.length - 1]?.exportedSpan; return { events: state.events, span, state }; } /** * Get all unique spans (returns the final state of each span) */ getAllSpans() { return Array.from(this.#spanStates.values()).map((state) => { const endEvent = state.events.find((e) => e.type === observability.TracingEventType.SPAN_ENDED); return endEvent?.exportedSpan ?? state.events[state.events.length - 1]?.exportedSpan; }).filter((span) => span !== void 0); } /** * Get only completed spans (those that have received SPAN_ENDED) */ getCompletedSpans() { return Array.from(this.#spanStates.values()).filter((state) => state.hasEnd).map((state) => { const endEvent = state.events.find((e) => e.type === observability.TracingEventType.SPAN_ENDED); return endEvent.exportedSpan; }); } /** * Get root spans only (spans with no parent) */ getRootSpans() { return this.getAllSpans().filter((span) => span.isRootSpan); } /** * Get incomplete spans (started but not yet ended) */ getIncompleteSpans() { return Array.from(this.#spanStates.entries()).filter(([_, state]) => !state.hasEnd).map(([spanId, state]) => ({ spanId, span: state.events[0]?.exportedSpan, state: { hasStart: state.hasStart, hasUpdate: state.hasUpdate, hasEnd: state.hasEnd } })); } /** * Get unique trace IDs from all collected signals */ getTraceIds() { const traceIds = /* @__PURE__ */ new Set(); for (const event of this.#tracingEvents) { traceIds.add(event.exportedSpan.traceId); } for (const event of this.#logEvents) { if (event.log.traceId) traceIds.add(event.log.traceId); } for (const event of this.#scoreEvents) { if (event.score.traceId) { traceIds.add(event.score.traceId); } } for (const event of this.#feedbackEvents) { if (event.feedback.traceId) { traceIds.add(event.feedback.traceId); } } return Array.from(traceIds); } // ============================================================================ // Log Query Methods // ============================================================================ /** * Get all collected log events */ getLogEvents() { return [...this.#logEvents]; } /** * Get all collected logs (unwrapped from events) */ getAllLogs() { return this.#logEvents.map((e) => e.log); } /** * Get logs filtered by level */ getLogsByLevel(level) { return this.#logEvents.filter((e) => e.log.level === level).map((e) => e.log); } /** * Get logs for a specific trace */ getLogsByTraceId(traceId) { return this.#logEvents.filter((e) => e.log.traceId === traceId).map((e) => e.log); } // ============================================================================ // Metric Query Methods // ============================================================================ /** * Get all collected metric events */ getMetricEvents() { return [...this.#metricEvents]; } /** * Get all collected metrics (unwrapped from events) */ getAllMetrics() { return this.#metricEvents.map((e) => e.metric); } /** * Get metrics filtered by name */ getMetricsByName(name) { return this.#metricEvents.filter((e) => e.metric.name === name).map((e) => e.metric); } /** * @deprecated MetricType is no longer stored. Use getMetricsByName() instead. */ getMetricsByType(_metricType) { throw new Error( "getMetricsByType() has been removed: metricType is no longer stored. Use getMetricsByName(metricName) instead to filter metrics by name." ); } // ============================================================================ // Score Query Methods // ============================================================================ /** * Get all collected score events */ getScoreEvents() { return [...this.#scoreEvents]; } /** * Get all collected scores (unwrapped from events) */ getAllScores() { return this.#scoreEvents.map((e) => e.score); } /** * Get scores filtered by scorer id */ getScoresByScorer(scorerId) { return this.#scoreEvents.filter((e) => e.score.scorerId === scorerId).map((e) => e.score); } /** * Get scores for a specific trace */ getScoresByTraceId(traceId) { return this.#scoreEvents.filter((e) => e.score.traceId === traceId).map((e) => e.score); } // ============================================================================ // Feedback Query Methods // ============================================================================ /** * Get all collected feedback events */ getFeedbackEvents() { return [...this.#feedbackEvents]; } /** * Get all collected feedback (unwrapped from events) */ getAllFeedback() { return this.#feedbackEvents.map((e) => e.feedback); } /** * Get feedback filtered by type */ getFeedbackByType(feedbackType) { return this.#feedbackEvents.filter((e) => e.feedback.feedbackType === feedbackType).map((e) => e.feedback); } /** * Get feedback for a specific trace */ getFeedbackByTraceId(traceId) { return this.#feedbackEvents.filter((e) => e.feedback.traceId === traceId).map((e) => e.feedback); } // ============================================================================ // Statistics // ============================================================================ /** * Get comprehensive statistics about all collected signals */ getStatistics() { const bySpanType = {}; let completedSpans = 0; let incompleteSpans = 0; for (const state of this.#spanStates.values()) { if (state.hasEnd) { completedSpans++; const endEvent = state.events.find((e) => e.type === observability.TracingEventType.SPAN_ENDED); const spanType = endEvent?.exportedSpan.type; if (spanType) { bySpanType[spanType] = (bySpanType[spanType] || 0) + 1; } } else { incompleteSpans++; } } const logsByLevel = {}; for (const event of this.#logEvents) { const level = event.log.level; logsByLevel[level] = (logsByLevel[level] || 0) + 1; } const metricsByName = {}; for (const event of this.#metricEvents) { const mName = event.metric.name; metricsByName[mName] = (metricsByName[mName] || 0) + 1; } const scoresByScorer = {}; for (const event of this.#scoreEvents) { const scorer = event.score.scorerId; scoresByScorer[scorer] = (scoresByScorer[scorer] || 0) + 1; } const feedbackByType = {}; for (const event of this.#feedbackEvents) { const fbType = event.feedback.feedbackType; feedbackByType[fbType] = (feedbackByType[fbType] || 0) + 1; } return { totalTracingEvents: this.#tracingEvents.length, totalEvents: this.#tracingEvents.length, // deprecated alias totalSpans: this.#spanStates.size, totalTraces: this.getTraceIds().length, completedSpans, incompleteSpans, byEventType: { started: this.#tracingEvents.filter((e) => e.type === observability.TracingEventType.SPAN_STARTED).length, updated: this.#tracingEvents.filter((e) => e.type === observability.TracingEventType.SPAN_UPDATED).length, ended: this.#tracingEvents.filter((e) => e.type === observability.TracingEventType.SPAN_ENDED).length }, bySpanType, totalLogs: this.#logEvents.length, logsByLevel, totalMetrics: this.#metricEvents.length, metricsByName, totalScores: this.#scoreEvents.length, scoresByScorer, totalFeedback: this.#feedbackEvents.length, feedbackByType }; } // ============================================================================ // JSON Output // ============================================================================ /** * Serialize all collected data to JSON string * * @param options - Serialization options * @returns JSON string of all collected data */ toJSON(options) { const indent = options?.indent ?? this.#config.jsonIndent; const includeEvents = options?.includeEvents ?? true; const includeStats = options?.includeStats ?? true; const data = { spans: this.getAllSpans() }; if (this.#logEvents.length > 0) { data.logs = this.getAllLogs(); } if (this.#metricEvents.length > 0) { data.metrics = this.getAllMetrics(); } if (this.#scoreEvents.length > 0) { data.scores = this.getAllScores(); } if (this.#feedbackEvents.length > 0) { data.feedback = this.getAllFeedback(); } if (includeEvents) { data.events = this.#tracingEvents; } if (includeStats) { data.statistics = this.getStatistics(); } return JSON.stringify(data, this.#jsonReplacer, indent); } /** * Build a tree structure from spans, nesting children under their parents * * @returns Array of root span tree nodes (spans with no parent) */ buildSpanTree() { const spans = this.getAllSpans(); const nodeMap = /* @__PURE__ */ new Map(); const roots = []; for (const span of spans) { nodeMap.set(span.id, { span, children: [] }); } for (const span of spans) { const node = nodeMap.get(span.id); if (span.parentSpanId && nodeMap.has(span.parentSpanId)) { nodeMap.get(span.parentSpanId).children.push(node); } else { roots.push(node); } } const sortChildren = (node) => { node.children.sort((a, b) => new Date(a.span.startTime).getTime() - new Date(b.span.startTime).getTime()); node.children.forEach(sortChildren); }; roots.forEach(sortChildren); return roots; } /** * Serialize spans as a tree structure to JSON string * * @param options - Serialization options * @returns JSON string with spans nested in tree format */ toTreeJSON(options) { const indent = options?.indent ?? this.#config.jsonIndent; const includeStats = options?.includeStats ?? true; const data = { tree: this.buildSpanTree() }; if (includeStats) { data.statistics = this.getStatistics(); } return JSON.stringify(data, this.#jsonReplacer, indent); } /** * Build a normalized tree structure suitable for snapshot testing. * * Normalizations applied: * - Span IDs replaced with stable placeholders (, , etc.) * - Trace IDs replaced with stable placeholders (, , etc.) * - parentSpanId replaced with normalized parent ID * - Timestamps replaced with durationMs (or null if not ended) * - Empty children arrays are omitted * * @returns Array of normalized root tree nodes */ buildNormalizedTree() { const tree = this.buildSpanTree(); const spanIdMap = /* @__PURE__ */ new Map(); const traceIdMap = /* @__PURE__ */ new Map(); const uuidMapsByKey = /* @__PURE__ */ new Map(); const uuidCountersByKey = /* @__PURE__ */ new Map(); let spanIdCounter = 1; let traceIdCounter = 1; const uuidRegex = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; const hexId32Regex = /^[0-9a-f]{32}$/i; const prefixedUuidRegex = /^([a-z_]+)_([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})$/i; const embeddedPrefixedUuidTest = /([a-z_]+)_([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})/i; const embeddedPrefixedUuidRegex = /([a-z_]+)_([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})/gi; const normalizeUuid = (uuid, key) => { if (!uuidMapsByKey.has(key)) { uuidMapsByKey.set(key, /* @__PURE__ */ new Map()); uuidCountersByKey.set(key, 1); } const keyMap = uuidMapsByKey.get(key); if (!keyMap.has(uuid)) { const counter = uuidCountersByKey.get(key); keyMap.set(uuid, `<${key}-${counter}>`); uuidCountersByKey.set(key, counter + 1); } return keyMap.get(uuid); }; const normalizeValue = (value, key) => { if (value instanceof Date) { return ""; } if (key === "createdAt" && typeof value === "number") { return ""; } if (typeof value === "string") { if (key === "traceId" && (uuidRegex.test(value) || hexId32Regex.test(value))) { if (!traceIdMap.has(value)) { traceIdMap.set(value, ``); } return traceIdMap.get(value); } if (uuidRegex.test(value)) { return normalizeUuid(value, key ?? "uuid"); } const prefixMatch = prefixedUuidRegex.exec(value); if (prefixMatch && prefixMatch[1] && prefixMatch[2]) { const prefix = prefixMatch[1]; const uuid = prefixMatch[2]; return `${prefix}_${normalizeUuid(uuid, prefix)}`; } if (embeddedPrefixedUuidTest.test(value)) { return value.replace(embeddedPrefixedUuidRegex, (_match, prefix, uuid) => { return `${prefix}_${normalizeUuid(uuid, prefix)}`; }); } } if (Array.isArray(value)) { return value.map((v) => normalizeValue(v, key)); } if (value && typeof value === "object") { const normalized = {}; for (const [k, v] of Object.entries(value)) { if (key === "providerOptions" && k === "mastra" && v && typeof v === "object") { const mastraOptions = v; const remainingMastraOptions = Object.fromEntries( Object.entries(mastraOptions).filter(([mastraKey]) => mastraKey !== "createdAt") ); if (Object.keys(remainingMastraOptions).length > 0) { normalized[k] = normalizeValue(remainingMastraOptions, k); } continue; } const normalizedValue = normalizeValue(v, k); if (normalizedValue !== void 0) { normalized[k] = normalizedValue; } } if (key === "providerOptions" && Object.keys(normalized).length === 0) { return void 0; } return normalized; } return value; }; const assignIds = (nodes) => { for (const node of nodes) { spanIdMap.set(node.span.id, ``); if (!traceIdMap.has(node.span.traceId)) { traceIdMap.set(node.span.traceId, ``); } assignIds(node.children); } }; assignIds(tree); const normalizeNode = (node) => { const span = node.span; const completed = span.endTime !== void 0 && span.endTime !== null; const normalizedSpan = { id: spanIdMap.get(span.id), traceId: traceIdMap.get(span.traceId), name: normalizeValue(span.name, "name"), type: span.type, completed, isEvent: span.isEvent, isRootSpan: span.isRootSpan }; if (span.parentSpanId && spanIdMap.has(span.parentSpanId)) { normalizedSpan.parentId = spanIdMap.get(span.parentSpanId); } if (span.entityType) { normalizedSpan.entityType = span.entityType; } if (span.entityId) { normalizedSpan.entityId = normalizeValue(span.entityId, "entityId"); } if (span.attributes && Object.keys(span.attributes).length > 0) { normalizedSpan.attributes = normalizeValue(span.attributes); } if (span.metadata && Object.keys(span.metadata).length > 0) { normalizedSpan.metadata = normalizeValue(span.metadata); } if (span.input !== void 0) { normalizedSpan.input = normalizeValue(span.input); } if (span.output !== void 0) { normalizedSpan.output = normalizeValue(span.output); } if (span.errorInfo) { normalizedSpan.errorInfo = normalizeValue(span.errorInfo); } if (span.tags && span.tags.length > 0) { normalizedSpan.tags = span.tags; } const result = { span: normalizedSpan }; if (node.children.length > 0) { result.children = node.children.map(normalizeNode); } return result; }; return tree.map(normalizeNode); } /** * Generate an ASCII tree structure graph for debugging. * Shows span type and name in a hierarchical format. * * @param nodes - Normalized tree nodes (defaults to current normalized tree) * @returns Array of strings representing the tree structure * * @example * ``` * agent_run: "agent run: 'test-agent'" * ├── processor_run: "input processor: validator" * │ └── agent_run: "agent run: 'validator-agent'" * └── model_generation: "llm: 'mock-model-id'" * ``` */ generateStructureGraph(nodes) { const tree = nodes ?? this.buildNormalizedTree(); const lines = []; const buildLines = (node, prefix, isLast, isRoot) => { const connector = isRoot ? "" : isLast ? "\u2514\u2500\u2500 " : "\u251C\u2500\u2500 "; const line = `${prefix}${connector}${node.span.type}: "${node.span.name}"`; lines.push(line); const children = node.children ?? []; const childPrefix = isRoot ? "" : prefix + (isLast ? " " : "\u2502 "); children.forEach((child, index) => { const childIsLast = index === children.length - 1; buildLines(child, childPrefix, childIsLast, false); }); }; tree.forEach((rootNode, index) => { if (index > 0) { lines.push(""); } buildLines(rootNode, "", true, true); }); return lines; } /** * Serialize spans as a normalized tree structure for snapshot testing. * Includes a __structure__ field with an ASCII tree graph for readability. * * @param options - Serialization options * @returns JSON string with normalized spans in tree format */ toNormalizedTreeJSON(options) { const indent = options?.indent ?? this.#config.jsonIndent; const includeStructure = options?.includeStructure ?? true; const normalizedTree = this.buildNormalizedTree(); if (includeStructure) { const structureGraph = this.generateStructureGraph(normalizedTree); const data = { __structure__: structureGraph, spans: normalizedTree }; return JSON.stringify(data, null, indent); } return JSON.stringify(normalizedTree, null, indent); } /** * Write collected data to a JSON file * * @param filePath - Path to write the JSON file * @param options - Serialization options */ async writeToFile(filePath, options) { const format = options?.format ?? "flat"; let json; if (format === "normalized") { json = this.toNormalizedTreeJSON({ indent: options?.indent }); } else if (format === "tree") { json = this.toTreeJSON({ indent: options?.indent, includeStats: options?.includeStats }); } else { json = this.toJSON(options); } const { writeFile } = await import('fs/promises'); await writeFile(filePath, json, "utf-8"); this.logger.info(`TestExporter: wrote ${this.#tracingEvents.length} tracing events to ${filePath}`); } /** * Assert that the current normalized tree matches a snapshot file. * Throws an error with a diff if they don't match. * * The snapshot format includes: * - `__structure__`: ASCII tree graph (compared first for quick validation) * - `spans`: The normalized span tree (detailed comparison) * * Supports special markers in the snapshot: * - `{"__or__": ["value1", "value2"]}` - matches if actual equals any listed value * - `{"__any__": "string"}` - matches any string value * - `{"__any__": "number"}` - matches any number value * - `{"__any__": "boolean"}` - matches any boolean value * - `{"__any__": "object"}` - matches any object value * - `{"__any__": "array"}` - matches any array value * - `{"__any__": true}` - matches any non-null/undefined value * * Environment variables: * Use `{ updateSnapshot: true }` option to update the snapshot instead of comparing * * @param snapshotName - Name of the snapshot file (resolved relative to __snapshots__ directory) * @param options - Options for snapshot comparison * @param options.updateSnapshot - If true, update the snapshot file instead of comparing * @throws Error if the snapshot doesn't match (and updateSnapshot is false) */ async assertMatchesSnapshot(snapshotName, options) { const { join } = await import('path'); const snapshotPath = join(await getSnapshotsDir(), snapshotName); const normalizedTree = this.buildNormalizedTree(); const structureGraph = this.generateStructureGraph(normalizedTree); const currentData = { __structure__: structureGraph, spans: normalizedTree }; const currentJson = JSON.stringify(currentData, null, this.#config.jsonIndent); const shouldUpdate = options?.updateSnapshot; if (shouldUpdate) { const { writeFile } = await import('fs/promises'); await writeFile(snapshotPath, currentJson, "utf-8"); this.logger.info(`TestExporter: updated snapshot ${snapshotPath}`); return; } let snapshotData; let snapshotContent; try { const { readFile } = await import('fs/promises'); snapshotContent = await readFile(snapshotPath, "utf-8"); } catch (err) { if (err && typeof err === "object" && "code" in err && err.code === "ENOENT") { throw new Error( `Snapshot file not found: ${snapshotPath} Run with { updateSnapshot: true } to create it.` ); } throw err; } try { snapshotData = JSON.parse(snapshotContent); } catch (err) { throw new Error(`Failed to parse snapshot ${snapshotPath}: ${err instanceof Error ? err.message : String(err)}`); } let expectedSpans; let expectedStructure; if (Array.isArray(snapshotData)) { expectedSpans = snapshotData; } else if (snapshotData && typeof snapshotData === "object" && "spans" in snapshotData) { expectedSpans = snapshotData.spans; expectedStructure = snapshotData.__structure__; } else { throw new Error( `Invalid snapshot format in ${snapshotPath}. Expected an array or object with 'spans' property.` ); } if (expectedStructure) { const structureMismatches = this.#compareStructure(structureGraph, expectedStructure); if (structureMismatches.length > 0) { throw new Error( `Structure mismatch in snapshot: Expected: ${expectedStructure.join("\n")} Actual: ${structureGraph.join("\n")} Differences: ${structureMismatches.join("\n")} Snapshot: ${snapshotPath} Run with { updateSnapshot: true } to update.` ); } } const mismatches = []; this.#deepCompareWithMarkers(normalizedTree, expectedSpans, "$.spans", mismatches); if (mismatches.length > 0) { const mismatchDetails = mismatches.map( (m, i) => `${i + 1}. ${m.path} Expected: ${JSON.stringify(m.expected)} Actual: ${JSON.stringify(m.actual)}` ).join("\n\n"); throw new Error( `Snapshot has ${mismatches.length} mismatch${mismatches.length > 1 ? "es" : ""}: ${mismatchDetails} Snapshot: ${snapshotPath} Run with { updateSnapshot: true } to update.` ); } } /** * Compare two structure graphs and return differences */ #compareStructure(actual, expected) { const diffs = []; const maxLen = Math.max(actual.length, expected.length); for (let i = 0; i < maxLen; i++) { const actualLine = actual[i]; const expectedLine = expected[i]; if (actualLine !== expectedLine) { if (actualLine === void 0) { diffs.push(`Line ${i + 1}: Missing in actual`); diffs.push(` Expected: ${expectedLine}`); } else if (expectedLine === void 0) { diffs.push(`Line ${i + 1}: Extra in actual`); diffs.push(` Actual: ${actualLine}`); } else { diffs.push(`Line ${i + 1}:`); diffs.push(` Expected: ${expectedLine}`); diffs.push(` Actual: ${actualLine}`); } } } return diffs; } /** * Deep compare two values, supporting special markers like __or__ and __any__. * Collects all mismatches into the provided array. */ #deepCompareWithMarkers(actual, expected, path2, mismatches) { if (this.#isOrMarker(expected)) { const allowedValues = expected.__or__; const matches = allowedValues.some((allowed) => { const tempMismatches = []; this.#deepCompareWithMarkers(actual, allowed, path2, tempMismatches); return tempMismatches.length === 0; }); if (!matches) { mismatches.push({ path: path2, expected: { __or__: allowedValues }, actual }); } return; } if (this.#isAnyMarker(expected)) { const typeConstraint = expected.__any__; if (actual === null || actual === void 0) { mismatches.push({ path: path2, expected: { __any__: typeConstraint }, actual }); return; } if (typeConstraint === true) { return; } const actualType = Array.isArray(actual) ? "array" : typeof actual; if (actualType !== typeConstraint) { mismatches.push({ path: path2, expected: { __any__: typeConstraint }, actual: `(${actualType}) ${JSON.stringify(actual).slice(0, 50)}...` }); } return; } if (Array.isArray(expected)) { if (!Array.isArray(actual)) { mismatches.push({ path: path2, expected, actual }); return; } if (actual.length !== expected.length) { mismatches.push({ path: `${path2}.length`, expected: expected.length, actual: actual.length }); return; } for (let i = 0; i < expected.length; i++) { this.#deepCompareWithMarkers(actual[i], expected[i], `${path2}[${i}]`, mismatches); } return; } if (expected !== null && typeof expected === "object") { if (actual === null || typeof actual !== "object" || Array.isArray(actual)) { mismatches.push({ path: path2, expected, actual }); return; } const expectedObj = expected; const actualObj = actual; for (const key of Object.keys(expectedObj)) { if (this.#isMetadataKey(key)) { continue; } if (!(key in actualObj)) { if (expectedObj[key] !== void 0) { mismatches.push({ path: `${path2}.${key}`, expected: expectedObj[key], actual: void 0 }); } continue; } this.#deepCompareWithMarkers(actualObj[key], expectedObj[key], `${path2}.${key}`, mismatches); } for (const key of Object.keys(actualObj)) { if (this.#isMetadataKey(key)) { continue; } if (!(key in expectedObj)) { if (actualObj[key] !== void 0) { mismatches.push({ path: `${path2}.${key}`, expected: void 0, actual: actualObj[key] }); } } } return; } if (actual !== expected) { mismatches.push({ path: path2, expected, actual }); } } /** * Check if a value is an __or__ marker object */ #isOrMarker(value) { return value !== null && typeof value === "object" && !Array.isArray(value) && "__or__" in value && Array.isArray(value.__or__); } /** * Check if a value is an __any__ marker object */ #isAnyMarker(value) { if (value === null || typeof value !== "object" || Array.isArray(value)) { return false; } if (!("__any__" in value)) { return false; } const constraint = value.__any__; return constraint === true || ["string", "number", "boolean", "object", "array"].includes(constraint); } /** * Check if a key should be skipped during comparison (metadata keys like __structure__) */ #isMetadataKey(key) { return key.startsWith("__") && key.endsWith("__"); } /** * Custom JSON replacer to handle Date objects and other special types */ #jsonReplacer = (_key, value) => { if (value instanceof Date) { return value.toISOString(); } return value; }; // ============================================================================ // Debugging Helpers // ============================================================================ /** * Get all stored debug logs (internal exporter logging, not signal logs) */ getLogs() { return [...this.#debugLogs]; } /** * Dump debug logs to console for debugging (uses console.error for visibility in test output) */ dumpLogs() { console.error("\n=== TestExporter Logs ==="); this.#debugLogs.forEach((log) => { console.error(log); }); console.error("=== End Logs ===\n"); } /** * Validate final state - useful for test assertions * * @returns Object with validation results */ validateFinalState() { const traceIds = this.getTraceIds(); const incompleteSpans = this.getIncompleteSpans(); const singleTraceId = traceIds.length === 1; const allSpansComplete = incompleteSpans.length === 0; return { valid: singleTraceId && allSpansComplete, singleTraceId, allSpansComplete, traceIds, incompleteSpans }; } // ============================================================================ // Reset & Lifecycle // ============================================================================ /** * Clear all collected events and state across all signals */ clearEvents() { this.#tracingEvents = []; this.#spanStates.clear(); this.#logEvents = []; this.#metricEvents = []; this.#scoreEvents = []; this.#feedbackEvents = []; this.#debugLogs = []; } /** * Alias for clearEvents (compatibility with TestExporter) */ reset() { this.clearEvents(); } /** * Get internal metrics about the exporter's own activity. */ getInternalMetrics() { const json = this.toJSON({ includeEvents: false, includeStats: false }); return { startedAt: this.#internalMetrics.startedAt, lastEventAt: this.#internalMetrics.lastEventAt, totalEventsReceived: this.#internalMetrics.totalEventsReceived, bySignal: { ...this.#internalMetrics.bySignal }, flushCount: this.#internalMetrics.flushCount, estimatedJsonBytes: new TextEncoder().encode(json).byteLength }; } /** * Flush buffered data and log internal metrics summary. */ async flush() { this.#internalMetrics.flushCount++; if (this.#config.logMetricsOnFlush) { const metrics = this.getInternalMetrics(); const uptimeMs = Date.now() - metrics.startedAt.getTime(); const summary = [ `[TestExporter] flush #${metrics.flushCount} summary:`, ` uptime: ${(uptimeMs / 1e3).toFixed(1)}s`, ` total events received: ${metrics.totalEventsReceived}`, ` by signal: tracing=${metrics.bySignal.tracing}, log=${metrics.bySignal.log}, metric=${metrics.bySignal.metric}, score=${metrics.bySignal.score}, feedback=${metrics.bySignal.feedback}`, ` buffered: spans=${this.#spanStates.size}, logs=${this.#logEvents.length}, metrics=${this.#metricEvents.length}, scores=${this.#scoreEvents.length}, feedback=${this.#feedbackEvents.length}`, ` estimated JSON size: ${(metrics.estimatedJsonBytes / 1024).toFixed(1)}KB` ].join("\n"); this.logger.info(summary); } } async shutdown() { await this.flush(); this.logger.info("TestExporter shutdown"); } // ============================================================================ // Private Helpers // ============================================================================ /** * Extract unique spans from a list of events */ #getUniqueSpansFromEvents(events) { const spanMap = /* @__PURE__ */ new Map(); for (const event of events) { const span = event.exportedSpan; if (event.type === observability.TracingEventType.SPAN_ENDED || !spanMap.has(span.id)) { spanMap.set(span.id, span); } } return Array.from(spanMap.values()); } }; var JsonExporter = TestExporter; // src/instances/default.ts var DefaultObservabilityInstance = class extends BaseObservabilityInstance { constructor(config) { super(config); } createSpan(options) { return new DefaultSpan(options, this); } }; function nullToUndefined(value) { return value ?? void 0; } function mergeMetadata(base, extra) { if (!base && !extra) return void 0; return { ...base ?? {}, ...extra ?? {} }; } function normalizeErrorInfo(error) { if (!error || typeof error !== "object" || !("message" in error) || typeof error.message !== "string") { return void 0; } return { message: error.message, id: "id" in error && typeof error.id === "string" ? error.id : void 0, name: "name" in error && typeof error.name === "string" ? error.name : void 0, stack: "stack" in error && typeof error.stack === "string" ? error.stack : void 0, domain: "domain" in error && typeof error.domain === "string" ? error.domain : void 0, category: "category" in error && typeof error.category === "string" ? error.category : void 0, details: "details" in error && error.details && typeof error.details === "object" ? error.details : void 0 }; } function buildCorrelationContext(span, rootSpan, parent) { return { tags: rootSpan.tags ?? void 0, entityType: nullToUndefined(span.entityType), entityId: nullToUndefined(span.entityId), entityName: nullToUndefined(span.entityName), parentEntityType: parent?.entityType ?? void 0, parentEntityId: parent?.entityId ?? void 0, parentEntityName: parent?.entityName ?? void 0, rootEntityType: nullToUndefined(rootSpan.entityType), rootEntityId: nullToUndefined(rootSpan.entityId), rootEntityName: nullToUndefined(rootSpan.entityName), userId: nullToUndefined(span.userId), organizationId: nullToUndefined(span.organizationId), resourceId: nullToUndefined(span.resourceId), runId: nullToUndefined(span.runId), sessionId: nullToUndefined(span.sessionId), threadId: nullToUndefined(span.threadId), requestId: nullToUndefined(span.requestId), environment: nullToUndefined(span.environment), source: nullToUndefined(span.source), serviceName: nullToUndefined(span.serviceName), experimentId: nullToUndefined(span.experimentId) }; } function buildScoreEvent(args) { const { traceId, spanId, correlationContext, score, inheritedMetadata } = args; return { type: "score", score: { scoreId: observability.generateSignalId(), timestamp: /* @__PURE__ */ new Date(), traceId, spanId, scorerId: score.scorerId, scorerName: score.scorerName, scorerVersion: score.scorerVersion, source: score.source, scoreSource: score.scoreSource, score: score.score, reason: score.reason, experimentId: score.experimentId, scoreTraceId: score.scoreTraceId, targetEntityType: score.targetEntityType, correlationContext, metadata: mergeMetadata(inheritedMetadata, score.metadata) } }; } function buildFeedbackEvent(args) { const { traceId, spanId, correlationContext, feedback, inheritedMetadata } = args; return { type: "feedback", feedback: { feedbackId: observability.generateSignalId(), timestamp: /* @__PURE__ */ new Date(), traceId, spanId, source: feedback.source, feedbackSource: feedback.feedbackSource, feedbackType: feedback.feedbackType, value: feedback.value, userId: feedback.userId, feedbackUserId: feedback.feedbackUserId, comment: feedback.comment, sourceId: feedback.sourceId, experimentId: feedback.experimentId, correlationContext, metadata: mergeMetadata(inheritedMetadata, feedback.metadata) } }; } function findSpanById(spans, spanId) { if (!spanId) return void 0; return spans.find((span) => span.spanId === spanId); } function buildRecordedScoreEventFromTrace(args) { const rootSpan = findRootSpan(args.trace.spans); if (!rootSpan) return null; const span = args.spanId ? findSpanById(args.trace.spans, args.spanId) : rootSpan; if (!span) return null; const parent = span.parentSpanId ? findSpanById(args.trace.spans, span.parentSpanId) : void 0; return buildScoreEvent({ traceId: span.traceId, spanId: args.spanId, correlationContext: buildCorrelationContext(span, rootSpan, parent), score: args.score, inheritedMetadata: span.metadata }); } function buildRecordedFeedbackEventFromTrace(args) { const rootSpan = findRootSpan(args.trace.spans); if (!rootSpan) return null; const span = args.spanId ? findSpanById(args.trace.spans, args.spanId) : rootSpan; if (!span) return null; const parent = span.parentSpanId ? findSpanById(args.trace.spans, span.parentSpanId) : void 0; return buildFeedbackEvent({ traceId: span.traceId, spanId: args.spanId, correlationContext: buildCorrelationContext(span, rootSpan, parent), feedback: args.feedback, inheritedMetadata: span.metadata }); } var RecordedSpanImpl = class { id; traceId; name; type; entityType; entityId; entityName; startTime; endTime; attributes; metadata; tags; input; output; errorInfo; requestContext; isEvent; isRootSpan; parentSpanId; parent; children = []; #raw; #rootSpan; #emitRecordedEvent; #canEmitRecordedEvent; #debugRecordedAnnotationUnavailable; constructor(args) { const { raw, rootSpan, emitRecordedEvent, canEmitRecordedEvent, debugRecordedAnnotationUnavailable } = args; this.#raw = raw; this.#rootSpan = rootSpan; this.#emitRecordedEvent = emitRecordedEvent; this.#canEmitRecordedEvent = canEmitRecordedEvent; this.#debugRecordedAnnotationUnavailable = debugRecordedAnnotationUnavailable; this.id = raw.spanId; this.traceId = raw.traceId; this.name = raw.name; this.type = raw.spanType; this.entityType = raw.entityType ?? void 0; this.entityId = raw.entityId ?? void 0; this.entityName = raw.entityName ?? void 0; this.startTime = raw.startedAt; this.endTime = raw.endedAt ?? void 0; this.attributes = raw.attributes ?? void 0; this.metadata = raw.metadata ?? void 0; this.tags = raw.tags ?? void 0; this.input = raw.input ?? void 0; this.output = raw.output ?? void 0; this.errorInfo = normalizeErrorInfo(raw.error); this.requestContext = raw.requestContext ?? void 0; this.isEvent = raw.isEvent; this.isRootSpan = !raw.parentSpanId; this.parentSpanId = raw.parentSpanId ?? void 0; } async addScore(score) { if (!this.#canEmitRecordedEvent()) { this.#debugRecordedAnnotationUnavailable({ kind: "score", traceId: this.traceId, spanId: this.id }); return; } await this.#emitRecordedEvent( buildScoreEvent({ traceId: this.#raw.traceId, spanId: this.id, correlationContext: buildCorrelationContext(this.#raw, this.#rootSpan, this.parent), score, inheritedMetadata: this.#raw.metadata }) ); } async addFeedback(feedback) { if (!this.#canEmitRecordedEvent()) { this.#debugRecordedAnnotationUnavailable({ kind: "feedback", traceId: this.traceId, spanId: this.id }); return; } await this.#emitRecordedEvent( buildFeedbackEvent({ traceId: this.#raw.traceId, spanId: this.id, correlationContext: buildCorrelationContext(this.#raw, this.#rootSpan, this.parent), feedback, inheritedMetadata: this.#raw.metadata }) ); } }; var RecordedTraceImpl = class { traceId; rootSpan; spans; #rootRecord; #emitRecordedEvent; #spanMap; #canEmitRecordedEvent; #debugRecordedAnnotationUnavailable; constructor(args) { this.traceId = args.traceId; this.rootSpan = args.rootSpan; this.#rootRecord = args.rootRecord; this.spans = args.spans; this.#emitRecordedEvent = args.emitRecordedEvent; this.#spanMap = new Map(args.spans.map((span) => [span.id, span])); this.#canEmitRecordedEvent = args.canEmitRecordedEvent; this.#debugRecordedAnnotationUnavailable = args.debugRecordedAnnotationUnavailable; } getSpan(spanId) { return this.#spanMap.get(spanId) ?? null; } async addScore(score) { if (!this.#canEmitRecordedEvent()) { this.#debugRecordedAnnotationUnavailable({ kind: "score", traceId: this.traceId }); return; } await this.#emitRecordedEvent( buildScoreEvent({ traceId: this.#rootRecord.traceId, correlationContext: buildCorrelationContext(this.#rootRecord, this.#rootRecord), score, inheritedMetadata: this.#rootRecord.metadata }) ); } async addFeedback(feedback) { if (!this.#canEmitRecordedEvent()) { this.#debugRecordedAnnotationUnavailable({ kind: "feedback", traceId: this.traceId }); return; } await this.#emitRecordedEvent( buildFeedbackEvent({ traceId: this.#rootRecord.traceId, correlationContext: buildCorrelationContext(this.#rootRecord, this.#rootRecord), feedback, inheritedMetadata: this.#rootRecord.metadata }) ); } }; function findRootSpan(spans) { const spanIds = new Set(spans.map((span) => span.spanId)); return spans.find((span) => !span.parentSpanId || !spanIds.has(span.parentSpanId)) ?? spans[0]; } function hydrateRecordedTrace(args) { const { trace, emitRecordedEvent, canEmitRecordedEvent = () => true, debugRecordedAnnotationUnavailable = () => { } } = args; const rootSpan = findRootSpan(trace.spans); if (!rootSpan) { return null; } const recordedSpans = trace.spans.map( (raw) => new RecordedSpanImpl({ raw, rootSpan, emitRecordedEvent, canEmitRecordedEvent, debugRecordedAnnotationUnavailable }) ); const spanMap = new Map(recordedSpans.map((span) => [span.id, span])); for (const span of recordedSpans) { if (!span.parentSpanId) continue; const parent = spanMap.get(span.parentSpanId); if (!parent) continue; span.parent = parent; parent.children.push(span); } const hydratedRootSpan = spanMap.get(rootSpan.spanId); if (!hydratedRootSpan) { return null; } return new RecordedTraceImpl({ traceId: trace.traceId, rootSpan: hydratedRootSpan, rootRecord: rootSpan, spans: recordedSpans, emitRecordedEvent, canEmitRecordedEvent, debugRecordedAnnotationUnavailable }); } // src/registry.ts var ObservabilityRegistry = class { #instances = /* @__PURE__ */ new Map(); #defaultInstance; #configSelector; /** * Register a tracing instance */ register(name, instance, isDefault = false) { if (this.#instances.has(name)) { throw new Error(`Tracing instance '${name}' already registered`); } this.#instances.set(name, instance); if (isDefault || !this.#defaultInstance) { this.#defaultInstance = instance; } } /** * Get a tracing instance by name */ get(name) { return this.#instances.get(name); } /** * Get the default tracing instance */ getDefault() { return this.#defaultInstance; } /** * Set the tracing selector function */ setSelector(selector) { this.#configSelector = selector; } /** * Get the selected tracing instance based on context */ getSelected(options) { if (this.#configSelector) { const selected = this.#configSelector(options, this.#instances); if (selected && this.#instances.has(selected)) { return this.#instances.get(selected); } } return this.#defaultInstance; } /** * Unregister a tracing instance */ unregister(name) { const instance = this.#instances.get(name); const deleted = this.#instances.delete(name); if (deleted && instance === this.#defaultInstance) { const next = this.#instances.values().next(); this.#defaultInstance = next.done ? void 0 : next.value; } return deleted; } /** * Shutdown all instances and clear the registry */ async shutdown() { const shutdownPromises = Array.from(this.#instances.values()).map((instance) => instance.shutdown()); await Promise.allSettled(shutdownPromises); this.#instances.clear(); this.#instances.clear(); this.#defaultInstance = void 0; this.#configSelector = void 0; } /** * Clear all instances without shutdown */ clear() { this.#instances.clear(); this.#defaultInstance = void 0; this.#configSelector = void 0; } /** * list all registered instances */ list() { return new Map(this.#instances); } }; // src/span_processors/sensitive-data-filter.ts var SensitiveDataFilter = class { name = "sensitive-data-filter"; sensitiveFields; redactionToken; redactionStyle; constructor(options = {}) { this.sensitiveFields = (options.sensitiveFields || [ "password", "token", "secret", "key", "apikey", "auth", "authorization", "bearer", "bearertoken", "jwt", "credential", "clientsecret", "privatekey", "refresh", "ssn" ]).map((f) => this.normalizeKey(f)); this.redactionToken = options.redactionToken ?? "[REDACTED]"; this.redactionStyle = options.redactionStyle ?? "full"; } /** * Process a span by filtering sensitive data across its key fields. * Fields processed: attributes, metadata, input, output, errorInfo. * * @param span - The input span to filter * @returns A new span with sensitive values redacted */ process(span) { span.attributes = this.tryFilter(span.attributes); span.metadata = this.tryFilter(span.metadata); span.input = this.tryFilter(span.input); span.output = this.tryFilter(span.output); span.errorInfo = this.tryFilter(span.errorInfo); return span; } /** * Recursively filter objects/arrays for sensitive keys. * Handles circular references by replacing with a marker. * Also attempts to parse and redact JSON strings. */ deepFilter(obj, seen = /* @__PURE__ */ new WeakSet()) { if (obj === null || typeof obj !== "object") { if (typeof obj === "string") { const trimmed = obj.trim(); if (trimmed.startsWith("{") || trimmed.startsWith("[")) { return this.redactJsonString(obj); } } return obj; } if (seen.has(obj)) { return "[Circular Reference]"; } seen.add(obj); if (obj instanceof Date) { return obj; } if (Array.isArray(obj)) { return obj.map((item) => this.deepFilter(item, seen)); } const filtered = {}; for (const key of Object.keys(obj)) { const normKey = this.normalizeKey(key); if (this.isSensitive(normKey)) { if (obj[key] && typeof obj[key] === "object") { filtered[key] = this.deepFilter(obj[key], seen); } else { filtered[key] = this.redactValue(obj[key]); } } else { filtered[key] = this.deepFilter(obj[key], seen); } } return filtered; } tryFilter(value) { try { return this.deepFilter(value); } catch { return { error: { processor: this.name } }; } } /** * Normalize keys by lowercasing and stripping non-alphanumeric characters. * Ensures consistent matching for variants like "api-key", "api_key", "Api Key". */ normalizeKey(key) { return key.toLowerCase().replace(/[^a-z0-9]/g, ""); } /** * Check whether a normalized key exactly matches any sensitive field. * Both key and sensitive fields are normalized by removing all non-alphanumeric * characters and converting to lowercase before comparison. * * Examples: * - "api_key", "api-key", "ApiKey" all normalize to "apikey" → MATCHES "apikey" * - "promptTokens", "prompt_tokens" normalize to "prompttokens" → DOES NOT MATCH "token" */ isSensitive(normalizedKey) { return this.sensitiveFields.some((sensitiveField) => { return normalizedKey === sensitiveField; }); } /** * Attempt to parse a string as JSON and redact sensitive fields within it. * If parsing fails or no sensitive data is found, returns the original string. */ redactJsonString(str) { try { const parsed = JSON.parse(str); if (parsed && typeof parsed === "object") { const filtered = this.deepFilter(parsed, /* @__PURE__ */ new WeakSet()); return JSON.stringify(filtered); } return str; } catch { return str; } } /** * Redact a sensitive value. * - Full style: replaces with a fixed token. * - Partial style: shows 3 chars at start and end, hides the middle. * * Non-string values are converted to strings before partial redaction. */ redactValue(value) { if (this.redactionStyle === "full") { return this.redactionToken; } const str = String(value); const len = str.length; if (len <= 6) { return this.redactionToken; } return str.slice(0, 3) + "\u2026" + str.slice(len - 3); } async shutdown() { } }; // src/default.ts function isInstance(obj) { return obj instanceof BaseObservabilityInstance; } var Observability = class extends base.MastraBase { #registry = new ObservabilityRegistry(); #mastra; #clientObservabilityProxy; constructor(config) { super({ component: logger.RegisteredLogger.OBSERVABILITY, name: "Observability" }); if (config === void 0) { config = {}; } const validationResult = observabilityRegistryConfigSchema.safeParse(config); if (!validationResult.success) { const errorMessages = validationResult.error.issues.map( (err) => `${err.path.join(".") || "config"}: ${err.message}` ).join("; "); throw new error.MastraError({ id: "OBSERVABILITY_INVALID_CONFIG", text: `Invalid observability configuration: ${errorMessages}`, domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER, details: { validationErrors: errorMessages } }); } if (config.configs) { for (const [name, configValue] of Object.entries(config.configs)) { if (!isInstance(configValue)) { const configValidation = observabilityConfigValueSchema.safeParse(configValue); if (!configValidation.success) { const errorMessages = configValidation.error.issues.map( (err) => `${err.path.join(".")}: ${err.message}` ).join("; "); throw new error.MastraError({ id: "OBSERVABILITY_INVALID_INSTANCE_CONFIG", text: `Invalid configuration for observability instance '${name}': ${errorMessages}`, domain: error.ErrorDomain.MASTRA_OBSERVABILITY, category: error.ErrorCategory.USER, details: { instanceName: name, validationErrors: errorMessages } }); } } } } const sensitiveDataFilterSetting = config.sensitiveDataFilter ?? true; const shouldAutoApplySensitiveFilter = sensitiveDataFilterSetting !== false; const sensitiveDataFilterOptions = typeof sensitiveDataFilterSetting === "object" && sensitiveDataFilterSetting !== null ? sensitiveDataFilterSetting : void 0; const buildAutoSensitiveFilter = () => { if (!shouldAutoApplySensitiveFilter) { return void 0; } return new SensitiveDataFilter(sensitiveDataFilterOptions); }; if (config.default?.enabled) { console.warn( '[Mastra Observability] The "default: { enabled: true }" configuration is deprecated and will be removed in a future version. Please use explicit configs with MastraStorageExporter and MastraPlatformExporter instead. Sensitive data filtering is applied by default and can be controlled via the top-level "sensitiveDataFilter" option. See https://mastra.ai/docs/observability/tracing/overview for the recommended configuration.' ); const autoFilter = buildAutoSensitiveFilter(); const defaultInstance = new DefaultObservabilityInstance({ serviceName: "mastra", name: "default", sampling: { type: "always" /* ALWAYS */ }, exporters: [new MastraStorageExporter(), new MastraPlatformExporter()], spanOutputProcessors: autoFilter ? [autoFilter] : [] }); this.#registry.register("default", defaultInstance, true); } if (config.configs) { const instances = Object.entries(config.configs); instances.forEach(([name, tracingDef], index) => { let instance; if (isInstance(tracingDef)) { instance = tracingDef; if (shouldAutoApplySensitiveFilter) { const processors = instance.getSpanOutputProcessors?.() ?? []; const hasFilter = processors.some((p) => p instanceof SensitiveDataFilter); if (!hasFilter) { this.logger?.warn( "[Mastra Observability] Pre-instantiated observability instance does not include a SensitiveDataFilter. Auto-applied filtering is skipped for pre-instantiated instances. Add a SensitiveDataFilter to spanOutputProcessors when constructing the instance to redact sensitive data.", { instanceName: name } ); } } } else { const userProcessors = tracingDef.spanOutputProcessors ?? []; const hasFilter = userProcessors.some((p) => p instanceof SensitiveDataFilter); const autoFilter = !hasFilter ? buildAutoSensitiveFilter() : void 0; const spanOutputProcessors = autoFilter ? [...userProcessors, autoFilter] : userProcessors; instance = new DefaultObservabilityInstance({ ...tracingDef, name, spanOutputProcessors }); } const isDefault = !config.default?.enabled && index === 0; this.#registry.register(name, instance, isDefault); }); } if (config.configSelector) { this.#registry.setSelector(config.configSelector); } } /** Initialize all exporter instances with the Mastra context (storage, config, etc.). */ setMastraContext(options) { const instances = this.listInstances(); const { mastra } = options; this.#mastra = mastra; const mastraEnvironment = mastra.getEnvironment?.(); instances.forEach((instance) => { instance.__setMastraEnvironment?.(mastraEnvironment); const config = instance.getConfig(); const exporters = instance.getExporters(); const emitDropEvent = instance instanceof BaseObservabilityInstance ? (event) => instance.getObservabilityBus().emitDropEvent(event) : void 0; exporters.forEach((exporter) => { if ("init" in exporter && typeof exporter.init === "function") { try { exporter.init({ mastra, config, emitDropEvent }); } catch (error) { this.logger?.warn("Failed to initialize observability exporter", { exporterName: exporter.name, error: error instanceof Error ? error.message : String(error) }); } } }); }); } /** Propagate a logger to this instance and all registered observability instances. */ setLogger(options) { super.__setLogger(options.logger); this.listInstances().forEach((instance) => { instance.__setLogger(options.logger); }); } /** Get the observability instance chosen by the config selector for the given options. */ getSelectedInstance(options) { return this.#registry.getSelected(options); } async getRecordedTrace(args) { const observabilityStorage = await this.#getObservabilityStorage(); if (!observabilityStorage) { return null; } const trace = await observabilityStorage.getTrace({ traceId: args.traceId }); if (!trace) { return null; } return hydrateRecordedTrace({ trace, emitRecordedEvent: (event) => this.#emitRecordedEvent(event), canEmitRecordedEvent: () => !!this.#getRecordedTraceInstance(), debugRecordedAnnotationUnavailable: ({ kind, traceId, spanId }) => { this.logger?.debug( kind === "score" ? "addScore() is unavailable; rehydrate the trace before calling addScore()" : "addFeedback() is unavailable; rehydrate the trace before calling addFeedback()", { traceId, spanId } ); } }); } async addScore(args) { const targetTraceId = args.traceId ?? args.correlationContext?.traceId; const targetSpanId = args.spanId ?? args.correlationContext?.spanId; if (args.correlationContext) { await this.#emitRecordedEvent( buildScoreEvent({ ...targetTraceId ? { traceId: targetTraceId } : {}, ...targetSpanId ? { spanId: targetSpanId } : {}, correlationContext: args.correlationContext, score: args.score }) ); return; } if (!args.traceId) { return; } const trace = await this.#getStoredTrace(args.traceId); if (!trace) { return; } const event = buildRecordedScoreEventFromTrace({ trace, spanId: args.spanId, score: args.score }); if (!event) { return; } await this.#emitRecordedEvent(event); } async addFeedback(args) { const targetTraceId = args.traceId ?? args.correlationContext?.traceId; const targetSpanId = args.spanId ?? args.correlationContext?.spanId; if (args.correlationContext) { await this.#emitRecordedEvent( buildFeedbackEvent({ ...targetTraceId ? { traceId: targetTraceId } : {}, ...targetSpanId ? { spanId: targetSpanId } : {}, correlationContext: args.correlationContext, feedback: args.feedback }) ); return; } if (!args.traceId) { return; } const trace = await this.#getStoredTrace(args.traceId); if (!trace) { return; } const event = buildRecordedFeedbackEventFromTrace({ trace, spanId: args.spanId, feedback: args.feedback }); if (!event) { return; } await this.#emitRecordedEvent(event); } /** Register a named observability instance, optionally marking it as default. */ registerInstance(name, instance, isDefault = false) { this.#registry.register(name, instance, isDefault); if (this.#mastra) { instance.__setMastraEnvironment?.(this.#mastra.getEnvironment?.()); } } /** Get a registered instance by name. */ getInstance(name) { return this.#registry.get(name); } /** Get the default observability instance. */ getDefaultInstance() { return this.#registry.getDefault(); } /** List all registered observability instances. */ listInstances() { return this.#registry.list(); } /** Unregister an instance by name. Returns true if it was found and removed. */ unregisterInstance(name) { return this.#registry.unregister(name); } /** Check whether an instance with the given name is registered. */ hasInstance(name) { return !!this.#registry.get(name); } /** Set the config selector used to choose an instance at runtime. */ setConfigSelector(selector) { this.#registry.setSelector(selector); } /** Remove all registered instances and reset the registry. */ clear() { this.#registry.clear(); } /** Shut down all registered instances, flushing any pending data. */ async shutdown() { await this.#registry.shutdown(); } /** * Returns the proxy responsible for client observability (W3C trace * context injection + OTLP/JSON payload reception for spans/logs * returned from client-side execution). * * Lazily constructed on first call. Resolves the target observability * instance per receive call so config selection works the same way * as for server-side spans. */ getClientObservabilityProxy() { if (!this.#clientObservabilityProxy) { this.#clientObservabilityProxy = createClientObservabilityProxy({ resolveInstance: () => this.getDefaultInstance(), logger: this.logger }); } return this.#clientObservabilityProxy; } async #getObservabilityStorage() { const storage = this.#mastra?.getStorage(); if (!storage) { return null; } return await storage.getStore("observability") ?? null; } async #getStoredTrace(traceId) { const observabilityStorage = await this.#getObservabilityStorage(); if (!observabilityStorage) { return null; } return observabilityStorage.getTrace({ traceId }); } #getRecordedTraceInstance() { return this.getDefaultInstance() ?? Array.from(this.listInstances().values())[0]; } async #emitRecordedEvent(event) { const instance = this.#getRecordedTraceInstance(); if (!instance) { this.logger?.debug( event.type === "score" ? "Score event was dropped because no observability instance is registered" : "Feedback event was dropped because no observability instance is registered", { eventType: event.type } ); return; } if (instance instanceof BaseObservabilityInstance) { instance.__emitRecordedEvent(event); return; } const bridge = instance.getBridge(); const handlerResults = [ ...instance.getExporters().map((exporter) => routeToHandler(exporter, event, this.logger)), ...bridge ? [routeToHandler(bridge, event, this.logger)] : [] ].filter((result) => !!result && typeof result.then === "function"); if (handlerResults.length > 0) { await Promise.allSettled(handlerResults); } } }; // src/features.ts var observabilityFeatures = /* @__PURE__ */ new Set(["model-inference-span", "internal-usage-rollup"]); // src/tracing-options.ts function buildTracingOptions(...updaters) { return updaters.reduce((opts, updater) => updater(opts), {}); } exports.BaseExporter = BaseExporter; exports.BaseObservabilityEventBus = BaseObservabilityEventBus; exports.BaseObservabilityInstance = BaseObservabilityInstance; exports.BaseSpan = BaseSpan; exports.CardinalityFilter = CardinalityFilter; exports.CloudExporter = CloudExporter; exports.ConsoleExporter = ConsoleExporter; exports.DEFAULT_DEEP_CLEAN_OPTIONS = DEFAULT_DEEP_CLEAN_OPTIONS; exports.DEFAULT_KEYS_TO_STRIP = DEFAULT_KEYS_TO_STRIP; exports.DEFAULT_LIMITS = DEFAULT_LIMITS; exports.DefaultExporter = DefaultExporter; exports.DefaultObservabilityInstance = DefaultObservabilityInstance; exports.DefaultSpan = DefaultSpan; exports.JsonExporter = JsonExporter; exports.LoggerContextImpl = LoggerContextImpl; exports.MastraPlatformExporter = MastraPlatformExporter; exports.MastraStorageExporter = MastraStorageExporter; exports.MetricsContextImpl = MetricsContextImpl; exports.ModelSpanTracker = ModelSpanTracker; exports.NoOpSpan = NoOpSpan; exports.Observability = Observability; exports.ObservabilityBus = ObservabilityBus; exports.SamplingStrategyType = SamplingStrategyType; exports.SensitiveDataFilter = SensitiveDataFilter; exports.TestExporter = TestExporter; exports.TraceData = TraceData; exports.TrackingExporter = TrackingExporter; exports.buildExportedLog = buildExportedLog; exports.buildExportedSpan = buildExportedSpan; exports.buildTracingOptions = buildTracingOptions; exports.chainFormatters = chainFormatters; exports.createClientObservabilityProxy = createClientObservabilityProxy; exports.decodeResourceLogs = decodeResourceLogs; exports.decodeResourceSpans = decodeResourceSpans; exports.deepClean = deepClean; exports.formatBaggage = formatBaggage; exports.formatTraceparent = formatTraceparent; exports.getExternalParentId = getExternalParentId; exports.isSerializedMap = isSerializedMap; exports.mergeSerializationOptions = mergeSerializationOptions; exports.observabilityConfigValueSchema = observabilityConfigValueSchema; exports.observabilityFeatures = observabilityFeatures; exports.observabilityInstanceConfigSchema = observabilityInstanceConfigSchema; exports.observabilityRegistryConfigSchema = observabilityRegistryConfigSchema; exports.otlpSeverityToLogLevel = otlpSeverityToLogLevel; exports.parseBaggage = parseBaggage; exports.parseTraceparent = parseTraceparent; exports.reconstructSerializedMap = reconstructSerializedMap; exports.routeToHandler = routeToHandler; exports.samplingStrategySchema = samplingStrategySchema; exports.serializationOptionsSchema = serializationOptionsSchema; exports.truncateString = truncateString; //# sourceMappingURL=index.cjs.map //# sourceMappingURL=index.cjs.map