fix: gate tenant Langfuse fanout

.env.example

@@ -131,6 +131,8 @@ NODE_MAX_OLD_SPACE_SIZE=6144
 # See otel/langfuse-fanout/README.md.
 # LANGFUSE_FANOUT_ENABLED=false
 # LANGFUSE_FANOUT_COLLECTOR_URL=http://langfuse-fanout-collector:4318
+# Emergency switch: set false to keep central collector export but skip tenant trace/score export.
+# LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=true
 # Langfuse Cloud base URL options: https://cloud.langfuse.com (EU),
 # https://us.cloud.langfuse.com (US), https://jp.cloud.langfuse.com (JP),
 # https://hipaa.cloud.langfuse.com (HIPAA).

deploy-compose.langfuse-fanout.yml

@@ -5,6 +5,7 @@ services:
     environment:
       - LANGFUSE_FANOUT_ENABLED=true
       - LANGFUSE_FANOUT_COLLECTOR_URL=http://langfuse-fanout-collector:4318
+      - LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=${LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED:-true}
     networks:
       - default
       - langfuse-fanout

docker-compose.langfuse-fanout.yml

@@ -5,6 +5,7 @@ services:
     environment:
       - LANGFUSE_FANOUT_ENABLED=true
       - LANGFUSE_FANOUT_COLLECTOR_URL=http://langfuse-fanout-collector:4318
+      - LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=${LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED:-true}
     networks:
       - default
       - langfuse-fanout

helm/librechat/readme.md

@@ -64,6 +64,10 @@ When enabled, the chart also sets `LANGFUSE_FANOUT_ENABLED` and
 `LANGFUSE_FANOUT_COLLECTOR_URL` for the LibreChat app unless those values are
 already provided in `librechat.configEnv`.
 
+Set `librechat.configEnv.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=false` to keep
+central trace export flowing through the collector while disabling tenant trace
+and score export.
+
 Langfuse base URLs are startup configuration for the LibreChat server and
 collector. Tenant API keys can still be added through tenant app configuration at
 runtime without restarting either component.

helm/librechat/templates/langfuse-fanout-configmap.yaml

@@ -27,6 +27,14 @@ data:
         check_interval: 1s
         limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_LIMIT_MIB}
         spike_limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB}
+      filter/tenant_export:
+        error_mode: ignore
+        traces:
+          span:
+            - attributes["langfuse.trace.metadata.librechat.langfuse.tenant_export.enabled"] != "true"
+      batch/central:
+        timeout: 1s
+        send_batch_size: 128
       batch/by_auth:
         timeout: 1s
         send_batch_size: 128
@@ -53,8 +61,12 @@ data:
     service:
       extensions: [headers_setter/tenant_passthrough]
       pipelines:
-        traces:
+        traces/central:
           receivers: [otlp]
-          processors: [memory_limiter, batch/by_auth]
-          exporters: [otlphttp/central, otlphttp/tenant]
+          processors: [memory_limiter, batch/central]
+          exporters: [otlphttp/central]
+        traces/tenant:
+          receivers: [otlp]
+          processors: [memory_limiter, filter/tenant_export, batch/by_auth]
+          exporters: [otlphttp/tenant]
 {{- end }}

otel/langfuse-fanout/README.md

@@ -15,6 +15,10 @@ disabled unless you explicitly deploy the fanout collector.
 - The collector also exports the same trace to the tenant Langfuse project by
   forwarding the tenant `Authorization` header that LibreChat attaches to the
   OTLP request.
+- Tenant export is conditional. LibreChat marks traces as tenant-exportable only
+  when tenant keys are configured and `LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED` is
+  not false; unmarked traces are still exported to central but are dropped by
+  the tenant pipeline.
 - User feedback scores use Langfuse's direct REST API. Scores are sent to the
   central project and, when tenant Langfuse keys are configured, the tenant
   project.
@@ -31,6 +35,9 @@ defined in this collector config.
   starts.
 - Tenant Langfuse API keys can be added, changed, or disabled in tenant app
   configuration at runtime without restarting LibreChat or the collector.
+- `LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=false` can be set on LibreChat as an
+  emergency switch to stop tenant trace and score export while keeping central
+  collector export active.
 - This supports Langfuse Cloud and self-hosted Langfuse as long as the relevant
   central and tenant base URLs are known at collector/server startup.
 - All tenant traces that use deployment-level fanout are sent to the same
@@ -46,6 +53,7 @@ Set the central Langfuse destination in `.env`:
 LANGFUSE_FANOUT_CENTRAL_BASE_URL=https://cloud.langfuse.com
 LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER=Basic <base64-public-key-colon-secret-key>
 LANGFUSE_FANOUT_TENANT_BASE_URL=https://cloud.langfuse.com
+LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED=true
 ```
 
 Langfuse Cloud base URL options:

otel/langfuse-fanout/otelcol.yaml

@@ -18,6 +18,14 @@ processors:
     check_interval: 1s
     limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_LIMIT_MIB}
     spike_limit_mib: ${env:LANGFUSE_FANOUT_MEMORY_SPIKE_LIMIT_MIB}
+  filter/tenant_export:
+    error_mode: ignore
+    traces:
+      span:
+        - attributes["langfuse.trace.metadata.librechat.langfuse.tenant_export.enabled"] != "true"
+  batch/central:
+    timeout: 1s
+    send_batch_size: 128
   batch/by_auth:
     timeout: 1s
     send_batch_size: 128
@@ -44,7 +52,11 @@ exporters:
 service:
   extensions: [headers_setter/tenant_passthrough]
   pipelines:
-    traces:
+    traces/central:
       receivers: [otlp]
-      processors: [memory_limiter, batch/by_auth]
-      exporters: [otlphttp/central, otlphttp/tenant]
+      processors: [memory_limiter, batch/central]
+      exporters: [otlphttp/central]
+    traces/tenant:
+      receivers: [otlp]
+      processors: [memory_limiter, filter/tenant_export, batch/by_auth]
+      exporters: [otlphttp/tenant]

packages/api/src/agents/__tests__/run-summarization.test.ts

@@ -205,6 +205,7 @@ beforeEach(() => {
   jest.clearAllMocks();
   delete process.env.LANGFUSE_FANOUT_ENABLED;
   delete process.env.LANGFUSE_FANOUT_COLLECTOR_URL;
+  delete process.env.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED;
 });
 
 // ---------------------------------------------------------------------------
@@ -1194,7 +1195,10 @@ describe('Langfuse run config', () => {
       publicKey: 'pk-tenant-1',
       secretKey: 'sk-tenant-1',
       baseUrl: 'http://langfuse-fanout-collector:4318',
-      metadata: { 'librechat.tenant.id': 'tenant-1' },
+      metadata: {
+        'librechat.tenant.id': 'tenant-1',
+        'librechat.langfuse.tenant_export.enabled': 'true',
+      },
       tags: ['tenant:tenant-1'],
     });
   });
@@ -1217,10 +1221,14 @@ describe('Langfuse run config', () => {
       publicKey: 'pk-tenant-1',
       secretKey: 'sk-tenant-1',
       baseUrl: 'http://collector-from-env:4318',
+      metadata: {
+        'librechat.tenant.id': 'tenant-1',
+        'librechat.langfuse.tenant_export.enabled': 'true',
+      },
     });
   });
 
-  it('does not route to fanout when deployment fanout is enabled without tenant keys', async () => {
+  it('routes central-only traces to the collector when deployment fanout is enabled without tenant keys', async () => {
     process.env.LANGFUSE_FANOUT_ENABLED = 'true';
     process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector-from-env:4318';
 
@@ -1233,6 +1241,30 @@ describe('Langfuse run config', () => {
 
     expect(callArgs.langfuse).toEqual({
       deterministicTraceId: true,
+      baseUrl: 'http://collector-from-env:4318',
+      metadata: { 'librechat.tenant.id': 'tenant-1' },
+      tags: ['tenant:tenant-1'],
+    });
+  });
+
+  it('routes central-only traces to the collector when tenant fanout export is disabled', async () => {
+    process.env.LANGFUSE_FANOUT_ENABLED = 'true';
+    process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector-from-env:4318';
+    process.env.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED = 'false';
+
+    const callArgs = await callAndCaptureRunConfig({
+      tenantId: 'tenant-1',
+      appConfig: {
+        langfuse: {
+          publicKey: 'pk-tenant-1',
+          secretKey: 'sk-tenant-1',
+        },
+      } as AppConfig,
+    });
+
+    expect(callArgs.langfuse).toEqual({
+      deterministicTraceId: true,
+      baseUrl: 'http://collector-from-env:4318',
       metadata: { 'librechat.tenant.id': 'tenant-1' },
       tags: ['tenant:tenant-1'],
     });

packages/api/src/langfuse/config.ts

@@ -1,8 +1,13 @@
 import type { AppConfig } from '@librechat/data-schemas';
 import type { RunConfig } from '@librechat/agents';
-import { normalizeString } from './utils';
+import { isFalseEnv, normalizeString } from './utils';
 
 type LangfuseRunConfig = NonNullable<RunConfig['langfuse']>;
+const TENANT_EXPORT_METADATA_KEY = 'librechat.langfuse.tenant_export.enabled';
+
+function isTenantExportEnabled(): boolean {
+  return !isFalseEnv(process.env.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED);
+}
 
 function mergeTraceMetadata(
   base: LangfuseRunConfig['metadata'],
@@ -56,23 +61,29 @@ export function buildLangfuseConfig({
   const publicKey = normalizeString(config?.publicKey);
   const secretKey = normalizeString(config?.secretKey);
   const hasTenantCredentials = Boolean(publicKey && secretKey);
-  if (hasTenantCredentials) {
+  const fanout = config?.fanout;
+  const fanoutEnabled =
+    fanout?.enabled !== false &&
+    (fanout?.enabled === true || process.env.LANGFUSE_FANOUT_ENABLED === 'true');
+  const fanoutCollectorUrl =
+    normalizeString(fanout?.collectorUrl) ?? normalizeString(process.env.LANGFUSE_FANOUT_COLLECTOR_URL);
+  const tenantExportEnabled = hasTenantCredentials && fanoutEnabled && isTenantExportEnabled();
+
+  if (hasTenantCredentials && (!fanoutEnabled || tenantExportEnabled)) {
     langfuse.publicKey = publicKey;
     langfuse.secretKey = secretKey;
+  }
 
-    const fanout = config?.fanout;
-    const fanoutEnabled =
-      fanout?.enabled !== false &&
-      (fanout?.enabled === true || process.env.LANGFUSE_FANOUT_ENABLED === 'true');
-    const fanoutCollectorUrl =
-      normalizeString(fanout?.collectorUrl) ??
-      normalizeString(process.env.LANGFUSE_FANOUT_COLLECTOR_URL);
-    const baseUrl =
-      fanoutEnabled && fanoutCollectorUrl ? fanoutCollectorUrl : normalizeString(config?.baseUrl);
+  const baseUrl = fanoutEnabled && fanoutCollectorUrl ? fanoutCollectorUrl : normalizeString(config?.baseUrl);
+  if (baseUrl) {
+    langfuse.baseUrl = baseUrl;
+  }
 
-    if (baseUrl) {
-      langfuse.baseUrl = baseUrl;
-    }
+  if (tenantExportEnabled) {
+    langfuse.metadata = {
+      ...(langfuse.metadata ?? {}),
+      [TENANT_EXPORT_METADATA_KEY]: 'true',
+    };
   }
 
   return langfuse;

packages/api/src/langfuse/destinations.ts

@@ -1,5 +1,5 @@
 import type { AppConfig } from '@librechat/data-schemas';
-import { normalizeString, toBasicAuthorization } from './utils';
+import { isFalseEnv, normalizeString, toBasicAuthorization } from './utils';
 
 const DEFAULT_BASE_URL = 'https://cloud.langfuse.com';
 
@@ -9,10 +9,6 @@ export type LangfuseScoreDestination = {
   authorization: string;
 };
 
-function isFalseEnv(value?: string): boolean {
-  return value != null && ['0', 'false', 'no', 'off'].includes(value.trim().toLowerCase());
-}
-
 function isSampleRateEnabled(value?: string): boolean {
   if (value == null || value.trim() === '') {
     return true;
@@ -68,6 +64,9 @@ function getTenantScoreDestination(appConfig?: AppConfig): LangfuseScoreDestinat
   if (!isTracingEnabled()) {
     return undefined;
   }
+  if (isFalseEnv(process.env.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED)) {
+    return undefined;
+  }
 
   const config = appConfig?.langfuse;
   if (config?.enabled === false) {

packages/api/src/langfuse/feedback.spec.ts

@@ -23,6 +23,7 @@ const langfuseEnvKeys = [
   'LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER',
   'LANGFUSE_FANOUT_CENTRAL_BASE_URL',
   'LANGFUSE_FANOUT_TENANT_BASE_URL',
+  'LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED',
 ];
 let fetchMock: jest.SpiedFunction<typeof fetch>;
 
@@ -278,6 +279,32 @@ describe('Langfuse feedback scores', () => {
     );
   });
 
+  it('skips tenant scores when tenant fanout export is disabled but keeps central scores', async () => {
+    process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = 'Basic central-auth';
+    process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://central-langfuse:3000';
+    process.env.LANGFUSE_FANOUT_TENANT_BASE_URL = 'http://tenant-langfuse:3000';
+    process.env.LANGFUSE_FANOUT_TENANT_EXPORT_ENABLED = 'false';
+    const { sendFeedbackScore } = await loadFeedback();
+
+    await sendFeedbackScore({
+      traceId: 'trace-id',
+      feedback: { rating: 'thumbsUp' },
+      appConfig: appConfigWithLangfuse({
+        publicKey: 'tenant-public-key',
+        secretKey: 'tenant-secret-key',
+      }),
+    });
+
+    expect(getFetchMock()).toHaveBeenCalledTimes(1);
+    expect(getFetchMock()).toHaveBeenCalledWith(
+      'http://central-langfuse:3000/api/public/scores',
+      expect.objectContaining({
+        method: 'POST',
+        headers: expect.objectContaining({ Authorization: 'Basic central-auth' }),
+      }),
+    );
+  });
+
   it('deduplicates matching central and tenant score destinations', async () => {
     process.env.LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER = getTenantAuthorization();
     process.env.LANGFUSE_FANOUT_CENTRAL_BASE_URL = 'http://shared-langfuse:3000';

packages/api/src/langfuse/utils.ts

@@ -2,6 +2,10 @@ export function normalizeString(value: unknown): string | undefined {
   return typeof value === 'string' && value.trim() !== '' ? value.trim() : undefined;
 }
 
+export function isFalseEnv(value?: string): boolean {
+  return value != null && ['0', 'false', 'no', 'off'].includes(value.trim().toLowerCase());
+}
+
 export function toBasicAuthorization(publicKey: string, secretKey: string): string {
   return `Basic ${Buffer.from(`${publicKey}:${secretKey}`).toString('base64')}`;
 }