From 56f0cde9a56e7de94d45499cea04f0b2a70370a9 Mon Sep 17 00:00:00 2001 From: Ravi Kumar L Date: Wed, 26 Aug 2026 13:38:30 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=A9=BB=20feat(langfuse):=20add=20tenant?= =?UTF-8?q?=20export=20telemetry=20(#15247)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- otel/langfuse-fanout/README.md | 16 ++- .../cmd/langfuse-fanout/main.go | 134 ++++++++++++++++-- .../cmd/langfuse-fanout/main_test.go | 70 ++++++++- .../cmd/langfuse-fanout/metrics.go | 14 +- .../api/src/admin/langfuse.handler.spec.ts | 22 +++ packages/api/src/admin/langfuse.ts | 71 +++++++++- .../__tests__/run-summarization.test.ts | 43 +++++- packages/api/src/langfuse/config.spec.ts | 55 +++++++ packages/api/src/langfuse/config.ts | 118 +++++++++++++-- .../src/config/requestLogContext.spec.ts | 26 ++++ .../src/config/requestLogContext.ts | 18 ++- 11 files changed, 552 insertions(+), 35 deletions(-) diff --git a/otel/langfuse-fanout/README.md b/otel/langfuse-fanout/README.md index 3052c72670..6da4a6fd5d 100644 --- a/otel/langfuse-fanout/README.md +++ b/otel/langfuse-fanout/README.md @@ -260,7 +260,7 @@ Useful gateway metrics include: - `langfuse_fanout_http_requests_total` - `langfuse_fanout_upstream_requests_total` -- `langfuse_fanout_trace_exports_total` +- `langfuse_fanout_trace_exports_total` (`destination`, `result`, and `tenant_id` labels) - `langfuse_fanout_media_upload_plans_created_total` - `langfuse_fanout_media_upload_plans_completed_total` - `langfuse_fanout_media_upload_plan_misses_total` @@ -268,6 +268,20 @@ Useful gateway metrics include: - `langfuse_fanout_media_upload_bytes` - `langfuse_fanout_media_divergence_total` +LibreChat stamps `librechat.tenant.id`, `librechat.langfuse.export_plan`, and +`librechat.langfuse.export_reason` on Langfuse run spans. The gateway reads the +tenant ID from each OTLP batch for the trace export counter. Batches without a +tenant ID use ``; batches containing more than one tenant use ``. +Angle brackets keep these synthetic values outside LibreChat's accepted tenant-ID grammar. +The `tenant_id` label is intentionally high-cardinality and must only receive +traffic from trusted LibreChat deployments. Each distinct tenant creates a +Prometheus time series for every destination and result combination. + +Successful admin connection updates emit the structured log event +`librechat.langfuse.connection.changed`. It includes the tenant, configuration +state, destination, verification result, a primary `change`, and all `changes`. +It does not include the Langfuse public or secret key. + `langfuse_fanout_media_divergence_total{kind="media_id"}` is the correctness signal for trace/media token fanout. `kind="upload_url_presence"` records that some destinations returned an upload URL while others treated the media as diff --git a/otel/langfuse-fanout/cmd/langfuse-fanout/main.go b/otel/langfuse-fanout/cmd/langfuse-fanout/main.go index d85e696c83..6fa069c711 100644 --- a/otel/langfuse-fanout/cmd/langfuse-fanout/main.go +++ b/otel/langfuse-fanout/cmd/langfuse-fanout/main.go @@ -42,6 +42,9 @@ const ( metricsPath = "/metrics" tenantExportAttribute = "librechat.langfuse.tenant_export.enabled" tenantDestAttribute = "librechat.langfuse.destination" + tenantIDAttribute = "librechat.tenant.id" + unknownTenantID = "" + multipleTenantIDs = "" ) type config struct { @@ -269,19 +272,25 @@ func (g *gateway) handle(w http.ResponseWriter, r *http.Request) { func (g *gateway) handleTraces(w http.ResponseWriter, r *http.Request, route route) { body, err := readMaybeGzip(r) if err != nil { + g.recordTraceExport(route, "error", unknownTenantID) g.writeGatewayError(w, r, route, "trace_export", http.StatusBadRequest, "failed to read request body", err) return } contentType := r.Header.Get("Content-Type") - if route.destination != "" && + tenantID := unknownTenantID + authorizedTenantRoute := route.destination != "" && g.cfg.tenants[route.destination] != "" && - strings.TrimSpace(r.Header.Get("Authorization")) != "" { - body, err = addTenantRouteAttributes(body, contentType, route.destination) + strings.TrimSpace(r.Header.Get("Authorization")) != "" + if authorizedTenantRoute { + body, tenantID, err = addTenantRouteAttributes(body, contentType, route.destination) if err != nil { + g.recordTraceExport(route, "error", tenantID) g.writeGatewayError(w, r, route, "trace_route_attributes", http.StatusBadRequest, "failed to add OTLP tenant routing attributes", err) return } + } else if g.metrics != nil { + tenantID = extractTraceTenantID(body, contentType) } contentEncoding := "" @@ -289,6 +298,7 @@ func (g *gateway) handleTraces(w http.ResponseWriter, r *http.Request, route rou contentEncoding = "gzip" body, err = gzipBytes(body) if err != nil { + g.recordTraceExport(route, "error", tenantID) g.writeGatewayError(w, r, route, "trace_gzip", http.StatusInternalServerError, "failed to encode request body", err) return } @@ -296,13 +306,13 @@ func (g *gateway) handleTraces(w http.ResponseWriter, r *http.Request, route rou resp, err := g.forwardTraceToCollector(r.Context(), r.Header, body, contentType, contentEncoding) if err != nil { - g.recordTraceExport(route, "error") + g.recordTraceExport(route, "error", tenantID) g.writeGatewayError(w, r, route, "trace_collector", http.StatusBadGateway, "trace collector export failed", err) return } defer resp.Body.Close() - g.recordTraceExport(route, "success") + g.recordTraceExport(route, "success", tenantID) copyResponseHeaders(w.Header(), resp.Header) w.WriteHeader(resp.StatusCode) _, _ = io.Copy(w, resp.Body) @@ -734,11 +744,11 @@ func (g *gateway) doUpstreamWithClient(client *http.Client, req *http.Request, o return resp, nil } -func (g *gateway) recordTraceExport(route route, result string) { +func (g *gateway) recordTraceExport(route route, result string, tenantID string) { if g.metrics == nil { return } - g.metrics.recordTraceExport(routeDestinationLabel(route), result) + g.metrics.recordTraceExport(routeDestinationLabel(route), result, tenantID) } func (g *gateway) recordMediaDivergence(kind string, destination string) { @@ -774,18 +784,111 @@ func (g *gateway) recordUploadPlanStoreError(operation string) { } } -func addTenantRouteAttributes(body []byte, contentType string, destination string) ([]byte, error) { +func addTenantRouteAttributes(body []byte, contentType string, destination string) ([]byte, string, error) { if isJSONContentType(contentType) { return addJSONTenantRouteAttributes(body, destination) } return addProtobufTenantRouteAttributes(body, destination) } -func addProtobufTenantRouteAttributes(body []byte, destination string) ([]byte, error) { +func extractTraceTenantID(body []byte, contentType string) string { + if isJSONContentType(contentType) { + return extractJSONTraceTenantID(body) + } + return extractProtobufTraceTenantID(body) +} + +func resolveTraceTenantID(tenantIDs map[string]struct{}) string { + if len(tenantIDs) == 0 { + return unknownTenantID + } + if len(tenantIDs) > 1 { + return multipleTenantIDs + } + for tenantID := range tenantIDs { + return tenantID + } + return unknownTenantID +} + +func extractProtobufTraceTenantID(body []byte) string { var request tracepb.ExportTraceServiceRequest if err := proto.Unmarshal(body, &request); err != nil { - return nil, err + return unknownTenantID } + return protobufTraceTenantID(&request) +} + +func protobufTraceTenantID(request *tracepb.ExportTraceServiceRequest) string { + tenantIDs := make(map[string]struct{}) + for _, resourceSpan := range request.ResourceSpans { + for _, scopeSpan := range resourceSpan.ScopeSpans { + for _, span := range scopeSpan.Spans { + for _, attribute := range span.Attributes { + if attribute.Key != tenantIDAttribute { + continue + } + tenantID := strings.TrimSpace(attribute.Value.GetStringValue()) + if tenantID != "" { + tenantIDs[tenantID] = struct{}{} + if len(tenantIDs) > 1 { + return multipleTenantIDs + } + } + } + } + } + } + return resolveTraceTenantID(tenantIDs) +} + +func extractJSONTraceTenantID(body []byte) string { + var request map[string]any + if err := json.Unmarshal(body, &request); err != nil { + return unknownTenantID + } + return jsonTraceTenantID(request) +} + +func jsonTraceTenantID(request map[string]any) string { + tenantIDs := make(map[string]struct{}) + resourceSpans, _ := request["resourceSpans"].([]any) + for _, resourceSpan := range resourceSpans { + resourceSpanMap, _ := resourceSpan.(map[string]any) + scopeSpans, _ := resourceSpanMap["scopeSpans"].([]any) + for _, scopeSpan := range scopeSpans { + scopeSpanMap, _ := scopeSpan.(map[string]any) + spans, _ := scopeSpanMap["spans"].([]any) + for _, span := range spans { + spanMap, _ := span.(map[string]any) + attributes, _ := spanMap["attributes"].([]any) + for _, attribute := range attributes { + attributeMap, _ := attribute.(map[string]any) + if attributeMap["key"] != tenantIDAttribute { + continue + } + value, _ := attributeMap["value"].(map[string]any) + tenantID, _ := value["stringValue"].(string) + tenantID = strings.TrimSpace(tenantID) + if tenantID != "" { + tenantIDs[tenantID] = struct{}{} + if len(tenantIDs) > 1 { + return multipleTenantIDs + } + } + } + } + } + } + return resolveTraceTenantID(tenantIDs) +} + +func addProtobufTenantRouteAttributes(body []byte, destination string) ([]byte, string, error) { + var request tracepb.ExportTraceServiceRequest + if err := proto.Unmarshal(body, &request); err != nil { + return nil, unknownTenantID, err + } + tenantID := protobufTraceTenantID(&request) for _, resourceSpan := range request.ResourceSpans { for _, scopeSpan := range resourceSpan.ScopeSpans { @@ -796,7 +899,8 @@ func addProtobufTenantRouteAttributes(body []byte, destination string) ([]byte, } } - return proto.Marshal(&request) + updatedBody, err := proto.Marshal(&request) + return updatedBody, tenantID, err } func upsertSpanStringAttribute(span *tracev1.Span, key string, value string) { @@ -818,11 +922,12 @@ func stringAnyValue(value string) *commonv1.AnyValue { } } -func addJSONTenantRouteAttributes(body []byte, destination string) ([]byte, error) { +func addJSONTenantRouteAttributes(body []byte, destination string) ([]byte, string, error) { var request map[string]any if err := json.Unmarshal(body, &request); err != nil { - return nil, err + return nil, unknownTenantID, err } + tenantID := jsonTraceTenantID(request) resourceSpans, _ := request["resourceSpans"].([]any) for _, resourceSpan := range resourceSpans { resourceSpanMap, _ := resourceSpan.(map[string]any) @@ -837,7 +942,8 @@ func addJSONTenantRouteAttributes(body []byte, destination string) ([]byte, erro } } } - return json.Marshal(request) + updatedBody, err := json.Marshal(request) + return updatedBody, tenantID, err } func upsertJSONSpanStringAttribute(span map[string]any, key string, value string) { diff --git a/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go b/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go index f71b0da5c9..9893d665dc 100644 --- a/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go +++ b/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go @@ -11,6 +11,7 @@ import ( "net/http" "net/http/httptest" "net/url" + "regexp" "strings" "sync" "testing" @@ -1047,7 +1048,7 @@ func TestTraceProxyRecordsPrometheusMetrics(t *testing.T) { defer collector.Close() gw := newTestGatewayWithCollector(collector.URL) - body := buildTraceRequest(t, nil) + body := buildTraceRequest(t, map[string]string{tenantIDAttribute: "tenant-123"}) req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+otelTracePath, bytes.NewReader(body)) req.Header.Set("Content-Type", "application/x-protobuf") req.Header.Set("Authorization", "Basic tenant") @@ -1059,7 +1060,7 @@ func TestTraceProxyRecordsPrometheusMetrics(t *testing.T) { } metrics := scrapeMetrics(t, gw) - if !strings.Contains(metrics, `langfuse_fanout_trace_exports_total{destination="tenant_eu",result="success"} 1`) { + if !strings.Contains(metrics, `langfuse_fanout_trace_exports_total{destination="tenant_eu",result="success",tenant_id="tenant-123"} 1`) { t.Fatalf("missing trace export metric:\n%s", metrics) } if !strings.Contains(metrics, `langfuse_fanout_upstream_requests_total{destination="collector",operation="trace_collector",status_class="2xx"} 1`) { @@ -1067,6 +1068,71 @@ func TestTraceProxyRecordsPrometheusMetrics(t *testing.T) { } } +func TestExtractTraceTenantID(t *testing.T) { + t.Parallel() + + t.Run("protobuf", func(t *testing.T) { + body := buildTraceRequest(t, map[string]string{tenantIDAttribute: "tenant-123"}) + if tenantID := extractTraceTenantID(body, "application/x-protobuf"); tenantID != "tenant-123" { + t.Fatalf("tenant ID = %q", tenantID) + } + }) + + t.Run("json", func(t *testing.T) { + body := []byte(`{"resourceSpans":[{"scopeSpans":[{"spans":[{"attributes":[{"key":"librechat.tenant.id","value":{"stringValue":"tenant-456"}}]}]}]}]}`) + if tenantID := extractTraceTenantID(body, "application/json"); tenantID != "tenant-456" { + t.Fatalf("tenant ID = %q", tenantID) + } + }) + + t.Run("missing", func(t *testing.T) { + body := buildTraceRequest(t, nil) + if tenantID := extractTraceTenantID(body, "application/x-protobuf"); tenantID != unknownTenantID { + t.Fatalf("tenant ID = %q", tenantID) + } + }) + + t.Run("multiple", func(t *testing.T) { + tenantIDs := map[string]struct{}{"tenant-1": {}, "tenant-2": {}} + if tenantID := resolveTraceTenantID(tenantIDs); tenantID != multipleTenantIDs { + t.Fatalf("tenant ID = %q", tenantID) + } + }) + + t.Run("sentinels are outside tenant ID grammar", func(t *testing.T) { + validTenantID := regexp.MustCompile(`^[-a-zA-Z0-9_.]+$`) + for _, sentinel := range []string{unknownTenantID, multipleTenantIDs} { + if validTenantID.MatchString(sentinel) { + t.Fatalf("sentinel %q is a valid tenant ID", sentinel) + } + } + }) +} + +func TestMalformedTenantTraceRecordsErrorMetric(t *testing.T) { + t.Parallel() + + gw := newTestGatewayWithCollector("http://collector.invalid") + req := httptest.NewRequest( + http.MethodPost, + tenantPrefix+"eu"+otelTracePath, + bytes.NewReader([]byte("not protobuf")), + ) + req.Header.Set("Content-Type", "application/x-protobuf") + req.Header.Set("Authorization", "Basic tenant") + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusBadRequest { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + + metrics := scrapeMetrics(t, gw) + if !strings.Contains(metrics, `langfuse_fanout_trace_exports_total{destination="tenant_eu",result="error",tenant_id=""} 1`) { + t.Fatalf("missing invalid trace export metric:\n%s", metrics) + } +} + func newTestGateway(centralURL string, tenants map[string]string) *gateway { return newTestGatewayWithStore(centralURL, tenants, newFakeUploadPlanStore()) } diff --git a/otel/langfuse-fanout/cmd/langfuse-fanout/metrics.go b/otel/langfuse-fanout/cmd/langfuse-fanout/metrics.go index 22259732e8..d3bb05abbf 100644 --- a/otel/langfuse-fanout/cmd/langfuse-fanout/metrics.go +++ b/otel/langfuse-fanout/cmd/langfuse-fanout/metrics.go @@ -55,7 +55,7 @@ func newGatewayMetrics() *gatewayMetrics { traceExports: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "langfuse_fanout_trace_exports_total", Help: "Total trace export attempts through the Langfuse fanout gateway.", - }, []string{"destination", "result"}), + }, []string{"destination", "result", "tenant_id"}), mediaDivergence: prometheus.NewCounterVec(prometheus.CounterOpts{ Name: "langfuse_fanout_media_divergence_total", Help: "Media fanout upstream response divergence by kind. Values are counts only; no media IDs or URLs are exposed.", @@ -125,11 +125,19 @@ func (m *gatewayMetrics) recordUpstream(operation string, destination string, st m.upstreamDuration.With(labels).Observe(duration.Seconds()) } -func (m *gatewayMetrics) recordTraceExport(destination string, result string) { +func (m *gatewayMetrics) recordTraceExport(destination string, result string, tenantID string) { if m == nil { return } - m.traceExports.WithLabelValues(normalizeMetricLabel(destination), result).Inc() + m.traceExports.WithLabelValues(normalizeMetricLabel(destination), result, tenantMetricLabel(tenantID)).Inc() +} + +func tenantMetricLabel(tenantID string) string { + tenantID = strings.TrimSpace(tenantID) + if tenantID == "" { + return "unknown" + } + return tenantID } func (m *gatewayMetrics) recordMediaDivergence(kind string, destination string) { diff --git a/packages/api/src/admin/langfuse.handler.spec.ts b/packages/api/src/admin/langfuse.handler.spec.ts index 474029cab1..2bc5ecb7a1 100644 --- a/packages/api/src/admin/langfuse.handler.spec.ts +++ b/packages/api/src/admin/langfuse.handler.spec.ts @@ -106,6 +106,7 @@ function createHandlers(overrides = {}) { ), getMessages: jest.fn().mockResolvedValue([]), invalidateConfigCaches: jest.fn().mockResolvedValue(undefined), + recordConnectionUpdate: jest.fn(), ...overrides, }; const handlers = createAdminLangfuseHandlers(deps); @@ -529,6 +530,18 @@ describe('createAdminLangfuseHandlers', () => { expect(fields['langfuse.projectId']).toBe('project-1'); expect(res.body?.secretKey).toBeUndefined(); expect(deps.invalidateConfigCaches).toHaveBeenCalledWith('t1'); + expect(deps.recordConnectionUpdate).toHaveBeenCalledWith({ + event_name: 'librechat.langfuse.connection.changed', + tenant_id: 't1', + configured: true, + enabled: true, + destination: 'eu', + change: 'created', + changes: ['created'], + verification_result: 'success', + }); + expect(JSON.stringify(deps.recordConnectionUpdate.mock.calls)).not.toContain('sk-lf-secret'); + expect(JSON.stringify(deps.recordConnectionUpdate.mock.calls)).not.toContain('pk-lf-1'); }); it('requires a new secret when connection fields change', async () => { @@ -615,6 +628,7 @@ describe('createAdminLangfuseHandlers', () => { error: 'Langfuse rejected these keys. Check the destination and keys', }); expect(deps.patchConfigFields).not.toHaveBeenCalled(); + expect(deps.recordConnectionUpdate).not.toHaveBeenCalled(); }); it('rejects credentials when Langfuse does not return a stable project identity', async () => { @@ -667,6 +681,14 @@ describe('createAdminLangfuseHandlers', () => { expect(deps.patchConfigFields).toHaveBeenCalledTimes(1); expect(deps.patchConfigFields.mock.calls[0][3]['langfuse.enabled']).toBe(true); expect(deps.patchConfigFields.mock.calls[0][3]['langfuse.projectId']).toBe('project-1'); + expect(deps.recordConnectionUpdate).toHaveBeenCalledWith( + expect.objectContaining({ + tenant_id: 't1', + change: 'enabled', + changes: ['enabled'], + verification_result: 'skipped', + }), + ); }); it('allows an existing connection to be disabled after its destination is removed', async () => { diff --git a/packages/api/src/admin/langfuse.ts b/packages/api/src/admin/langfuse.ts index 6a77fd6568..9cbe10b507 100644 --- a/packages/api/src/admin/langfuse.ts +++ b/packages/api/src/admin/langfuse.ts @@ -28,6 +28,26 @@ const DEFAULT_PRIORITY = 10; const ENCRYPTED_PREFIX = 'v3:'; const LANGFUSE_VERIFICATION_TIMEOUT_MS = 10_000; +type LangfuseConnectionChange = + | 'created' + | 'credentials_rotated' + | 'destination_changed' + | 'disabled' + | 'enabled' + | 'updated'; +type LangfuseConnectionChanges = [LangfuseConnectionChange, ...LangfuseConnectionChange[]]; + +export interface LangfuseConnectionEvent { + event_name: 'librechat.langfuse.connection.changed'; + tenant_id?: string; + configured: boolean; + enabled: boolean; + destination?: string; + change: LangfuseConnectionChange; + changes: LangfuseConnectionChange[]; + verification_result: 'skipped' | 'success'; +} + export interface AdminLangfuseDeps { findConfigByPrincipal: ( principalType: PrincipalType, @@ -51,6 +71,7 @@ export interface AdminLangfuseDeps { ) => Promise; getMessages: MessageMethods['getMessages']; invalidateConfigCaches?: (tenantId?: string) => Promise; + recordConnectionUpdate?: (event: LangfuseConnectionEvent) => void; } function getTenantId(req: ServerRequest): string | undefined { @@ -79,6 +100,33 @@ function buildStatus(config: IConfig | null): TLangfuseConnectionStatus { }; } +function getConnectionChanges( + stored: TCustomConfig['langfuse'], + enabled: boolean, + destination: string, + publicKey: string, + secretKey: string, +): LangfuseConnectionChanges { + if (!stored?.publicKey || !stored.secretKey) { + return ['created']; + } + const changes: LangfuseConnectionChange[] = []; + if (stored.destination !== destination) { + changes.push('destination_changed'); + } + if (stored.publicKey !== publicKey || secretKey !== '') { + changes.push('credentials_rotated'); + } + if (stored.enabled !== true && enabled) { + changes.push('enabled'); + } + if (stored.enabled === true && !enabled) { + changes.push('disabled'); + } + const [change, ...additionalChanges] = changes; + return change ? [change, ...additionalChanges] : ['updated']; +} + function rejectWhenConnectionUnavailable(res: Response): Response | undefined { if (isLangfuseConnectionAvailable()) { return undefined; @@ -243,6 +291,8 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): { toggleConfigActive, getMessages, invalidateConfigCaches, + recordConnectionUpdate = (event) => + logger.info({ message: '[adminLangfuse] Connection updated', ...event }), } = deps; function findBaseConfig(options?: { includeInactive?: boolean }): Promise { @@ -415,11 +465,30 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): { updated = await toggleConfigActive(PrincipalType.ROLE, BASE_CONFIG_PRINCIPAL_ID, true); } + const status = buildStatus(updated ?? existing); + const changes = getConnectionChanges( + stored, + enabled, + persistedDestination, + publicKey, + secretKey, + ); + recordConnectionUpdate({ + event_name: 'librechat.langfuse.connection.changed', + tenant_id: getTenantId(req), + configured: status.configured, + enabled: status.enabled, + destination: status.destination, + change: changes[0], + changes, + verification_result: connectionChanged ? 'success' : 'skipped', + }); + invalidateConfigCaches?.(getTenantId(req))?.catch((err) => logger.error('[adminLangfuse] Cache invalidation failed after update:', err), ); - return res.status(200).json(buildStatus(updated ?? existing)); + return res.status(200).json(status); } catch (error) { logger.error('[adminLangfuse] updateConnection error:', error); return res.status(500).json({ error: 'Failed to update Langfuse connection' }); diff --git a/packages/api/src/agents/__tests__/run-summarization.test.ts b/packages/api/src/agents/__tests__/run-summarization.test.ts index 43d7c810ab..7327148dfd 100644 --- a/packages/api/src/agents/__tests__/run-summarization.test.ts +++ b/packages/api/src/agents/__tests__/run-summarization.test.ts @@ -2096,10 +2096,19 @@ async function callAndCaptureRunConfig({ // --------------------------------------------------------------------------- // Suite: Langfuse run config // --------------------------------------------------------------------------- +const exportTelemetry = (plan: string, reason: string, tenantId?: string) => ({ + ...(tenantId ? { 'librechat.tenant.id': tenantId } : {}), + 'librechat.langfuse.export_plan': plan, + 'librechat.langfuse.export_reason': reason, +}); + describe('Langfuse run config', () => { it('passes deterministic Langfuse trace config without tenant metadata by default', async () => { const callArgs = await callAndCaptureRunConfig(); - expect(callArgs.langfuse).toEqual({ deterministicTraceId: true }); + expect(callArgs.langfuse).toEqual({ + deterministicTraceId: true, + librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled'), + }); }); it('adds the explicit request tenant id to Langfuse trace metadata and tags', async () => { @@ -2111,6 +2120,7 @@ describe('Langfuse run config', () => { }); expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, + librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled', 'tenant-1'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2124,6 +2134,7 @@ describe('Langfuse run config', () => { }); expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, + librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled', 'tenant-2'), metadata: { 'librechat.tenant.id': 'tenant-2' }, tags: ['tenant:tenant-2'], }); @@ -2152,6 +2163,7 @@ describe('Langfuse run config', () => { baseUrl: 'http://langfuse-fanout-collector:4318/tenant/eu', metadata: { 'librechat.tenant.id': 'tenant-1' }, librechatTraceAttributes: { + ...exportTelemetry('tenant_fanout', 'configured', 'tenant-1'), 'librechat.langfuse.tenant_export.enabled': 'true', 'librechat.langfuse.destination': 'eu', }, @@ -2181,6 +2193,7 @@ describe('Langfuse run config', () => { publicKey: 'pk-central', secretKey: 'sk-central', baseUrl: 'https://central.langfuse.example', + librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled', 'tenant-1'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2207,6 +2220,11 @@ describe('Langfuse run config', () => { expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry( + 'central_only', + 'destination_unconfigured', + 'tenant-1', + ), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2234,6 +2252,7 @@ describe('Langfuse run config', () => { baseUrl: 'http://collector-from-env:4318/tenant/us', metadata: { 'librechat.tenant.id': 'tenant-1' }, librechatTraceAttributes: { + ...exportTelemetry('tenant_fanout', 'configured', 'tenant-1'), 'librechat.langfuse.tenant_export.enabled': 'true', 'librechat.langfuse.destination': 'us', }, @@ -2284,6 +2303,7 @@ describe('Langfuse run config', () => { secretKey: 'sk-tenant-1', baseUrl: 'http://collector-from-env:4318/tenant/us', librechatTraceAttributes: { + ...exportTelemetry('tenant_fanout', 'configured', 'tenant-1'), 'librechat.langfuse.tenant_export.enabled': 'true', 'librechat.langfuse.destination': 'us', }, @@ -2317,6 +2337,7 @@ describe('Langfuse run config', () => { publicKey: 'pk-central', secretKey: 'sk-central', baseUrl: 'https://central.langfuse.example', + librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled', 'tenant-1'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2346,10 +2367,10 @@ describe('Langfuse run config', () => { publicKey: 'pk-central', secretKey: 'sk-central', baseUrl: 'https://central.langfuse.example', + librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled', 'tenant-1'), }); expect(callArgs.langfuse).not.toMatchObject({ baseUrl: 'http://collector-from-env:4318/tenant/eu', - librechatTraceAttributes: expect.any(Object), }); }); @@ -2376,6 +2397,11 @@ describe('Langfuse run config', () => { publicKey: 'pk-central', secretKey: 'sk-central', baseUrl: 'https://central.langfuse.example', + librechatTraceAttributes: exportTelemetry( + 'central_only', + 'collector_unconfigured', + 'tenant-1', + ), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2404,6 +2430,11 @@ describe('Langfuse run config', () => { expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry( + 'central_only', + 'destination_unconfigured', + 'tenant-1', + ), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2426,6 +2457,7 @@ describe('Langfuse run config', () => { expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'missing_credentials', 'tenant-1'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2445,6 +2477,7 @@ describe('Langfuse run config', () => { expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled', 'tenant-1'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2472,6 +2505,7 @@ describe('Langfuse run config', () => { expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'emergency_disabled', 'tenant-1'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2504,6 +2538,7 @@ describe('Langfuse run config', () => { secretKey: 'sk-tenant-1', tags: ['tenant:tenant-1'], librechatTraceAttributes: { + ...exportTelemetry('tenant_fanout', 'configured', 'tenant-1'), 'librechat.langfuse.tenant_export.enabled': 'true', 'librechat.langfuse.destination': 'eu', }, @@ -2534,6 +2569,7 @@ describe('Langfuse run config', () => { expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'emergency_disabled', 'tenant-1'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2569,6 +2605,7 @@ describe('Langfuse run config', () => { secretKey: 'sk-tenant-1', tags: ['tenant:tenant-1'], librechatTraceAttributes: { + ...exportTelemetry('tenant_fanout', 'configured', 'tenant-1'), 'librechat.langfuse.tenant_export.enabled': 'true', 'librechat.langfuse.destination': 'eu', }, @@ -2594,6 +2631,7 @@ describe('Langfuse run config', () => { expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled', 'tenant-1'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -2617,6 +2655,7 @@ describe('Langfuse run config', () => { expect(callArgs.langfuse).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled', 'tenant-1'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); diff --git a/packages/api/src/langfuse/config.spec.ts b/packages/api/src/langfuse/config.spec.ts index 76c53bb7d0..857a84bfe2 100644 --- a/packages/api/src/langfuse/config.spec.ts +++ b/packages/api/src/langfuse/config.spec.ts @@ -4,6 +4,11 @@ process.env.CREDS_KEY = process.env.CREDS_KEY ?? '0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef'; const CENTRAL_EXPORT_ATTRIBUTE = 'librechat.langfuse.central_export.enabled'; +const exportTelemetry = (plan: string, reason: string) => ({ + 'librechat.langfuse.export_plan': plan, + 'librechat.langfuse.export_reason': reason, + 'librechat.tenant.id': 'tenant-1', +}); const envKeys = [ 'LANGFUSE_PUBLIC_KEY', 'LANGFUSE_SECRET_KEY', @@ -181,6 +186,7 @@ describe('buildLangfuseConfig', () => { secretKey: 'sk-tenant-1', baseUrl: 'http://langfuse-fanout-collector:4318/tenant/eu', librechatTraceAttributes: { + ...exportTelemetry('tenant_fanout', 'configured'), 'librechat.langfuse.tenant_export.enabled': 'true', 'librechat.langfuse.destination': 'eu', }, @@ -208,6 +214,7 @@ describe('buildLangfuseConfig', () => { expect(config).toEqual({ deterministicTraceId: true, baseUrl: 'http://langfuse-fanout-collector:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'missing_credentials'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -233,6 +240,7 @@ describe('buildLangfuseConfig', () => { expect(config).toEqual({ deterministicTraceId: true, baseUrl: 'http://langfuse-fanout-collector:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'missing_credentials'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -249,11 +257,49 @@ describe('buildLangfuseConfig', () => { publicKey: 'pk-central', secretKey: 'sk-central', baseUrl: 'https://central.langfuse.example', + librechatTraceAttributes: exportTelemetry('central_only', 'fanout_disabled'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); }); + it('records when fanout is enabled without a collector URL', async () => { + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_PUBLIC_KEY = 'pk-central'; + process.env.LANGFUSE_SECRET_KEY = 'sk-central'; + const { buildLangfuseConfig } = await import('./config'); + + expect(buildLangfuseConfig({ tenantId: 'tenant-1' })).toMatchObject({ + publicKey: 'pk-central', + secretKey: 'sk-central', + librechatTraceAttributes: exportTelemetry('central_only', 'collector_unconfigured'), + }); + }); + + it('records when an enabled tenant connection has an unknown destination', async () => { + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector-from-env:4318'; + const { encryptV3 } = await import('@librechat/data-schemas'); + const { buildLangfuseConfig } = await import('./config'); + + expect( + buildLangfuseConfig({ + tenantId: 'tenant-1', + appConfig: { + langfuse: { + enabled: true, + publicKey: 'pk-tenant-1', + secretKey: encryptV3('sk-tenant-1'), + destination: 'unknown-destination', + }, + } as unknown as AppConfig, + }), + ).toMatchObject({ + baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'destination_unconfigured'), + }); + }); + it('disables direct central tracing when central export is disabled', async () => { process.env.LANGFUSE_PUBLIC_KEY = 'pk-central'; process.env.LANGFUSE_SECRET_KEY = 'sk-central'; @@ -271,6 +317,7 @@ describe('buildLangfuseConfig', () => { enabled: false, librechatTraceAttributes: { [CENTRAL_EXPORT_ATTRIBUTE]: 'false', + ...exportTelemetry('disabled', 'fanout_disabled'), }, tags: ['tenant:tenant-1'], }); @@ -294,6 +341,7 @@ describe('buildLangfuseConfig', () => { enabled: false, librechatTraceAttributes: { [CENTRAL_EXPORT_ATTRIBUTE]: 'false', + ...exportTelemetry('disabled', 'tenant_disabled'), }, tags: ['tenant:tenant-1'], }); @@ -326,6 +374,7 @@ describe('buildLangfuseConfig', () => { metadata: { 'librechat.tenant.id': 'tenant-1' }, librechatTraceAttributes: { [CENTRAL_EXPORT_ATTRIBUTE]: 'false', + ...exportTelemetry('tenant_fanout', 'configured'), 'librechat.langfuse.tenant_export.enabled': 'true', 'librechat.langfuse.destination': 'us', }, @@ -359,6 +408,7 @@ describe('buildLangfuseConfig', () => { enabled: false, librechatTraceAttributes: { [CENTRAL_EXPORT_ATTRIBUTE]: 'false', + ...exportTelemetry('disabled', 'emergency_disabled'), }, tags: ['tenant:tenant-1'], }); @@ -385,6 +435,7 @@ describe('buildLangfuseConfig', () => { ).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -410,6 +461,7 @@ describe('buildLangfuseConfig', () => { ).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -436,6 +488,7 @@ describe('buildLangfuseConfig', () => { enabled: false, librechatTraceAttributes: { [CENTRAL_EXPORT_ATTRIBUTE]: 'false', + ...exportTelemetry('disabled', 'tenant_disabled'), }, tags: ['tenant:tenant-1'], }); @@ -453,6 +506,7 @@ describe('buildLangfuseConfig', () => { deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', mediaUploadEnabled: false, + librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); @@ -470,6 +524,7 @@ describe('buildLangfuseConfig', () => { expect(buildLangfuseConfig({ tenantId: 'tenant-1' })).toEqual({ deterministicTraceId: true, baseUrl: 'http://collector-from-env:4318', + librechatTraceAttributes: exportTelemetry('central_only', 'tenant_disabled'), metadata: { 'librechat.tenant.id': 'tenant-1' }, tags: ['tenant:tenant-1'], }); diff --git a/packages/api/src/langfuse/config.ts b/packages/api/src/langfuse/config.ts index b5232ac1fc..b9dc1f3bb6 100644 --- a/packages/api/src/langfuse/config.ts +++ b/packages/api/src/langfuse/config.ts @@ -22,10 +22,17 @@ type LangfuseRunConfigWithTraceAttributes = LangfuseRunConfig & { additionalHeaders?: Record; }; type LangfuseTenantDestination = NonNullable>; +type TenantExportBlockReason = + | 'collector_unconfigured' + | 'destination_unconfigured' + | 'emergency_disabled' + | 'fanout_disabled' + | 'missing_credentials' + | 'tenant_disabled'; type LangfuseExportPlan = - | { type: 'directCentral' } - | { type: 'disabled' } - | { type: 'fanoutCollector'; collectorUrl: string } + | { type: 'directCentral'; reason: 'collector_unconfigured' | 'fanout_disabled' } + | { type: 'disabled'; reason: TenantExportBlockReason } + | { type: 'fanoutCollector'; collectorUrl: string; reason: TenantExportBlockReason } | { type: 'tenantFanout'; collectorUrl: string; @@ -36,6 +43,9 @@ type LangfuseExportPlan = const TENANT_EXPORT_ATTRIBUTE = 'librechat.langfuse.tenant_export.enabled'; const TENANT_DESTINATION_ATTRIBUTE = 'librechat.langfuse.destination'; const CENTRAL_EXPORT_ATTRIBUTE = 'librechat.langfuse.central_export.enabled'; +const EXPORT_PLAN_ATTRIBUTE = 'librechat.langfuse.export_plan'; +const EXPORT_REASON_ATTRIBUTE = 'librechat.langfuse.export_reason'; +const TENANT_ID_ATTRIBUTE = 'librechat.tenant.id'; const CENTRAL_MEDIA_DISABLED_SEGMENT = 'central-media-disabled'; const DEFAULT_BASE_URL = 'https://cloud.langfuse.com'; @@ -108,29 +118,94 @@ function disableCentralExport(langfuse: LangfuseRunConfigWithTraceAttributes): v }; } +function getTenantExportBlockReason({ + tenantLangfuseEnabled, + hasTenantCredentials, + tenantExportEmergencyEnabled, + tenantDestination, +}: { + tenantLangfuseEnabled: boolean; + hasTenantCredentials: boolean; + tenantExportEmergencyEnabled: boolean; + tenantDestination?: LangfuseTenantDestination; +}): TenantExportBlockReason { + if (!tenantLangfuseEnabled) { + return 'tenant_disabled'; + } + if (!hasTenantCredentials) { + return 'missing_credentials'; + } + if (!tenantExportEmergencyEnabled) { + return 'emergency_disabled'; + } + if (tenantDestination == null) { + return 'destination_unconfigured'; + } + return 'missing_credentials'; +} + +function applyExportPlanTelemetry( + langfuse: LangfuseRunConfigWithTraceAttributes, + exportPlan: LangfuseExportPlan, + tenantId?: string, +): void { + let exportPlanName = 'central_only'; + if (exportPlan.type === 'tenantFanout') { + exportPlanName = 'tenant_fanout'; + } else if (exportPlan.type === 'disabled') { + exportPlanName = 'disabled'; + } + const exportReason = exportPlan.type === 'tenantFanout' ? 'configured' : exportPlan.reason; + + langfuse.librechatTraceAttributes = { + ...(langfuse.librechatTraceAttributes ?? {}), + ...(tenantId ? { [TENANT_ID_ATTRIBUTE]: tenantId } : {}), + [EXPORT_PLAN_ATTRIBUTE]: exportPlanName, + [EXPORT_REASON_ATTRIBUTE]: exportReason, + }; +} + function resolveLangfuseExportPlan({ centralTraceExportEnabled, fanoutEnabled, + fanoutRequested, fanoutCollectorUrl, - tenantExportEnabled, + tenantLangfuseEnabled, + hasTenantCredentials, + tenantExportEmergencyEnabled, publicKey, secretKey, tenantDestination, }: { centralTraceExportEnabled: boolean; fanoutEnabled: boolean; + fanoutRequested: boolean; fanoutCollectorUrl?: string; - tenantExportEnabled: boolean; + tenantLangfuseEnabled: boolean; + hasTenantCredentials: boolean; + tenantExportEmergencyEnabled: boolean; publicKey?: string; secretKey?: string; tenantDestination?: LangfuseTenantDestination; }): LangfuseExportPlan { if (!fanoutEnabled || fanoutCollectorUrl == null) { - return centralTraceExportEnabled ? { type: 'directCentral' } : { type: 'disabled' }; + const reason = fanoutRequested ? 'collector_unconfigured' : 'fanout_disabled'; + if (centralTraceExportEnabled) { + return { + type: 'directCentral', + reason, + }; + } + return { type: 'disabled', reason }; } const canRouteTenantFanout = - tenantExportEnabled && publicKey != null && secretKey != null && tenantDestination != null; + tenantLangfuseEnabled && + hasTenantCredentials && + tenantExportEmergencyEnabled && + publicKey != null && + secretKey != null && + tenantDestination != null; if (canRouteTenantFanout) { return { @@ -145,10 +220,27 @@ function resolveLangfuseExportPlan({ // Direct central export can use the collector normally. Central-suppressed // runs only reach the collector through a concrete tenant fanout route. if (centralTraceExportEnabled) { - return { type: 'fanoutCollector', collectorUrl: fanoutCollectorUrl }; + return { + type: 'fanoutCollector', + collectorUrl: fanoutCollectorUrl, + reason: getTenantExportBlockReason({ + tenantLangfuseEnabled, + hasTenantCredentials, + tenantExportEmergencyEnabled, + tenantDestination, + }), + }; } - return { type: 'disabled' }; + return { + type: 'disabled', + reason: getTenantExportBlockReason({ + tenantLangfuseEnabled, + hasTenantCredentials, + tenantExportEmergencyEnabled, + tenantDestination, + }), + }; } export function buildLangfuseConfig({ @@ -200,6 +292,7 @@ export function buildLangfuseConfig({ const tenantCredentials = resolveTenantCredentials(config); const hasTenantCredentials = Boolean(tenantCredentials); const fanoutEnabled = isLangfuseFanoutEnabled(); + const fanoutRequested = normalizeBoolean(process.env.LANGFUSE_FANOUT_ENABLED) === true; const fanoutCollectorUrl = normalizeString(process.env.LANGFUSE_FANOUT_COLLECTOR_URL); const tenantDestination = resolveLangfuseTenantDestination(config?.destination); const tenantExportEmergencyEnabled = isLangfuseTenantExportEnabled(); @@ -222,13 +315,16 @@ export function buildLangfuseConfig({ const exportPlan = resolveLangfuseExportPlan({ centralTraceExportEnabled, fanoutEnabled, + fanoutRequested, fanoutCollectorUrl, - tenantExportEnabled: - tenantLangfuseEnabled && hasTenantCredentials && tenantExportEmergencyEnabled, + tenantLangfuseEnabled, + hasTenantCredentials, + tenantExportEmergencyEnabled, publicKey: tenantCredentials?.publicKey, secretKey: tenantCredentials?.secretKey, tenantDestination, }); + applyExportPlanTelemetry(langfuse, exportPlan, normalizedTenantId); switch (exportPlan.type) { case 'tenantFanout': diff --git a/packages/data-schemas/src/config/requestLogContext.spec.ts b/packages/data-schemas/src/config/requestLogContext.spec.ts index 6469a2c23e..e2039fdadf 100644 --- a/packages/data-schemas/src/config/requestLogContext.spec.ts +++ b/packages/data-schemas/src/config/requestLogContext.spec.ts @@ -64,6 +64,32 @@ describe('attachRequestContext', () => { }); })); + it('renders tenant Langfuse connection events with identity and outcome fields', () => + tenantStorage.run(context, () => { + const result = attachRequestContext({ + level: 'info', + message: '[adminLangfuse] Connection updated', + event_name: 'librechat.langfuse.connection.changed', + tenant_id: 'tenant-123', + configured: true, + enabled: true, + destination: 'eu', + change: 'created', + changes: ['created'], + verification_result: 'success', + }); + + const rendered = formatLogContext(result); + expect(rendered).toContain('"event_name":"librechat.langfuse.connection.changed"'); + expect(rendered).toContain('"tenant_id":"tenant-123"'); + expect(rendered).toContain('"configured":true'); + expect(rendered).toContain('"enabled":true'); + expect(rendered).toContain('"destination":"eu"'); + expect(rendered).toContain('"change":"created"'); + expect(rendered).toContain('"changes":["created"]'); + expect(rendered).toContain('"verification_result":"success"'); + })); + it('keeps application paths separate from the safe request route', () => tenantStorage.run(context, () => { const result = attachRequestContext({ diff --git a/packages/data-schemas/src/config/requestLogContext.ts b/packages/data-schemas/src/config/requestLogContext.ts index 11372c790d..1dc715f80d 100644 --- a/packages/data-schemas/src/config/requestLogContext.ts +++ b/packages/data-schemas/src/config/requestLogContext.ts @@ -26,6 +26,13 @@ const RESERVED_REQUEST_LOG_CONTEXT_KEYS = new Set([ const STRUCTURED_EVENT_LOG_CONTEXT_KEYS = [ 'event_name', + 'tenant_id', + 'configured', + 'enabled', + 'destination', + 'change', + 'changes', + 'verification_result', 'auth_strategy', 'primary_strategy', 'fallback_strategy', @@ -63,6 +70,10 @@ const IDENTITY_FREE_EVENT_NAMES = new Set([ 'jwt_auth_recovered', 'tenant_isolation_error', ]); +const STRUCTURED_EVENT_NAMES = new Set([ + ...IDENTITY_FREE_EVENT_NAMES, + 'librechat.langfuse.connection.changed', +]); const STRUCTURED_EVENT_LOG_CONTEXT_KEY_SET = new Set(STRUCTURED_EVENT_LOG_CONTEXT_KEYS); const MAX_LOG_CONTEXT_ARRAY_LENGTH = 10; @@ -75,6 +86,10 @@ function isIdentityFreeEvent(eventName: unknown): boolean { return typeof eventName === 'string' && IDENTITY_FREE_EVENT_NAMES.has(eventName); } +function isStructuredEvent(eventName: unknown): boolean { + return typeof eventName === 'string' && STRUCTURED_EVENT_NAMES.has(eventName); +} + function getLogTenantId(): string | undefined { const tenantId = getTenantId(); return tenantId === SYSTEM_TENANT_ID ? undefined : tenantId; @@ -104,9 +119,10 @@ function normalizeLogContextValue(value: unknown): LogContextValue | undefined { export function formatLogContext(info: LogContextInfo): string { const context: Partial> = {}; const omitIdentity = isIdentityFreeEvent(info.event_name); + const includeStructuredEvent = isStructuredEvent(info.event_name); LOG_CONTEXT_KEYS.forEach((key) => { - if (STRUCTURED_EVENT_LOG_CONTEXT_KEY_SET.has(key) && !omitIdentity) { + if (STRUCTURED_EVENT_LOG_CONTEXT_KEY_SET.has(key) && !includeStructuredEvent) { return; } if (omitIdentity && (key === 'tenantId' || key === 'userId')) {