From 8cfe4d8d07cb86548c075ccfd496601ba535192a Mon Sep 17 00:00:00 2001 From: Ravi Kumar L Date: Sun, 12 Jul 2026 14:03:27 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=AA=AD=20feat:=20support=20per-run=20cent?= =?UTF-8?q?ral=20langfuse=20export=20suppression=20(#14207)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 3 + deploy-compose.langfuse-fanout.yml | 1 + docker-compose.langfuse-fanout.yml | 1 + .../templates/langfuse-fanout-configmap.yaml | 9 +- otel/langfuse-fanout/README.md | 10 +- .../cmd/langfuse-fanout/main.go | 33 +++- .../cmd/langfuse-fanout/main_test.go | 137 +++++++++++++- otel/langfuse-fanout/otelcol.yaml | 9 +- packages/api/src/agents/run.ts | 12 +- packages/api/src/langfuse/config.spec.ts | 169 ++++++++++++++++++ packages/api/src/langfuse/config.ts | 142 ++++++++++++--- 11 files changed, 488 insertions(+), 38 deletions(-) create mode 100644 packages/api/src/langfuse/config.spec.ts diff --git a/.env.example b/.env.example index f61df95dee..44821dc0fe 100644 --- a/.env.example +++ b/.env.example @@ -152,6 +152,9 @@ NODE_MAX_OLD_SPACE_SIZE=6144 # Gateway-only Basic auth header for central trace/media export. LibreChat feedback # scores use LANGFUSE_PUBLIC_KEY/LANGFUSE_SECRET_KEY instead. # LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=Basic +# Set true on the gateway to disable central media export while leaving central +# trace export unchanged. +# LANGFUSE_FANOUT_CENTRAL_MEDIA_EXPORT_DISABLED=false # Compose's included gateway config supports the three listed destination keys. # Add custom keys only when the gateway is started with matching destination URLs. # LANGFUSE_FANOUT_TENANT_DESTINATIONS=eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com diff --git a/deploy-compose.langfuse-fanout.yml b/deploy-compose.langfuse-fanout.yml index 15f57c1d88..b4cf4424a1 100644 --- a/deploy-compose.langfuse-fanout.yml +++ b/deploy-compose.langfuse-fanout.yml @@ -25,6 +25,7 @@ services: environment: - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com} - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header} + - LANGFUSE_FANOUT_CENTRAL_MEDIA_EXPORT_DISABLED=${LANGFUSE_FANOUT_CENTRAL_MEDIA_EXPORT_DISABLED:-false} - LANGFUSE_FANOUT_TENANT_DESTINATIONS=${LANGFUSE_FANOUT_TENANT_DESTINATIONS:-eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com} - LANGFUSE_FANOUT_UPSTREAM_TIMEOUT=${LANGFUSE_FANOUT_UPSTREAM_TIMEOUT:-30s} - LANGFUSE_FANOUT_PUBLIC_URL=${LANGFUSE_FANOUT_PUBLIC_URL:-http://langfuse-fanout-collector:4318} diff --git a/docker-compose.langfuse-fanout.yml b/docker-compose.langfuse-fanout.yml index 15f57c1d88..b4cf4424a1 100644 --- a/docker-compose.langfuse-fanout.yml +++ b/docker-compose.langfuse-fanout.yml @@ -25,6 +25,7 @@ services: environment: - LANGFUSE_FANOUT_CENTRAL_BASE_URL=${LANGFUSE_FANOUT_CENTRAL_BASE_URL:-https://cloud.langfuse.com} - LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=${LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER:?Set LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER to central Langfuse Basic auth header} + - LANGFUSE_FANOUT_CENTRAL_MEDIA_EXPORT_DISABLED=${LANGFUSE_FANOUT_CENTRAL_MEDIA_EXPORT_DISABLED:-false} - LANGFUSE_FANOUT_TENANT_DESTINATIONS=${LANGFUSE_FANOUT_TENANT_DESTINATIONS:-eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com} - LANGFUSE_FANOUT_UPSTREAM_TIMEOUT=${LANGFUSE_FANOUT_UPSTREAM_TIMEOUT:-30s} - LANGFUSE_FANOUT_PUBLIC_URL=${LANGFUSE_FANOUT_PUBLIC_URL:-http://langfuse-fanout-collector:4318} diff --git a/helm/librechat/templates/langfuse-fanout-configmap.yaml b/helm/librechat/templates/langfuse-fanout-configmap.yaml index 2ae8a22ea5..a2f4457e50 100644 --- a/helm/librechat/templates/langfuse-fanout-configmap.yaml +++ b/helm/librechat/templates/langfuse-fanout-configmap.yaml @@ -44,8 +44,15 @@ data: traces: span: - attributes["librechat.langfuse.tenant_export.enabled"] != "true" + filter/central_export: + error_mode: ignore + traces: + span: + - attributes["librechat.langfuse.central_export.enabled"] == "false" attributes/drop_librechat_routing: actions: + - key: librechat.langfuse.central_export.enabled + action: delete - key: librechat.langfuse.tenant_export.enabled action: delete - key: librechat.langfuse.destination @@ -85,7 +92,7 @@ data: pipelines: traces/central: receivers: [otlp] - processors: [memory_limiter, attributes/drop_librechat_routing, batch/central] + processors: [memory_limiter, filter/central_export, attributes/drop_librechat_routing, batch/central] exporters: [otlphttp/central] traces/tenant: receivers: [otlp] diff --git a/otel/langfuse-fanout/README.md b/otel/langfuse-fanout/README.md index f78fa16bab..f3670cff0a 100644 --- a/otel/langfuse-fanout/README.md +++ b/otel/langfuse-fanout/README.md @@ -2,7 +2,7 @@ LibreChat can send tenant-scoped agent traces to a tenant Langfuse project and also copy those traces to a central Langfuse project. When trace payloads -contain Langfuse media references, the gateway also copies the media upload to +contain Langfuse media references, the gateway can also copy the media upload to central and tenant Langfuse storage. This is optional and is disabled unless you explicitly deploy the fanout gateway. @@ -40,6 +40,10 @@ The deployment is a hybrid: central and tenant Langfuse, returning a one-time gateway upload URL, then uploading the received bytes to each upstream presigned upload URL. The SDK's `PATCH /api/public/media/{mediaId}` status call is also fanned out. +- Central media export can be disabled independently of central trace export + with `LANGFUSE_FANOUT_CENTRAL_MEDIA_EXPORT_DISABLED=true`. Per-run central + trace suppression uses a destination-scoped gateway path that also skips + central media export for that run. - Tenant export is conditional. LibreChat uses a destination-scoped gateway URL only when tenant keys are configured, the tenant base URL matches a configured startup destination, and `LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED` is not true. @@ -68,6 +72,9 @@ defined in this gateway config. emergency switch to stop tenant trace and score export while keeping central gateway export active. When omitted, false, or blank, tenant export remains available if tenant keys and a known destination are configured. +- `LANGFUSE_FANOUT_CENTRAL_MEDIA_EXPORT_DISABLED=true` can be set on the gateway + to stop central media create/upload/patch fanout while leaving central trace + export unchanged. - This supports Langfuse Cloud and self-hosted Langfuse as long as each allowed tenant base URL is configured at LibreChat/gateway startup. Runtime tenant config selects from those known destinations; it does not inject arbitrary @@ -114,6 +121,7 @@ LANGFUSE_BASE_URL=https://cloud.langfuse.com # Used by the gateway for central trace and media export. LANGFUSE_FANOUT_CENTRAL_BASE_URL=https://cloud.langfuse.com LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=Basic +LANGFUSE_FANOUT_CENTRAL_MEDIA_EXPORT_DISABLED=false # Compose's included gateway config supports these three destination keys. LANGFUSE_FANOUT_TENANT_DESTINATIONS=eu=https://cloud.langfuse.com,us=https://us.cloud.langfuse.com,jp=https://jp.cloud.langfuse.com LANGFUSE_FANOUT_TRACE_DESTINATION_KEYS=eu,us,jp diff --git a/otel/langfuse-fanout/cmd/langfuse-fanout/main.go b/otel/langfuse-fanout/cmd/langfuse-fanout/main.go index fde79c27d6..a1fcf5e6ec 100644 --- a/otel/langfuse-fanout/cmd/langfuse-fanout/main.go +++ b/otel/langfuse-fanout/cmd/langfuse-fanout/main.go @@ -35,6 +35,7 @@ const ( defaultTraceCollector = "http://127.0.0.1:4319" centralName = "central" tenantPrefix = "/tenant/" + centralMediaDisabled = "central-media-disabled" mediaUploadProxyPath = "/__langfuse-fanout/media-upload/" otelTracePath = "/api/public/otel/v1/traces" mediaPath = "/api/public/media" @@ -49,6 +50,7 @@ type config struct { publicURL string metricsSecret string traceDestinationKeys map[string]bool + centralMediaExport bool central destination tenants map[string]string redis redisConfig @@ -70,8 +72,9 @@ type destination struct { } type route struct { - destination string - path string + destination string + path string + disableCentralMedia bool } type uploadDestination struct { @@ -212,6 +215,7 @@ func loadConfig() (config, error) { publicURL: publicURL, metricsSecret: firstNonEmptyEnv("LANGFUSE_FANOUT_METRICS_SECRET", "METRICS_SECRET"), traceDestinationKeys: traceDestinationKeys, + centralMediaExport: !isTrueEnv("LANGFUSE_FANOUT_CENTRAL_MEDIA_EXPORT_DISABLED"), central: destination{ name: centralName, baseURL: centralBaseURL, @@ -578,7 +582,10 @@ func (g *gateway) handleMediaUpload(w http.ResponseWriter, r *http.Request) { } func (g *gateway) mediaDestinations(route route, tenantAuth string) []destination { - destinations := []destination{g.cfg.central} + destinations := []destination{} + if g.cfg.centralMediaExport && !route.disableCentralMedia { + destinations = append(destinations, g.cfg.central) + } if route.destination == "" { return destinations } @@ -990,7 +997,16 @@ func parseRoute(path string) route { if normalizedDestination == "" { return route{path: path} } - return route{destination: normalizedDestination, path: "/" + suffix} + disableCentralMedia := false + if marker, markerSuffix, ok := strings.Cut(suffix, "/"); ok && marker == centralMediaDisabled { + disableCentralMedia = true + suffix = markerSuffix + } + return route{ + destination: normalizedDestination, + path: "/" + suffix, + disableCentralMedia: disableCentralMedia, + } } func readMaybeGzip(r *http.Request) ([]byte, error) { @@ -1159,6 +1175,15 @@ func firstNonEmptyEnv(keys ...string) string { return "" } +func isTrueEnv(key string) bool { + switch strings.ToLower(strings.TrimSpace(os.Getenv(key))) { + case "1", "true", "yes", "on": + return true + default: + return false + } +} + func parseDurationEnv(key string, fallback time.Duration) time.Duration { value := strings.TrimSpace(os.Getenv(key)) if value == "" { diff --git a/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go b/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go index b64de878e5..fac8edbb45 100644 --- a/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go +++ b/otel/langfuse-fanout/cmd/langfuse-fanout/main_test.go @@ -71,6 +71,22 @@ func TestLoadConfigRequiresRedisURI(t *testing.T) { } } +func TestLoadConfigDisablesCentralMediaExport(t *testing.T) { + t.Setenv("LANGFUSE_FANOUT_CENTRAL_BASE_URL", "https://cloud.langfuse.com") + t.Setenv("LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER", "Basic central") + t.Setenv("LANGFUSE_FANOUT_PUBLIC_URL", "http://fanout.local:4318") + t.Setenv("LANGFUSE_FANOUT_REDIS_URI", "redis://localhost:6379") + t.Setenv("LANGFUSE_FANOUT_CENTRAL_MEDIA_EXPORT_DISABLED", "true") + + cfg, err := loadConfig() + if err != nil { + t.Fatalf("loadConfig error: %v", err) + } + if cfg.centralMediaExport { + t.Fatal("central media export should be disabled") + } +} + func TestNormalizeBaseURLAllowsOnlyHTTPAndHTTPS(t *testing.T) { if got := normalizeBaseURL("http://localhost:3000/path/"); got != "http://localhost:3000/path" { t.Fatalf("http URL normalized to %q", got) @@ -354,6 +370,120 @@ func TestMediaUploadFansOutToCentralAndTenant(t *testing.T) { } } +func TestMediaUploadSkipsCentralForCentralMediaDisabledTenantRoute(t *testing.T) { + t.Parallel() + + var mu sync.Mutex + uploads := map[string]string{} + upstream := func(name string) *httptest.Server { + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodPost && r.URL.Path == mediaPath: + uploadURL := "http://" + r.Host + "/upload/" + name + writeJSON(w, http.StatusCreated, mediaUploadResponse{ + MediaID: "same-media-id", + UploadURL: &uploadURL, + }) + case r.Method == http.MethodPut && r.URL.Path == "/upload/"+name: + body, _ := io.ReadAll(r.Body) + mu.Lock() + uploads[name] = string(body) + mu.Unlock() + w.WriteHeader(http.StatusOK) + case r.Method == http.MethodPatch && r.URL.Path == mediaPath+"/same-media-id": + w.WriteHeader(http.StatusNoContent) + default: + http.NotFound(w, r) + } + })) + } + central := upstream("central") + defer central.Close() + tenant := upstream("tenant") + defer tenant.Close() + + store := newFakeUploadPlanStore() + createGateway := newTestGatewayWithStore(central.URL, map[string]string{"eu": tenant.URL}, store) + uploadGateway := newTestGatewayWithStore(central.URL, map[string]string{"eu": tenant.URL}, store) + createBody := `{"traceId":"trace","contentType":"image/png","contentLength":5,"sha256Hash":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","field":"input"}` + req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu/"+centralMediaDisabled+mediaPath, strings.NewReader(createBody)) + req.Header.Set("Authorization", "Basic tenant") + resp := httptest.NewRecorder() + + createGateway.handle(resp, req) + if resp.Code != http.StatusCreated { + t.Fatalf("create status = %d, body = %s", resp.Code, resp.Body.String()) + } + var create mediaUploadResponse + if err := json.NewDecoder(resp.Body).Decode(&create); err != nil { + t.Fatal(err) + } + if create.MediaID != "same-media-id" || create.UploadURL == nil { + t.Fatalf("unexpected create response: %#v", create) + } + + uploadReq := httptest.NewRequest(http.MethodPut, *create.UploadURL, strings.NewReader("hello")) + uploadReq.Header.Set("Content-Type", "image/png") + uploadResp := httptest.NewRecorder() + uploadGateway.handle(uploadResp, uploadReq) + if uploadResp.Code != http.StatusOK { + t.Fatalf("upload status = %d, body = %s", uploadResp.Code, uploadResp.Body.String()) + } + + patchReq := httptest.NewRequest(http.MethodPatch, tenantPrefix+"eu/"+centralMediaDisabled+mediaPath+"/same-media-id", strings.NewReader(`{"uploadHttpStatus":200}`)) + patchReq.Header.Set("Authorization", "Basic tenant") + patchResp := httptest.NewRecorder() + uploadGateway.handle(patchResp, patchReq) + if patchResp.Code != http.StatusNoContent { + t.Fatalf("patch status = %d, body = %s", patchResp.Code, patchResp.Body.String()) + } + + mu.Lock() + defer mu.Unlock() + if _, ok := uploads["central"]; ok { + t.Fatalf("central upload should be skipped, uploads = %#v", uploads) + } + if uploads["tenant"] != "hello" { + t.Fatalf("tenant upload missing, uploads = %#v", uploads) + } +} + +func TestMediaUploadSkipsCentralWhenCentralMediaExportDisabled(t *testing.T) { + t.Parallel() + + var centralCreates int + central := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + centralCreates++ + http.NotFound(w, r) + })) + defer central.Close() + + var tenantCreates int + tenant := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != mediaPath { + http.NotFound(w, r) + return + } + tenantCreates++ + writeJSON(w, http.StatusCreated, mediaUploadResponse{MediaID: "same-media-id"}) + })) + defer tenant.Close() + + gw := newTestGateway(central.URL, map[string]string{"eu": tenant.URL}) + gw.cfg.centralMediaExport = false + req := httptest.NewRequest(http.MethodPost, tenantPrefix+"eu"+mediaPath, strings.NewReader(`{"contentLength":0}`)) + req.Header.Set("Authorization", "Basic tenant") + resp := httptest.NewRecorder() + + gw.handle(resp, req) + if resp.Code != http.StatusCreated { + t.Fatalf("status = %d, body = %s", resp.Code, resp.Body.String()) + } + if centralCreates != 0 || tenantCreates != 1 { + t.Fatalf("centralCreates=%d tenantCreates=%d", centralCreates, tenantCreates) + } +} + func TestMediaUploadRejectsInvalidIDBeforeReadingBody(t *testing.T) { t.Parallel() @@ -876,9 +1006,10 @@ func newTestGateway(centralURL string, tenants map[string]string) *gateway { func newTestGatewayWithStore(centralURL string, tenants map[string]string, store uploadPlanStore) *gateway { return newGateway(config{ - traceCollectorURL: "http://collector.invalid", - publicURL: "http://fanout.local:4318", - metricsSecret: "test-secret", + traceCollectorURL: "http://collector.invalid", + publicURL: "http://fanout.local:4318", + metricsSecret: "test-secret", + centralMediaExport: true, central: destination{ name: centralName, baseURL: centralURL, diff --git a/otel/langfuse-fanout/otelcol.yaml b/otel/langfuse-fanout/otelcol.yaml index 21b1567d28..2e70476970 100644 --- a/otel/langfuse-fanout/otelcol.yaml +++ b/otel/langfuse-fanout/otelcol.yaml @@ -37,8 +37,15 @@ processors: traces: span: - attributes["librechat.langfuse.tenant_export.enabled"] != "true" + filter/central_export: + error_mode: ignore + traces: + span: + - attributes["librechat.langfuse.central_export.enabled"] == "false" attributes/drop_librechat_routing: actions: + - key: librechat.langfuse.central_export.enabled + action: delete - key: librechat.langfuse.tenant_export.enabled action: delete - key: librechat.langfuse.destination @@ -94,7 +101,7 @@ service: pipelines: traces/central: receivers: [otlp] - processors: [memory_limiter, attributes/drop_librechat_routing, batch/central] + processors: [memory_limiter, filter/central_export, attributes/drop_librechat_routing, batch/central] exporters: [otlphttp/central] traces/tenant: receivers: [otlp] diff --git a/packages/api/src/agents/run.ts b/packages/api/src/agents/run.ts index 7a76fc62b6..331f9f757e 100644 --- a/packages/api/src/agents/run.ts +++ b/packages/api/src/agents/run.ts @@ -933,6 +933,7 @@ export async function createRun({ requestBody, user, tenantId, + centralTraceExportEnabled, tokenCounter, customHandlers, indexTokenCountMap, @@ -954,6 +955,11 @@ export async function createRun({ requestBody?: t.RequestBody; user?: IUser; tenantId?: string; + /** + * Defaults to true. Set false to suppress central Langfuse export for this + * run. Tenant fanout can still export when tenant routing is available. + */ + centralTraceExportEnabled?: boolean; /** Message history for extracting previously discovered tools */ messages?: BaseMessage[]; /** @@ -1374,7 +1380,11 @@ export async function createRun({ // feedback can be scored against the trace without a lookup (see the // feedback route in api/server/routes/messages.js). No-op unless Langfuse // tracing is enabled. Requires @librechat/agents >= 3.2.21. - langfuse: buildLangfuseConfig({ appConfig, tenantId: tenantId ?? user?.tenantId }), + langfuse: buildLangfuseConfig({ + appConfig, + tenantId: tenantId ?? user?.tenantId, + centralTraceExportEnabled, + }), ...(enableToolOutputReferences && { toolOutputReferences: { enabled: true }, }), diff --git a/packages/api/src/langfuse/config.spec.ts b/packages/api/src/langfuse/config.spec.ts new file mode 100644 index 0000000000..942761f1b5 --- /dev/null +++ b/packages/api/src/langfuse/config.spec.ts @@ -0,0 +1,169 @@ +import type { AppConfig } from '@librechat/data-schemas'; + +jest.mock('@librechat/data-schemas', () => ({ + logger: { + debug: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + info: jest.fn(), + }, +})); + +import { buildLangfuseConfig } from './config'; + +const CENTRAL_EXPORT_ATTRIBUTE = 'librechat.langfuse.central_export.enabled'; + +function clearLangfuseEnv() { + delete process.env.LANGFUSE_PUBLIC_KEY; + delete process.env.LANGFUSE_SECRET_KEY; + delete process.env.LANGFUSE_BASE_URL; + delete process.env.LANGFUSE_BASEURL; + delete process.env.LANGFUSE_HOST; + delete process.env.LANGFUSE_FANOUT_ENABLED; + delete process.env.LANGFUSE_FANOUT_COLLECTOR_URL; + delete process.env.LANGFUSE_FANOUT_TENANT_BASE_URL; + delete process.env.LANGFUSE_FANOUT_TENANT_DESTINATIONS; + delete process.env.LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED; +} + +describe('buildLangfuseConfig central export control', () => { + beforeEach(() => { + clearLangfuseEnv(); + }); + + it('keeps central export enabled by default', () => { + process.env.LANGFUSE_PUBLIC_KEY = 'pk-central'; + process.env.LANGFUSE_SECRET_KEY = 'sk-central'; + process.env.LANGFUSE_BASE_URL = 'https://central.langfuse.example'; + + expect(buildLangfuseConfig({ tenantId: 'tenant-1' })).toEqual({ + deterministicTraceId: true, + publicKey: 'pk-central', + secretKey: 'sk-central', + baseUrl: 'https://central.langfuse.example', + metadata: { 'librechat.tenant.id': 'tenant-1' }, + tags: ['tenant:tenant-1'], + }); + }); + + it('disables direct central tracing when central export is disabled', () => { + process.env.LANGFUSE_PUBLIC_KEY = 'pk-central'; + process.env.LANGFUSE_SECRET_KEY = 'sk-central'; + process.env.LANGFUSE_BASE_URL = 'https://central.langfuse.example'; + + expect( + buildLangfuseConfig({ + tenantId: 'tenant-1', + centralTraceExportEnabled: false, + }), + ).toEqual({ + deterministicTraceId: true, + metadata: { 'librechat.tenant.id': 'tenant-1' }, + enabled: false, + librechatTraceAttributes: { + [CENTRAL_EXPORT_ATTRIBUTE]: 'false', + }, + tags: ['tenant:tenant-1'], + }); + }); + + it('does not emit central-suppressed traces when there is no tenant fanout route', () => { + process.env.LANGFUSE_PUBLIC_KEY = 'pk-central'; + process.env.LANGFUSE_SECRET_KEY = 'sk-central'; + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector-from-env:4318'; + + expect( + buildLangfuseConfig({ + tenantId: 'tenant-1', + centralTraceExportEnabled: false, + }), + ).toEqual({ + deterministicTraceId: true, + metadata: { 'librechat.tenant.id': 'tenant-1' }, + enabled: false, + librechatTraceAttributes: { + [CENTRAL_EXPORT_ATTRIBUTE]: 'false', + }, + tags: ['tenant:tenant-1'], + }); + }); + + it('routes tenant fanout traces while marking central export disabled', () => { + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector-from-env:4318'; + + expect( + buildLangfuseConfig({ + tenantId: 'tenant-1', + centralTraceExportEnabled: false, + appConfig: { + langfuse: { + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + baseUrl: 'https://us.cloud.langfuse.com', + }, + } as AppConfig, + }), + ).toMatchObject({ + deterministicTraceId: true, + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + baseUrl: 'http://collector-from-env:4318/tenant/us/central-media-disabled', + metadata: { 'librechat.tenant.id': 'tenant-1' }, + librechatTraceAttributes: { + [CENTRAL_EXPORT_ATTRIBUTE]: 'false', + 'librechat.langfuse.tenant_export.enabled': 'true', + 'librechat.langfuse.destination': 'us', + }, + tags: ['tenant:tenant-1'], + }); + }); + + it('does not emit central-suppressed traces when tenant fanout is emergency-disabled', () => { + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector-from-env:4318'; + process.env.LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED = 'true'; + + expect( + buildLangfuseConfig({ + tenantId: 'tenant-1', + centralTraceExportEnabled: false, + appConfig: { + langfuse: { + publicKey: 'pk-tenant-1', + secretKey: 'sk-tenant-1', + baseUrl: 'https://us.cloud.langfuse.com', + }, + } as AppConfig, + }), + ).toEqual({ + deterministicTraceId: true, + metadata: { 'librechat.tenant.id': 'tenant-1' }, + enabled: false, + librechatTraceAttributes: { + [CENTRAL_EXPORT_ATTRIBUTE]: 'false', + }, + tags: ['tenant:tenant-1'], + }); + }); + + it('honors tenant Langfuse enabled=false before adding routing attributes', () => { + expect( + buildLangfuseConfig({ + tenantId: 'tenant-1', + centralTraceExportEnabled: false, + appConfig: { + langfuse: { + enabled: false, + }, + } as AppConfig, + }), + ).toEqual({ + deterministicTraceId: true, + metadata: { 'librechat.tenant.id': 'tenant-1' }, + enabled: false, + tags: ['tenant:tenant-1'], + }); + }); +}); diff --git a/packages/api/src/langfuse/config.ts b/packages/api/src/langfuse/config.ts index c45a77f0dd..53e5472c02 100644 --- a/packages/api/src/langfuse/config.ts +++ b/packages/api/src/langfuse/config.ts @@ -12,8 +12,22 @@ export type LangfuseFanoutConfig = LangfuseAppConfig['fanout'] & { type LangfuseRunConfigWithTraceAttributes = LangfuseRunConfig & { librechatTraceAttributes?: Record; }; +type LangfuseTenantDestination = NonNullable>; +type LangfuseExportPlan = + | { type: 'directCentral' } + | { type: 'disabled' } + | { type: 'fanoutCollector'; collectorUrl: string } + | { + type: 'tenantFanout'; + collectorUrl: string; + destination: LangfuseTenantDestination; + publicKey: string; + secretKey: string; + }; 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 CENTRAL_MEDIA_DISABLED_SEGMENT = 'central-media-disabled'; const DEFAULT_BASE_URL = 'https://cloud.langfuse.com'; function appendPath(baseUrl: string, path: string): string { @@ -63,12 +77,69 @@ function applyCentralEnvConfig(langfuse: LangfuseRunConfigWithTraceAttributes): } } +function disableCentralExport(langfuse: LangfuseRunConfigWithTraceAttributes): void { + langfuse.librechatTraceAttributes = { + ...(langfuse.librechatTraceAttributes ?? {}), + [CENTRAL_EXPORT_ATTRIBUTE]: 'false', + }; +} + +function resolveLangfuseExportPlan({ + centralTraceExportEnabled, + fanoutEnabled, + fanoutCollectorUrl, + tenantExportEnabled, + publicKey, + secretKey, + tenantDestination, +}: { + centralTraceExportEnabled: boolean; + fanoutEnabled: boolean; + fanoutCollectorUrl?: string; + tenantExportEnabled: boolean; + publicKey?: string; + secretKey?: string; + tenantDestination?: LangfuseTenantDestination; +}): LangfuseExportPlan { + if (!fanoutEnabled || fanoutCollectorUrl == null) { + return centralTraceExportEnabled ? { type: 'directCentral' } : { type: 'disabled' }; + } + + const canRouteTenantFanout = + tenantExportEnabled && publicKey != null && secretKey != null && tenantDestination != null; + + if (canRouteTenantFanout) { + return { + type: 'tenantFanout', + collectorUrl: fanoutCollectorUrl, + destination: tenantDestination, + publicKey, + secretKey, + }; + } + + // 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: 'disabled' }; +} + export function buildLangfuseConfig({ appConfig, tenantId, + centralTraceExportEnabled = true, }: { appConfig?: AppConfig; tenantId?: string; + /** + * Defaults to true. Set false to suppress central Langfuse export for this + * run. Fanout deployments stamp a routing attribute that the collector uses + * to drop the central pipeline while preserving tenant fanout when available. + */ + centralTraceExportEnabled?: boolean; } = {}): LangfuseRunConfig { const normalizedTenantId = normalizeString(tenantId); const config = appConfig?.langfuse; @@ -91,6 +162,9 @@ export function buildLangfuseConfig({ enabled: false, }; } + if (!centralTraceExportEnabled) { + disableCentralExport(langfuse); + } const publicKey = normalizeString(config?.publicKey); const secretKey = normalizeString(config?.secretKey); @@ -100,35 +174,49 @@ export function buildLangfuseConfig({ const fanoutCollectorUrl = normalizeString(fanout?.collectorUrl) ?? normalizeString(process.env.LANGFUSE_FANOUT_COLLECTOR_URL); + const tenantExportEmergencyEnabled = isLangfuseTenantExportEnabled(); const tenantDestination = resolveLangfuseTenantDestination(config?.baseUrl); - const tenantExportDestination = hasTenantCredentials ? tenantDestination : undefined; - const tenantExportCollectorUrl = fanoutCollectorUrl; - const tenantExportEnabled = - hasTenantCredentials && - fanoutEnabled && - isLangfuseTenantExportEnabled() && - tenantExportDestination != null && - tenantExportCollectorUrl != null; + const exportPlan = resolveLangfuseExportPlan({ + centralTraceExportEnabled, + fanoutEnabled, + fanoutCollectorUrl, + tenantExportEnabled: hasTenantCredentials && tenantExportEmergencyEnabled, + publicKey, + secretKey, + tenantDestination, + }); - if (tenantExportEnabled && tenantExportDestination && tenantExportCollectorUrl) { - langfuse.publicKey = publicKey; - langfuse.secretKey = secretKey; - langfuse.baseUrl = appendPath( - tenantExportCollectorUrl, - `/tenant/${tenantExportDestination.key}`, - ); - // TODO: Add support in @librechat/agents for Langfuse additionalHeaders and - // route by headers if we need multiple tenant Langfuse exports for one run. - // The destination-scoped URL is the current app-to-gateway routing contract. - langfuse.librechatTraceAttributes = { - ...(langfuse.librechatTraceAttributes ?? {}), - [TENANT_EXPORT_ATTRIBUTE]: 'true', - [TENANT_DESTINATION_ATTRIBUTE]: tenantExportDestination.key, - }; - } else if (fanoutEnabled && fanoutCollectorUrl) { - langfuse.baseUrl = fanoutCollectorUrl; - } else { - applyCentralEnvConfig(langfuse); + switch (exportPlan.type) { + case 'tenantFanout': + langfuse.publicKey = exportPlan.publicKey; + langfuse.secretKey = exportPlan.secretKey; + langfuse.baseUrl = appendPath( + exportPlan.collectorUrl, + [ + '', + 'tenant', + exportPlan.destination.key, + ...(!centralTraceExportEnabled ? [CENTRAL_MEDIA_DISABLED_SEGMENT] : []), + ].join('/'), + ); + // TODO: Add support in @librechat/agents for Langfuse additionalHeaders and + // route by headers if we need multiple tenant Langfuse exports for one run. + // The destination-scoped URL is the current app-to-gateway routing contract. + langfuse.librechatTraceAttributes = { + ...(langfuse.librechatTraceAttributes ?? {}), + [TENANT_EXPORT_ATTRIBUTE]: 'true', + [TENANT_DESTINATION_ATTRIBUTE]: exportPlan.destination.key, + }; + break; + case 'fanoutCollector': + langfuse.baseUrl = exportPlan.collectorUrl; + break; + case 'disabled': + langfuse.enabled = false; + break; + case 'directCentral': + applyCentralEnvConfig(langfuse); + break; } return langfuse;