From 4cb7b09fb7e4edec1aa23bd99cbd5258defe6fb9 Mon Sep 17 00:00:00 2001 From: Ravi Kumar L Date: Mon, 27 Jul 2026 12:14:52 +0200 Subject: [PATCH] feat(langfuse): support single-tenant connection settings --- .env.example | 15 ++- api/server/routes/__tests__/config.spec.js | 54 ++++++++- api/server/routes/config.js | 4 +- .../Nav/Settings/__tests__/Sidebar.spec.tsx | 5 +- .../Nav/Settings/__tests__/registry.spec.ts | 10 +- .../src/components/Nav/Settings/context.tsx | 3 - .../src/components/Nav/Settings/registry.tsx | 2 +- client/src/components/Nav/Settings/types.ts | 3 +- .../api/src/admin/langfuse.handler.spec.ts | 52 +++++++- packages/api/src/admin/langfuse.ts | 14 +-- .../__tests__/run-summarization.test.ts | 7 ++ packages/api/src/agents/run.ts | 1 + packages/api/src/langfuse/config.spec.ts | 113 ++++++++++++++++++ packages/api/src/langfuse/config.ts | 48 ++++++-- packages/api/src/langfuse/destinations.ts | 82 ++++++++----- packages/api/src/langfuse/feedback.spec.ts | 83 +++++++++++++ packages/api/src/langfuse/feedback.ts | 2 +- packages/api/src/langfuse/index.ts | 2 +- packages/api/src/langfuse/policy.spec.ts | 103 ++++++++++++++++ packages/api/src/langfuse/policy.ts | 81 +++++++++++++ 20 files changed, 604 insertions(+), 80 deletions(-) create mode 100644 packages/api/src/langfuse/policy.spec.ts create mode 100644 packages/api/src/langfuse/policy.ts diff --git a/.env.example b/.env.example index 44821dc0fe..50acc31fee 100644 --- a/.env.example +++ b/.env.example @@ -131,14 +131,21 @@ NODE_MAX_OLD_SPACE_SIZE=6144 # LANGFUSE_PUBLIC_KEY= # LANGFUSE_SECRET_KEY= # LANGFUSE_BASE_URL= +# Set false to disable Langfuse traces and feedback scores. +# LANGFUSE_TRACING_ENABLED=true +# Trace-level sample rate from 0 to 1. Sampled-out traces do not receive scores. +# LANGFUSE_SAMPLE_RATE=1 + +# In single-tenant deployments without environment credentials, an admin can +# configure one encrypted Langfuse connection in the application settings. +# Complete environment credentials take precedence and hide those settings. # Optional Langfuse fanout for tenant-scoped Langfuse projects. # The fanout gateway is opt-in: add docker-compose.langfuse-fanout.yml, # deploy-compose.langfuse-fanout.yml, or enable helm langfuseFanout. -# Tenant public/secret keys are read from LibreChat tenant app configuration. -# Tenant Langfuse base URLs must be set in tenant app configuration and match -# one of the known startup destinations. Tenant API keys can be added or changed -# at runtime through tenant app configuration. +# Tenant public/secret keys and a destination key are read from LibreChat tenant +# app configuration. Destination keys resolve against known startup URLs. Tenant +# API keys can be added or changed at runtime through tenant app configuration. # See otel/langfuse-fanout/README.md. # LANGFUSE_FANOUT_ENABLED=false # LANGFUSE_FANOUT_COLLECTOR_URL=http://langfuse-fanout-collector:4318 diff --git a/api/server/routes/__tests__/config.spec.js b/api/server/routes/__tests__/config.spec.js index 871c690dde..1176e8d25a 100644 --- a/api/server/routes/__tests__/config.spec.js +++ b/api/server/routes/__tests__/config.spec.js @@ -108,6 +108,11 @@ afterEach(() => { delete process.env.HELP_AND_FAQ_URL; delete process.env.LANGFUSE_FANOUT_ENABLED; delete process.env.LANGFUSE_FANOUT_COLLECTOR_URL; + delete process.env.LANGFUSE_PUBLIC_KEY; + delete process.env.LANGFUSE_SECRET_KEY; + delete process.env.LANGFUSE_TRACING_ENABLED; + delete process.env.LANGFUSE_SAMPLE_RATE; + delete process.env.TENANT_ISOLATION_STRICT; }); describe('GET /api/config', () => { @@ -398,12 +403,12 @@ describe('GET /api/config', () => { let response = await request(app).get('/api/config'); expect(response.body.langfuseFanoutEnabled).toBe(false); - expect(response.body.langfuseConnectionAccess).toBe(false); + expect(response.body.langfuseConnectionAccess).toBe(true); process.env.LANGFUSE_FANOUT_COLLECTOR_URL = ' '; response = await request(app).get('/api/config'); expect(response.body.langfuseFanoutEnabled).toBe(false); - expect(response.body.langfuseConnectionAccess).toBe(false); + expect(response.body.langfuseConnectionAccess).toBe(true); process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://langfuse-fanout:4318'; response = await request(app).get('/api/config'); @@ -413,6 +418,7 @@ describe('GET /api/config', () => { it('advertises Langfuse connection access from capabilities rather than the user role', async () => { mockGetAppConfig.mockResolvedValue(baseAppConfig); + process.env.TENANT_ISOLATION_STRICT = 'true'; process.env.LANGFUSE_FANOUT_ENABLED = 'true'; process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://langfuse-fanout:4318'; const app = createApp({ ...mockUser, role: 'DELEGATED_ADMIN' }); @@ -430,6 +436,49 @@ describe('GET /api/config', () => { expect(response.body.langfuseConnectionAccess).toBe(false); }); + it('advertises Langfuse connection access by default in single-tenant mode', async () => { + mockGetAppConfig.mockResolvedValue(baseAppConfig); + mockHasCapability.mockResolvedValue(true); + mockHasConfigCapability.mockResolvedValue(true); + const app = createApp(mockUser); + + const response = await request(app).get('/api/config'); + + expect(response.body.langfuseFanoutEnabled).toBe(false); + expect(response.body.langfuseConnectionAccess).toBe(true); + }); + + it('hides single-tenant connection settings when environment credentials are configured', async () => { + mockGetAppConfig.mockResolvedValue(baseAppConfig); + mockHasCapability.mockResolvedValue(true); + mockHasConfigCapability.mockResolvedValue(true); + process.env.LANGFUSE_PUBLIC_KEY = 'pk-env'; + process.env.LANGFUSE_SECRET_KEY = 'sk-env'; + const app = createApp(mockUser); + + const response = await request(app).get('/api/config'); + + expect(response.body.langfuseConnectionAccess).toBe(false); + expect(mockHasCapability).not.toHaveBeenCalled(); + expect(mockHasConfigCapability).not.toHaveBeenCalled(); + }); + + it.each([ + ['LANGFUSE_TRACING_ENABLED', 'false'], + ['LANGFUSE_SAMPLE_RATE', '0'], + ])('hides Langfuse connection settings when %s=%s', async (key, value) => { + mockGetAppConfig.mockResolvedValue(baseAppConfig); + mockHasCapability.mockResolvedValue(true); + mockHasConfigCapability.mockResolvedValue(true); + process.env[key] = value; + const app = createApp(mockUser); + + const response = await request(app).get('/api/config'); + + expect(response.body.langfuseConnectionAccess).toBe(false); + expect(mockHasCapability).not.toHaveBeenCalled(); + }); + it('should include post-login informational fields', async () => { process.env.ANALYTICS_GTM_ID = 'GTM-XYZ'; process.env.CUSTOM_FOOTER = 'authenticated footer text'; @@ -540,6 +589,7 @@ describe('GET /api/config', () => { it('should not call hasCapability when allowAccountDeletion is already true', async () => { mockGetAppConfig.mockResolvedValue(baseAppConfig); + process.env.LANGFUSE_TRACING_ENABLED = 'false'; const app = createApp(mockUser); const response = await request(app).get('/api/config'); diff --git a/api/server/routes/config.js b/api/server/routes/config.js index 818b304505..35f28e9d76 100644 --- a/api/server/routes/config.js +++ b/api/server/routes/config.js @@ -1,6 +1,7 @@ const express = require('express'); const { isEnabled, + isLangfuseConnectionAvailable, isLangfuseFanoutEnabled, getBalanceConfig, getCloudFrontConfig, @@ -251,9 +252,10 @@ router.get('/', async function (req, res) { const balanceConfig = getBalanceConfig(appConfig); const cloudFront = buildCloudFrontStartupConfig(); const langfuseFanoutEnabled = isLangfuseFanoutEnabled(); + const langfuseConnectionAvailable = isLangfuseConnectionAvailable(); let langfuseConnectionAccess = false; - if (langfuseFanoutEnabled) { + if (langfuseConnectionAvailable) { try { const userId = req.user.id ?? req.user._id?.toString(); if (userId) { diff --git a/client/src/components/Nav/Settings/__tests__/Sidebar.spec.tsx b/client/src/components/Nav/Settings/__tests__/Sidebar.spec.tsx index b794158d73..7cf5d0b307 100644 --- a/client/src/components/Nav/Settings/__tests__/Sidebar.spec.tsx +++ b/client/src/components/Nav/Settings/__tests__/Sidebar.spec.tsx @@ -18,7 +18,6 @@ const ctx: SettingsContextValue = { allowAccountDeletion: true, aboutEnabled: false, engineTTS: 'browser', - langfuseFanoutEnabled: false, langfuseConnectionAccess: false, }; @@ -49,12 +48,12 @@ describe('Sidebar', () => { }); it('shows the Langfuse tab when Langfuse is available to the user', () => { - setup({ langfuseFanoutEnabled: true, langfuseConnectionAccess: true }); + setup({ langfuseConnectionAccess: true }); expect(screen.getByText('Langfuse')).toBeInTheDocument(); }); it('hides the Langfuse tab without Langfuse connection access', () => { - setup({ langfuseFanoutEnabled: true, langfuseConnectionAccess: false }); + setup({ langfuseConnectionAccess: false }); expect(screen.queryByText('Langfuse')).not.toBeInTheDocument(); }); diff --git a/client/src/components/Nav/Settings/__tests__/registry.spec.ts b/client/src/components/Nav/Settings/__tests__/registry.spec.ts index 6c9f2c6a47..c137b9cd48 100644 --- a/client/src/components/Nav/Settings/__tests__/registry.spec.ts +++ b/client/src/components/Nav/Settings/__tests__/registry.spec.ts @@ -20,7 +20,6 @@ const settingsContext: SettingsContextValue = { allowAccountDeletion: true, aboutEnabled: false, engineTTS: 'browser', - langfuseFanoutEnabled: false, langfuseConnectionAccess: false, }; @@ -60,11 +59,10 @@ describe('settings registry', () => { }); }); - it('shows the connection when fanout is enabled and the user can manage it', () => { + it('shows the connection when the user can manage it', () => { expect( langfuseEntry?.show?.({ ...settingsContext, - langfuseFanoutEnabled: true, langfuseConnectionAccess: true, }), ).toBe(true); @@ -74,20 +72,18 @@ describe('settings registry', () => { expect( langfuseEntry?.show?.({ ...settingsContext, - langfuseFanoutEnabled: true, langfuseConnectionAccess: false, }), ).toBe(false); }); - it('hides the connection when fanout is disabled', () => { + it('shows the connection in single-tenant mode without fanout', () => { expect( langfuseEntry?.show?.({ ...settingsContext, - langfuseFanoutEnabled: false, langfuseConnectionAccess: true, }), - ).toBe(false); + ).toBe(true); }); }); }); diff --git a/client/src/components/Nav/Settings/context.tsx b/client/src/components/Nav/Settings/context.tsx index b7b19a2e23..176b8b584b 100644 --- a/client/src/components/Nav/Settings/context.tsx +++ b/client/src/components/Nav/Settings/context.tsx @@ -27,7 +27,6 @@ export function useSettingsContext(): SettingsContextValue { }); const balanceEnabled = startupConfig?.balance?.enabled === true; - const langfuseFanoutEnabled = startupConfig?.langfuseFanoutEnabled === true; const langfuseConnectionAccess = startupConfig?.langfuseConnectionAccess === true; const isLocalProvider = user?.provider === 'local'; const twoFactorEnabled = user?.twoFactorEnabled === true; @@ -53,7 +52,6 @@ export function useSettingsContext(): SettingsContextValue { allowAccountDeletion, aboutEnabled, engineTTS, - langfuseFanoutEnabled, langfuseConnectionAccess, }), [ @@ -69,7 +67,6 @@ export function useSettingsContext(): SettingsContextValue { allowAccountDeletion, aboutEnabled, engineTTS, - langfuseFanoutEnabled, langfuseConnectionAccess, ], ); diff --git a/client/src/components/Nav/Settings/registry.tsx b/client/src/components/Nav/Settings/registry.tsx index 76a51f3499..3b66289a05 100644 --- a/client/src/components/Nav/Settings/registry.tsx +++ b/client/src/components/Nav/Settings/registry.tsx @@ -508,7 +508,7 @@ export const registry: SettingEntry[] = [ section: 'langfuse', labelKey: 'com_ui_langfuse_title', keywords: ['langfuse', 'observability', 'tracing', 'telemetry', 'traces'], - show: (ctx) => ctx.langfuseConnectionAccess && ctx.langfuseFanoutEnabled, + show: (ctx) => ctx.langfuseConnectionAccess, Component: LangfuseConnection, }, // Data controls ยท Danger zone diff --git a/client/src/components/Nav/Settings/types.ts b/client/src/components/Nav/Settings/types.ts index 1dbfd92885..e54dcedd07 100644 --- a/client/src/components/Nav/Settings/types.ts +++ b/client/src/components/Nav/Settings/types.ts @@ -48,7 +48,6 @@ export interface SettingsContextValue { allowAccountDeletion: boolean; aboutEnabled: boolean; engineTTS: string; - langfuseFanoutEnabled: boolean; langfuseConnectionAccess: boolean; } @@ -131,7 +130,7 @@ export const TABS: TabMeta[] = [ icon: createLangfuseIcon('h-3.5 w-3.5'), }, ], - show: (ctx) => ctx.langfuseConnectionAccess && ctx.langfuseFanoutEnabled, + show: (ctx) => ctx.langfuseConnectionAccess, }, { id: SettingsTabValues.DATA, diff --git a/packages/api/src/admin/langfuse.handler.spec.ts b/packages/api/src/admin/langfuse.handler.spec.ts index c04464fdb1..48dd348a1d 100644 --- a/packages/api/src/admin/langfuse.handler.spec.ts +++ b/packages/api/src/admin/langfuse.handler.spec.ts @@ -16,6 +16,7 @@ beforeAll(async () => { }); beforeEach(() => { + process.env.TENANT_ISOLATION_STRICT = 'true'; process.env.LANGFUSE_FANOUT_ENABLED = 'true'; process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://langfuse-fanout:4318'; global.fetch = jest.fn().mockResolvedValue({ ok: true, status: 200 }) as unknown as typeof fetch; @@ -24,6 +25,11 @@ beforeEach(() => { afterEach(() => { delete process.env.LANGFUSE_FANOUT_ENABLED; delete process.env.LANGFUSE_FANOUT_COLLECTOR_URL; + delete process.env.LANGFUSE_PUBLIC_KEY; + delete process.env.LANGFUSE_SECRET_KEY; + delete process.env.LANGFUSE_TRACING_ENABLED; + delete process.env.LANGFUSE_SAMPLE_RATE; + delete process.env.TENANT_ISOLATION_STRICT; global.fetch = realFetch; }); @@ -103,7 +109,7 @@ function rehydrate(fields: Record): Record { } describe('createAdminLangfuseHandlers', () => { - describe('fanout feature gate', () => { + describe('connection availability gate', () => { it('rejects connection reads when deployment fanout is disabled', async () => { delete process.env.LANGFUSE_FANOUT_ENABLED; const { handlers, deps } = createHandlers(); @@ -112,7 +118,7 @@ describe('createAdminLangfuseHandlers', () => { await handlers.getConnection(mockReq(), res); expect(res.statusCode).toBe(404); - expect(res.body).toEqual({ error: 'Langfuse fanout is not enabled' }); + expect(res.body).toEqual({ error: 'Langfuse connection settings are not available' }); expect(deps.findConfigByPrincipal).not.toHaveBeenCalled(); }); @@ -127,7 +133,7 @@ describe('createAdminLangfuseHandlers', () => { ); expect(res.statusCode).toBe(404); - expect(res.body).toEqual({ error: 'Langfuse fanout is not enabled' }); + expect(res.body).toEqual({ error: 'Langfuse connection settings are not available' }); expect(deps.patchConfigFields).not.toHaveBeenCalled(); }); @@ -139,7 +145,7 @@ describe('createAdminLangfuseHandlers', () => { await handlers.getConnection(mockReq(), res); expect(res.statusCode).toBe(404); - expect(res.body).toEqual({ error: 'Langfuse fanout is not enabled' }); + expect(res.body).toEqual({ error: 'Langfuse connection settings are not available' }); expect(deps.findConfigByPrincipal).not.toHaveBeenCalled(); }); @@ -155,10 +161,46 @@ describe('createAdminLangfuseHandlers', () => { ); expect(res.statusCode).toBe(404); - expect(res.body).toEqual({ error: 'Langfuse fanout is not enabled' }); + expect(res.body).toEqual({ error: 'Langfuse connection settings are not available' }); expect(deps.findConfigByPrincipal).not.toHaveBeenCalled(); expect(global.fetch).not.toHaveBeenCalled(); }); + + it('allows connection settings without fanout in single-tenant mode', async () => { + delete process.env.TENANT_ISOLATION_STRICT; + delete process.env.LANGFUSE_FANOUT_ENABLED; + delete process.env.LANGFUSE_FANOUT_COLLECTOR_URL; + const { handlers } = createHandlers(); + const res = mockRes(); + + await handlers.getConnection(mockReq(), res); + + expect(res.statusCode).toBe(200); + }); + + it('rejects single-tenant settings when environment credentials are configured', async () => { + delete process.env.TENANT_ISOLATION_STRICT; + process.env.LANGFUSE_PUBLIC_KEY = 'pk-env'; + process.env.LANGFUSE_SECRET_KEY = 'sk-env'; + const { handlers, deps } = createHandlers(); + const res = mockRes(); + + await handlers.getConnection(mockReq(), res); + + expect(res.statusCode).toBe(404); + expect(deps.findConfigByPrincipal).not.toHaveBeenCalled(); + }); + + it('rejects settings when tracing is disabled', async () => { + process.env.LANGFUSE_TRACING_ENABLED = 'false'; + const { handlers, deps } = createHandlers(); + const res = mockRes(); + + await handlers.getConnection(mockReq(), res); + + expect(res.statusCode).toBe(404); + expect(deps.findConfigByPrincipal).not.toHaveBeenCalled(); + }); }); describe('getConnection', () => { diff --git a/packages/api/src/admin/langfuse.ts b/packages/api/src/admin/langfuse.ts index 7ccbe8b24a..27a0f6653a 100644 --- a/packages/api/src/admin/langfuse.ts +++ b/packages/api/src/admin/langfuse.ts @@ -18,7 +18,7 @@ import { resolveLangfuseTenantDestination, } from '~/langfuse/tenantDestinations'; import { decryptConfigSecret, encryptConfigSecretFields } from './secrets'; -import { isLangfuseFanoutEnabled } from '~/langfuse/config'; +import { isLangfuseConnectionAvailable } from '~/langfuse/policy'; const DEFAULT_PRIORITY = 10; const ENCRYPTED_PREFIX = 'v3:'; @@ -71,12 +71,12 @@ function buildStatus(config: IConfig | null): TLangfuseConnectionStatus { }; } -function rejectWhenFanoutDisabled(res: Response): Response | undefined { - if (isLangfuseFanoutEnabled()) { +function rejectWhenConnectionUnavailable(res: Response): Response | undefined { + if (isLangfuseConnectionAvailable()) { return undefined; } - return res.status(404).json({ error: 'Langfuse fanout is not enabled' }); + return res.status(404).json({ error: 'Langfuse connection settings are not available' }); } function getLangfuseTestFailureMessage(status: number): string { @@ -170,7 +170,7 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): { } async function getConnection(req: ServerRequest, res: Response): Promise { - const disabledResponse = rejectWhenFanoutDisabled(res); + const disabledResponse = rejectWhenConnectionUnavailable(res); if (disabledResponse) { return disabledResponse; } @@ -185,7 +185,7 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): { } async function updateConnection(req: ServerRequest, res: Response): Promise { - const disabledResponse = rejectWhenFanoutDisabled(res); + const disabledResponse = rejectWhenConnectionUnavailable(res); if (disabledResponse) { return disabledResponse; } @@ -273,7 +273,7 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): { } async function testConnection(req: ServerRequest, res: Response): Promise { - const disabledResponse = rejectWhenFanoutDisabled(res); + const disabledResponse = rejectWhenConnectionUnavailable(res); if (disabledResponse) { return disabledResponse; } diff --git a/packages/api/src/agents/__tests__/run-summarization.test.ts b/packages/api/src/agents/__tests__/run-summarization.test.ts index 46c6c77f0f..24c50b9f08 100644 --- a/packages/api/src/agents/__tests__/run-summarization.test.ts +++ b/packages/api/src/agents/__tests__/run-summarization.test.ts @@ -224,6 +224,13 @@ beforeEach(() => { delete process.env.LANGFUSE_FANOUT_COLLECTOR_URL; delete process.env.LANGFUSE_FANOUT_TENANT_DESTINATIONS; delete process.env.LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED; + delete process.env.LANGFUSE_TRACING_ENABLED; + delete process.env.LANGFUSE_SAMPLE_RATE; + process.env.TENANT_ISOLATION_STRICT = 'true'; +}); + +afterAll(() => { + delete process.env.TENANT_ISOLATION_STRICT; }); // --------------------------------------------------------------------------- diff --git a/packages/api/src/agents/run.ts b/packages/api/src/agents/run.ts index f063fecdf0..f7c8dd0b98 100644 --- a/packages/api/src/agents/run.ts +++ b/packages/api/src/agents/run.ts @@ -1518,6 +1518,7 @@ export async function createRun({ // tracing is enabled. Requires @librechat/agents >= 3.2.21. langfuse: buildLangfuseConfig({ appConfig, + runId, tenantId: tenantId ?? user?.tenantId, centralTraceExportEnabled, }), diff --git a/packages/api/src/langfuse/config.spec.ts b/packages/api/src/langfuse/config.spec.ts index 6b18a46470..44d4a5f097 100644 --- a/packages/api/src/langfuse/config.spec.ts +++ b/packages/api/src/langfuse/config.spec.ts @@ -14,6 +14,9 @@ const envKeys = [ 'LANGFUSE_FANOUT_COLLECTOR_URL', 'LANGFUSE_FANOUT_TENANT_DESTINATIONS', 'LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED', + 'LANGFUSE_TRACING_ENABLED', + 'LANGFUSE_SAMPLE_RATE', + 'TENANT_ISOLATION_STRICT', ]; function clearEnv() { @@ -25,6 +28,7 @@ function clearEnv() { describe('buildLangfuseConfig', () => { beforeEach(() => { clearEnv(); + process.env.TENANT_ISOLATION_STRICT = 'true'; }); afterEach(() => { @@ -44,7 +48,116 @@ describe('buildLangfuseConfig', () => { expect(isLangfuseFanoutEnabled()).toBe(true); }); + it('uses a stored connection directly for every run in single-tenant mode', async () => { + delete process.env.TENANT_ISOLATION_STRICT; + const { encryptV3 } = await import('@librechat/data-schemas'); + const { buildLangfuseConfig } = await import('./config'); + + expect( + buildLangfuseConfig({ + runId: 'run-1', + appConfig: { + langfuse: { + enabled: true, + publicKey: 'pk-stored', + secretKey: encryptV3('sk-stored'), + destination: 'us', + }, + } as unknown as AppConfig, + }), + ).toEqual({ + deterministicTraceId: true, + publicKey: 'pk-stored', + secretKey: 'sk-stored', + baseUrl: 'https://us.cloud.langfuse.com', + }); + }); + + it('prefers environment credentials in single-tenant mode', async () => { + delete process.env.TENANT_ISOLATION_STRICT; + process.env.LANGFUSE_PUBLIC_KEY = 'pk-env'; + process.env.LANGFUSE_SECRET_KEY = 'sk-env'; + process.env.LANGFUSE_BASE_URL = 'https://env.langfuse.example'; + const { encryptV3 } = await import('@librechat/data-schemas'); + const { buildLangfuseConfig } = await import('./config'); + + expect( + buildLangfuseConfig({ + runId: 'run-1', + appConfig: { + langfuse: { + enabled: true, + publicKey: 'pk-stored', + secretKey: encryptV3('sk-stored'), + destination: 'us', + }, + } as unknown as AppConfig, + }), + ).toEqual({ + deterministicTraceId: true, + publicKey: 'pk-env', + secretKey: 'sk-env', + baseUrl: 'https://env.langfuse.example', + }); + }); + + it('does not trace a disabled stored connection in single-tenant mode', async () => { + delete process.env.TENANT_ISOLATION_STRICT; + const { encryptV3 } = await import('@librechat/data-schemas'); + const { buildLangfuseConfig } = await import('./config'); + + expect( + buildLangfuseConfig({ + runId: 'run-1', + appConfig: { + langfuse: { + enabled: false, + publicKey: 'pk-stored', + secretKey: encryptV3('sk-stored'), + destination: 'us', + }, + } as unknown as AppConfig, + }), + ).toEqual({ + deterministicTraceId: true, + enabled: false, + }); + }); + + it.each(['false', '0', 'no', 'off'])( + 'disables traces when LANGFUSE_TRACING_ENABLED is %s', + async (value) => { + process.env.LANGFUSE_TRACING_ENABLED = value; + process.env.LANGFUSE_PUBLIC_KEY = 'pk-central'; + process.env.LANGFUSE_SECRET_KEY = 'sk-central'; + const { buildLangfuseConfig } = await import('./config'); + + expect(buildLangfuseConfig({ runId: 'run-1' })).toEqual({ + deterministicTraceId: true, + enabled: false, + }); + }, + ); + + it('applies fractional sampling to deterministic run trace IDs', async () => { + process.env.LANGFUSE_SAMPLE_RATE = '0.5'; + process.env.LANGFUSE_PUBLIC_KEY = 'pk-central'; + process.env.LANGFUSE_SECRET_KEY = 'sk-central'; + const { buildLangfuseConfig } = await import('./config'); + + expect(buildLangfuseConfig({ runId: 'sampled-run' })).toEqual({ + deterministicTraceId: true, + enabled: false, + }); + expect(buildLangfuseConfig({ runId: 'unsampled-run' })).toMatchObject({ + deterministicTraceId: true, + publicKey: 'pk-central', + secretKey: 'sk-central', + }); + }); + it('decrypts encrypted tenant secrets for tenant trace export', async () => { + delete process.env.TENANT_ISOLATION_STRICT; process.env.LANGFUSE_FANOUT_ENABLED = 'true'; process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://langfuse-fanout-collector:4318'; const { encryptV3 } = await import('@librechat/data-schemas'); diff --git a/packages/api/src/langfuse/config.ts b/packages/api/src/langfuse/config.ts index 21b97c1d2e..e29ef4a4a1 100644 --- a/packages/api/src/langfuse/config.ts +++ b/packages/api/src/langfuse/config.ts @@ -1,8 +1,17 @@ import type { AppConfig } from '@librechat/data-schemas'; import type { RunConfig } from '@librechat/agents'; -import { isTrueEnv, normalizeBoolean, resolveTenantCredentials } from './utils'; +import { + hasLangfuseEnvCredentials, + isLangfuseFanoutEnabled, + isLangfuseTenantExportEnabled, + isLangfuseTraceSampled, + isLangfuseTracingEnabled, + usesLangfuseMultiTenantRouting, +} from './policy'; import { resolveLangfuseTenantDestination } from './tenantDestinations'; +import { normalizeBoolean, resolveTenantCredentials } from './utils'; import { normalizeString } from '~/utils/text'; +import { traceIdForMessage } from './trace'; type LangfuseRunConfig = NonNullable; type LangfuseRunConfigWithTraceAttributes = LangfuseRunConfig & { @@ -30,16 +39,7 @@ function appendPath(baseUrl: string, path: string): string { return `${baseUrl.replace(/\/+$/, '')}${path}`; } -export function isLangfuseTenantExportEnabled(): boolean { - return !isTrueEnv(process.env.LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED); -} - -export function isLangfuseFanoutEnabled(): boolean { - return ( - isTrueEnv(process.env.LANGFUSE_FANOUT_ENABLED) && - normalizeString(process.env.LANGFUSE_FANOUT_COLLECTOR_URL) != null - ); -} +export { isLangfuseFanoutEnabled, isLangfuseTenantExportEnabled } from './policy'; function mergeTraceMetadata( base: LangfuseRunConfig['metadata'], @@ -127,10 +127,12 @@ function resolveLangfuseExportPlan({ export function buildLangfuseConfig({ appConfig, + runId, tenantId, centralTraceExportEnabled = true, }: { appConfig?: AppConfig; + runId?: string; tenantId?: string; /** * Defaults to true. Set false to suppress central Langfuse export for this @@ -154,6 +156,14 @@ export function buildLangfuseConfig({ langfuse.tags = tags; } + if ( + !isLangfuseTracingEnabled() || + (runId != null && !isLangfuseTraceSampled(traceIdForMessage(runId))) + ) { + langfuse.enabled = false; + return langfuse; + } + const tenantLangfuseEnabled = normalizeBoolean(config?.enabled) === true; if (!centralTraceExportEnabled) { disableCentralExport(langfuse); @@ -165,6 +175,22 @@ export function buildLangfuseConfig({ const fanoutCollectorUrl = normalizeString(process.env.LANGFUSE_FANOUT_COLLECTOR_URL); const tenantDestination = resolveLangfuseTenantDestination(config?.destination); const tenantExportEmergencyEnabled = isLangfuseTenantExportEnabled(); + + if (!usesLangfuseMultiTenantRouting()) { + if (!centralTraceExportEnabled) { + langfuse.enabled = false; + } else if (hasLangfuseEnvCredentials()) { + applyCentralEnvConfig(langfuse); + } else if (tenantLangfuseEnabled && tenantCredentials != null && tenantDestination != null) { + langfuse.publicKey = tenantCredentials.publicKey; + langfuse.secretKey = tenantCredentials.secretKey; + langfuse.baseUrl = tenantDestination.baseUrl; + } else if (config != null) { + langfuse.enabled = false; + } + return langfuse; + } + const exportPlan = resolveLangfuseExportPlan({ centralTraceExportEnabled, fanoutEnabled, diff --git a/packages/api/src/langfuse/destinations.ts b/packages/api/src/langfuse/destinations.ts index 63f07a4a80..3a36683416 100644 --- a/packages/api/src/langfuse/destinations.ts +++ b/packages/api/src/langfuse/destinations.ts @@ -1,37 +1,23 @@ import type { AppConfig } from '@librechat/data-schemas'; import { - isFalseEnv, - normalizeBoolean, - resolveTenantCredentials, - toBasicAuthorization, -} from './utils'; -import { isLangfuseFanoutEnabled, isLangfuseTenantExportEnabled } from './config'; + hasLangfuseEnvCredentials, + isLangfuseFanoutEnabled, + isLangfuseTenantExportEnabled, + isLangfuseTraceSampled, + usesLangfuseMultiTenantRouting, +} from './policy'; +import { normalizeBoolean, resolveTenantCredentials, toBasicAuthorization } from './utils'; import { resolveLangfuseTenantDestination } from './tenantDestinations'; import { normalizeString } from '~/utils/text'; const DEFAULT_BASE_URL = 'https://cloud.langfuse.com'; export type LangfuseScoreDestination = { - name: 'central' | 'tenant'; + name: 'central' | 'tenant' | 'connection'; baseUrl: string; authorization: string; }; -function isSampleRateEnabled(value?: string): boolean { - if (value == null || value.trim() === '') { - return true; - } - const parsed = Number(value); - return !Number.isFinite(parsed) || parsed !== 0; -} - -function isTracingEnabled(): boolean { - return ( - !isFalseEnv(process.env.LANGFUSE_TRACING_ENABLED) && - isSampleRateEnabled(process.env.LANGFUSE_SAMPLE_RATE) - ); -} - function getCentralEnvBaseUrl(): string { return ( normalizeString(process.env.LANGFUSE_BASE_URL) ?? @@ -42,10 +28,6 @@ function getCentralEnvBaseUrl(): string { } function getCentralScoreDestination(): LangfuseScoreDestination | undefined { - if (!isTracingEnabled()) { - return undefined; - } - // Central feedback scores are sent directly by the app, not through the // collector, so they use LibreChat's normal central Langfuse credentials. // LANGFUSE_FANOUT_CENTRAL_AUTH_HEADER is intentionally collector-only. @@ -63,9 +45,6 @@ function getCentralScoreDestination(): LangfuseScoreDestination | undefined { } function getTenantScoreDestination(appConfig?: AppConfig): LangfuseScoreDestination | undefined { - if (!isTracingEnabled()) { - return undefined; - } if (!isLangfuseTenantExportEnabled()) { return undefined; } @@ -98,11 +77,50 @@ function getTenantScoreDestination(appConfig?: AppConfig): LangfuseScoreDestinat }; } +function getConfiguredScoreDestination( + appConfig?: AppConfig, +): LangfuseScoreDestination | undefined { + const config = appConfig?.langfuse; + if (normalizeBoolean(config?.enabled) !== true) { + return undefined; + } + + const credentials = resolveTenantCredentials(config); + const destination = resolveLangfuseTenantDestination(config?.destination); + if (!credentials || !destination) { + return undefined; + } + + return { + name: 'connection', + baseUrl: destination.baseUrl, + authorization: toBasicAuthorization(credentials.publicKey, credentials.secretKey), + }; +} + /** - * Score fanout uses Langfuse's direct REST API. The deployment-level collector - * URL is still required so tenant score fanout follows trace fanout availability. + * Scores use Langfuse's direct REST API. Multi-tenant score fanout follows the + * collector availability gate used by traces; single-tenant connections send + * directly to their configured destination. */ -export function getScoreDestinations(appConfig?: AppConfig): LangfuseScoreDestination[] { +export function getScoreDestinations( + appConfig: AppConfig | undefined, + traceId: string, +): LangfuseScoreDestination[] { + if (!isLangfuseTraceSampled(traceId)) { + return []; + } + + if (!usesLangfuseMultiTenantRouting()) { + return hasLangfuseEnvCredentials() + ? [getCentralScoreDestination()].filter( + (destination): destination is LangfuseScoreDestination => Boolean(destination), + ) + : [getConfiguredScoreDestination(appConfig)].filter( + (destination): destination is LangfuseScoreDestination => Boolean(destination), + ); + } + const destinations = [getCentralScoreDestination(), getTenantScoreDestination(appConfig)].filter( (destination): destination is LangfuseScoreDestination => Boolean(destination), ); diff --git a/packages/api/src/langfuse/feedback.spec.ts b/packages/api/src/langfuse/feedback.spec.ts index 1f625a0b6e..d3c96ff01f 100644 --- a/packages/api/src/langfuse/feedback.spec.ts +++ b/packages/api/src/langfuse/feedback.spec.ts @@ -44,6 +44,7 @@ const langfuseEnvKeys = [ 'LANGFUSE_FANOUT_TENANT_US_BASE_URL', 'LANGFUSE_FANOUT_TENANT_JP_BASE_URL', 'LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED', + 'TENANT_ISOLATION_STRICT', ]; let fetchMock: jest.SpiedFunction; @@ -59,6 +60,7 @@ function setLangfuseCredentials() { } function enableTenantFanout() { + process.env.TENANT_ISOLATION_STRICT = 'true'; process.env.LANGFUSE_FANOUT_ENABLED = 'true'; process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector:4318'; } @@ -176,8 +178,87 @@ describe('Langfuse feedback scores', () => { ); }); + it('posts scores only to the stored connection in single-tenant mode without env credentials', async () => { + delete process.env.LANGFUSE_PUBLIC_KEY; + delete process.env.LANGFUSE_SECRET_KEY; + process.env.LANGFUSE_FANOUT_TENANT_DESTINATIONS = 'us=https://us.cloud.langfuse.example'; + const { sendFeedbackScore } = await loadFeedback(); + + await sendFeedbackScore({ + traceId: '86d413435f8b0d7f32d4d010ce769e2e', + feedback: { rating: 'thumbsUp' }, + appConfig: appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: encryptedTenantSecret(), + destination: 'us', + }), + }); + + expect(getFetchMock()).toHaveBeenCalledTimes(1); + expect(getFetchMock()).toHaveBeenCalledWith( + 'https://us.cloud.langfuse.example/api/public/scores', + expect.objectContaining({ + headers: expect.objectContaining({ Authorization: getTenantAuthorization() }), + }), + ); + }); + + it('keeps scores on environment credentials in single-tenant mode', async () => { + const { sendFeedbackScore } = await loadFeedback(); + + await sendFeedbackScore({ + traceId: '86d413435f8b0d7f32d4d010ce769e2e', + feedback: { rating: 'thumbsUp' }, + appConfig: appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: encryptedTenantSecret(), + destination: 'us', + }), + }); + + expect(getFetchMock()).toHaveBeenCalledTimes(1); + expect(getFetchMock()).toHaveBeenCalledWith( + 'https://cloud.langfuse.com/api/public/scores', + expect.objectContaining({ + headers: expect.objectContaining({ Authorization: getCentralAuthorization() }), + }), + ); + }); + + it('does not send scores for a disabled stored connection in single-tenant mode', async () => { + delete process.env.LANGFUSE_PUBLIC_KEY; + delete process.env.LANGFUSE_SECRET_KEY; + const { sendFeedbackScore } = await loadFeedback(); + + await sendFeedbackScore({ + traceId: '86d413435f8b0d7f32d4d010ce769e2e', + feedback: { rating: 'thumbsUp' }, + appConfig: appConfigWithLangfuse({ + enabled: false, + publicKey: 'tenant-public-key', + secretKey: encryptedTenantSecret(), + destination: 'us', + }), + }); + + expect(getFetchMock()).not.toHaveBeenCalled(); + }); + + it('does not send a score for a trace excluded by fractional sampling', async () => { + process.env.LANGFUSE_SAMPLE_RATE = '0.5'; + const { sendFeedbackScore } = await loadFeedback(); + + await sendFeedbackScore({ + traceId: '658f74b0a232417fc3e6e4d9ef5f563a', + feedback: { rating: 'thumbsUp' }, + }); + + expect(getFetchMock()).not.toHaveBeenCalled(); + }); + it('posts feedback scores to central fanout and tenant Langfuse projects', async () => { enableTenantFanout(); + delete process.env.TENANT_ISOLATION_STRICT; process.env.LANGFUSE_BASE_URL = 'http://central-langfuse:3000'; process.env.LANGFUSE_FANOUT_TENANT_DESTINATIONS = 'eu=http://tenant-langfuse:3000'; const { sendFeedbackScore } = await loadFeedback(); @@ -805,6 +886,7 @@ describe('Langfuse feedback scores', () => { it.each(['true', '1', 'yes', 'on'])( 'enables tenant scores when global fanout is %s', async (value) => { + process.env.TENANT_ISOLATION_STRICT = 'true'; process.env.LANGFUSE_FANOUT_ENABLED = value; process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector:4318'; process.env.LANGFUSE_BASE_URL = 'http://central-langfuse:3000'; @@ -834,6 +916,7 @@ describe('Langfuse feedback scores', () => { it.each(['false', '0', 'no', 'off'])( 'keeps tenant scores disabled when global fanout is %s', async (value) => { + process.env.TENANT_ISOLATION_STRICT = 'true'; process.env.LANGFUSE_FANOUT_ENABLED = value; process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector:4318'; process.env.LANGFUSE_BASE_URL = 'http://central-langfuse:3000'; diff --git a/packages/api/src/langfuse/feedback.ts b/packages/api/src/langfuse/feedback.ts index 0e58ade588..a355e34007 100644 --- a/packages/api/src/langfuse/feedback.ts +++ b/packages/api/src/langfuse/feedback.ts @@ -111,7 +111,7 @@ export async function sendFeedbackScore({ return; } - const destinations = getScoreDestinations(appConfig); + const destinations = getScoreDestinations(appConfig, traceId); if (destinations.length === 0) { return; } diff --git a/packages/api/src/langfuse/index.ts b/packages/api/src/langfuse/index.ts index 43a5de0f83..ae4d0be745 100644 --- a/packages/api/src/langfuse/index.ts +++ b/packages/api/src/langfuse/index.ts @@ -1,3 +1,3 @@ export * from './feedback'; +export * from './policy'; export * from './trace'; -export { isLangfuseFanoutEnabled } from './config'; diff --git a/packages/api/src/langfuse/policy.spec.ts b/packages/api/src/langfuse/policy.spec.ts new file mode 100644 index 0000000000..5bdbe02d28 --- /dev/null +++ b/packages/api/src/langfuse/policy.spec.ts @@ -0,0 +1,103 @@ +const envKeys = [ + 'LANGFUSE_PUBLIC_KEY', + 'LANGFUSE_SECRET_KEY', + 'LANGFUSE_TRACING_ENABLED', + 'LANGFUSE_SAMPLE_RATE', + 'LANGFUSE_FANOUT_ENABLED', + 'LANGFUSE_FANOUT_COLLECTOR_URL', + 'TENANT_ISOLATION_STRICT', +]; + +function clearEnv() { + for (const key of envKeys) { + delete process.env[key]; + } +} + +describe('Langfuse policy', () => { + beforeEach(clearEnv); + afterEach(clearEnv); + + it('offers connection settings by default in single-tenant deployments', async () => { + const { isLangfuseConnectionAvailable } = await import('./policy'); + + expect(isLangfuseConnectionAvailable()).toBe(true); + }); + + it('hides single-tenant settings when environment credentials own the connection', async () => { + process.env.LANGFUSE_PUBLIC_KEY = 'pk-env'; + process.env.LANGFUSE_SECRET_KEY = 'sk-env'; + const { isLangfuseConnectionAvailable } = await import('./policy'); + + expect(isLangfuseConnectionAvailable()).toBe(false); + }); + + it('does not hide settings for incomplete environment credentials', async () => { + process.env.LANGFUSE_PUBLIC_KEY = 'pk-env'; + const { isLangfuseConnectionAvailable } = await import('./policy'); + + expect(isLangfuseConnectionAvailable()).toBe(true); + }); + + it('requires fanout in strict multi-tenant deployments', async () => { + process.env.TENANT_ISOLATION_STRICT = 'true'; + const { isLangfuseConnectionAvailable } = await import('./policy'); + + expect(isLangfuseConnectionAvailable()).toBe(false); + + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://collector:4318'; + expect(isLangfuseConnectionAvailable()).toBe(true); + }); + + it('uses explicit fanout routing without requiring strict tenant isolation', async () => { + 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:4318'; + const { isLangfuseConnectionAvailable, usesLangfuseMultiTenantRouting } = await import( + './policy' + ); + + expect(usesLangfuseMultiTenantRouting()).toBe(true); + expect(isLangfuseConnectionAvailable()).toBe(true); + }); + + it.each(['false', '0', 'no', 'off'])( + 'hides settings when tracing is disabled with %s', + async (value) => { + process.env.LANGFUSE_TRACING_ENABLED = value; + const { isLangfuseConnectionAvailable } = await import('./policy'); + + expect(isLangfuseConnectionAvailable()).toBe(false); + }, + ); + + it('hides settings when the sample rate is zero', async () => { + process.env.LANGFUSE_SAMPLE_RATE = '0'; + const { isLangfuseConnectionAvailable } = await import('./policy'); + + expect(isLangfuseConnectionAvailable()).toBe(false); + }); + + it('samples traces deterministically at fractional sample rates', async () => { + process.env.LANGFUSE_SAMPLE_RATE = '0.5'; + const { isLangfuseTraceSampled } = await import('./policy'); + + expect(isLangfuseTraceSampled('86d413435f8b0d7f32d4d010ce769e2e')).toBe(true); + expect(isLangfuseTraceSampled('658f74b0a232417fc3e6e4d9ef5f563a')).toBe(false); + }); + + it('clamps numeric sample rates and preserves tracing for invalid values', async () => { + const { getLangfuseSampleRate } = await import('./policy'); + + process.env.LANGFUSE_SAMPLE_RATE = '-1'; + expect(getLangfuseSampleRate()).toBe(0); + + process.env.LANGFUSE_SAMPLE_RATE = '2'; + expect(getLangfuseSampleRate()).toBe(1); + + process.env.LANGFUSE_SAMPLE_RATE = 'invalid'; + expect(getLangfuseSampleRate()).toBe(1); + }); +}); diff --git a/packages/api/src/langfuse/policy.ts b/packages/api/src/langfuse/policy.ts new file mode 100644 index 0000000000..4a79fcd058 --- /dev/null +++ b/packages/api/src/langfuse/policy.ts @@ -0,0 +1,81 @@ +import { isFalseEnv, isTrueEnv } from './utils'; +import { normalizeString } from '~/utils/text'; + +const DEFAULT_SAMPLE_RATE = 1; +const MAX_TRACE_ID_ACCUMULATION = 0xffffffff; + +export function isLangfuseTenantExportEnabled(): boolean { + return !isTrueEnv(process.env.LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED); +} + +export function isLangfuseFanoutEnabled(): boolean { + return ( + isTrueEnv(process.env.LANGFUSE_FANOUT_ENABLED) && + normalizeString(process.env.LANGFUSE_FANOUT_COLLECTOR_URL) != null + ); +} + +export function hasLangfuseEnvCredentials(): boolean { + return ( + normalizeString(process.env.LANGFUSE_PUBLIC_KEY) != null && + normalizeString(process.env.LANGFUSE_SECRET_KEY) != null + ); +} + +export function usesLangfuseMultiTenantRouting(): boolean { + return process.env.TENANT_ISOLATION_STRICT === 'true' || isLangfuseFanoutEnabled(); +} + +export function getLangfuseSampleRate(): number { + const value = normalizeString(process.env.LANGFUSE_SAMPLE_RATE); + if (value == null) { + return DEFAULT_SAMPLE_RATE; + } + + const sampleRate = Number(value); + if (!Number.isFinite(sampleRate)) { + return DEFAULT_SAMPLE_RATE; + } + return Math.min(1, Math.max(0, sampleRate)); +} + +export function isLangfuseTracingEnabled(): boolean { + return !isFalseEnv(process.env.LANGFUSE_TRACING_ENABLED) && getLangfuseSampleRate() > 0; +} + +function traceIdAccumulation(traceId: string): number { + // Match OpenTelemetry's TraceIdRatioBasedSampler so one trace has one stable + // sampling decision across trace export and later feedback scores. + let accumulation = 0; + for (let offset = 0; offset < 32; offset += 8) { + const part = Number.parseInt(traceId.slice(offset, offset + 8), 16); + accumulation = (accumulation ^ part) >>> 0; + } + return accumulation; +} + +export function isLangfuseTraceSampled(traceId: string): boolean { + if (!isLangfuseTracingEnabled()) { + return false; + } + + const sampleRate = getLangfuseSampleRate(); + if (sampleRate >= 1) { + return true; + } + if (!/^[0-9a-f]{32}$/i.test(traceId)) { + return false; + } + + return traceIdAccumulation(traceId) < Math.floor(sampleRate * MAX_TRACE_ID_ACCUMULATION); +} + +export function isLangfuseConnectionAvailable(): boolean { + if (!isLangfuseTracingEnabled()) { + return false; + } + if (usesLangfuseMultiTenantRouting()) { + return isLangfuseFanoutEnabled(); + } + return !hasLangfuseEnvCredentials(); +}