diff --git a/api/cache/distributedLock.js b/api/cache/distributedLock.js new file mode 100644 index 00000000000..4fe83e415a3 --- /dev/null +++ b/api/cache/distributedLock.js @@ -0,0 +1,6 @@ +const { ioredisClient, createDistributedLock } = require('@librechat/api'); + +/** The one `DistributedLock` for this process. Redis-backed when `ioredisClient` is set; otherwise an in-process lock that coordinates only callers holding this instance. */ +const distributedLock = createDistributedLock({ redisClient: ioredisClient }); + +module.exports = { distributedLock }; diff --git a/api/cache/distributedLock.spec.js b/api/cache/distributedLock.spec.js new file mode 100644 index 00000000000..97ae3563f6b --- /dev/null +++ b/api/cache/distributedLock.spec.js @@ -0,0 +1,18 @@ +const mockLock = { acquire: jest.fn(), extend: jest.fn(), release: jest.fn() }; +const mockCreateDistributedLock = jest.fn(() => mockLock); +const sentinelRedisClient = {}; + +jest.mock('@librechat/api', () => ({ + createDistributedLock: mockCreateDistributedLock, + ioredisClient: sentinelRedisClient, +})); + +describe('distributedLock', () => { + it('builds the process-wide lock once with the shared Redis client', () => { + const { distributedLock } = require('./distributedLock'); + + expect(mockCreateDistributedLock).toHaveBeenCalledTimes(1); + expect(mockCreateDistributedLock).toHaveBeenCalledWith({ redisClient: sentinelRedisClient }); + expect(distributedLock).toBe(mockLock); + }); +}); diff --git a/api/cache/index.js b/api/cache/index.js index 1bad242f13f..48df760bffa 100644 --- a/api/cache/index.js +++ b/api/cache/index.js @@ -1,4 +1,5 @@ const getLogStores = require('./getLogStores'); const logViolation = require('./logViolation'); +const { distributedLock } = require('./distributedLock'); -module.exports = { getLogStores, logViolation }; +module.exports = { getLogStores, logViolation, distributedLock }; diff --git a/api/server/routes/__tests__/config.spec.js b/api/server/routes/__tests__/config.spec.js index 51e8db8ada7..2a295e73a0a 100644 --- a/api/server/routes/__tests__/config.spec.js +++ b/api/server/routes/__tests__/config.spec.js @@ -10,10 +10,10 @@ jest.mock('~/server/services/Config/ldap', () => ({ })); const mockHasCapability = jest.fn(); -const mockHasConfigCapability = jest.fn(); +const mockGetHeldCapabilities = jest.fn(); jest.mock('~/server/middleware/roles/capabilities', () => ({ hasCapability: (...args) => mockHasCapability(...args), - hasConfigCapability: (...args) => mockHasConfigCapability(...args), + getHeldCapabilities: (...args) => mockGetHeldCapabilities(...args), })); const mockGetTenantId = jest.fn(() => undefined); @@ -37,6 +37,7 @@ jest.mock('@librechat/api', () => ({ const request = require('supertest'); const express = require('express'); +const { SystemCapabilities, configCapability } = require('@librechat/data-schemas'); const configRoute = require('../config'); function createApp(user) { @@ -816,10 +817,18 @@ describe('GET /api/config', () => { }); }); + /** A held-capability set that grants connection access and prompt-sync manage access. */ + const fullLangfuseAccess = () => + new Set([ + SystemCapabilities.ACCESS_ADMIN, + SystemCapabilities.MANAGE_CONFIGS, + configCapability('langfuse'), + SystemCapabilities.MANAGE_PROMPTS, + ]); + it('should advertise Langfuse fanout only when the toggle and collector URL are configured', async () => { mockGetAppConfig.mockResolvedValue(baseAppConfig); - mockHasCapability.mockResolvedValue(true); - mockHasConfigCapability.mockResolvedValue(true); + mockGetHeldCapabilities.mockResolvedValue(fullLangfuseAccess()); process.env.LANGFUSE_FANOUT_ENABLED = 'true'; const app = createApp(mockUser); @@ -840,8 +849,7 @@ describe('GET /api/config', () => { it('hides Langfuse connection access when tenant export is emergency-disabled', async () => { mockGetAppConfig.mockResolvedValue(baseAppConfig); - mockHasCapability.mockResolvedValue(true); - mockHasConfigCapability.mockResolvedValue(true); + mockGetHeldCapabilities.mockResolvedValue(fullLangfuseAccess()); process.env.LANGFUSE_FANOUT_ENABLED = 'true'; process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://langfuse-fanout:4318'; process.env.LANGFUSE_FANOUT_TENANT_EXPORT_DISABLED = 'true'; @@ -851,8 +859,7 @@ describe('GET /api/config', () => { expect(response.body.langfuseFanoutEnabled).toBe(true); expect(response.body.langfuseConnectionAccess).toBe(false); - expect(mockHasCapability).not.toHaveBeenCalled(); - expect(mockHasConfigCapability).not.toHaveBeenCalled(); + expect(mockGetHeldCapabilities).not.toHaveBeenCalled(); }); it('advertises Langfuse connection access from capabilities rather than the user role', async () => { @@ -862,34 +869,81 @@ describe('GET /api/config', () => { process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://langfuse-fanout:4318'; const app = createApp({ ...mockUser, role: 'DELEGATED_ADMIN' }); - mockHasCapability.mockImplementation( - async (_user, capability) => capability === 'access:admin', + mockGetHeldCapabilities.mockResolvedValue( + new Set([SystemCapabilities.ACCESS_ADMIN, configCapability('langfuse')]), ); - mockHasConfigCapability.mockResolvedValue(true); let response = await request(app).get('/api/config'); expect(response.body.langfuseConnectionAccess).toBe(true); - mockHasConfigCapability.mockResolvedValue(false); + mockGetHeldCapabilities.mockResolvedValue(new Set([SystemCapabilities.ACCESS_ADMIN])); response = await request(app).get('/api/config'); expect(response.body.langfuseFanoutEnabled).toBe(true); expect(response.body.langfuseConnectionAccess).toBe(false); }); - it('skips the Langfuse management capability check without admin access', async () => { + it('advertises prompt-sync manage access alongside connection access when the admin holds manage:prompts', async () => { mockGetAppConfig.mockResolvedValue(baseAppConfig); - mockHasCapability.mockResolvedValue(false); + mockGetHeldCapabilities.mockResolvedValue(fullLangfuseAccess()); + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://langfuse-fanout:4318'; + const app = createApp(mockUser); + + const response = await request(app).get('/api/config'); + + expect(response.body.langfuseConnectionAccess).toBe(true); + expect(response.body.langfusePromptManageAccess).toBe(true); + expect(mockGetHeldCapabilities).toHaveBeenCalledWith( + expect.anything(), + expect.arrayContaining([SystemCapabilities.MANAGE_PROMPTS]), + ); + }); + + it('withholds prompt-sync manage access when the admin lacks manage:prompts', async () => { + mockGetAppConfig.mockResolvedValue(baseAppConfig); + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://langfuse-fanout:4318'; + const app = createApp(mockUser); + + mockGetHeldCapabilities.mockResolvedValue( + new Set([SystemCapabilities.ACCESS_ADMIN, SystemCapabilities.MANAGE_CONFIGS]), + ); + + const response = await request(app).get('/api/config'); + + expect(response.body.langfuseConnectionAccess).toBe(true); + expect(response.body.langfusePromptManageAccess).toBe(false); + }); + + it('withholds prompt-sync manage access when Langfuse connection access itself is unavailable, even if manage:prompts is held', async () => { + mockGetAppConfig.mockResolvedValue(baseAppConfig); + process.env.LANGFUSE_FANOUT_ENABLED = 'true'; + process.env.LANGFUSE_FANOUT_COLLECTOR_URL = 'http://langfuse-fanout:4318'; + const app = createApp(mockUser); + + mockGetHeldCapabilities.mockResolvedValue( + new Set([SystemCapabilities.ACCESS_ADMIN, SystemCapabilities.MANAGE_PROMPTS]), + ); + + const response = await request(app).get('/api/config'); + + expect(response.body.langfuseConnectionAccess).toBe(false); + expect(response.body.langfusePromptManageAccess).toBe(false); + expect(mockGetHeldCapabilities).toHaveBeenCalledTimes(1); + }); + + it('withholds Langfuse connection access without admin access', async () => { + mockGetAppConfig.mockResolvedValue(baseAppConfig); + mockGetHeldCapabilities.mockResolvedValue(new Set()); const app = createApp(mockUser); const response = await request(app).get('/api/config'); expect(response.body.langfuseConnectionAccess).toBe(false); - expect(mockHasConfigCapability).not.toHaveBeenCalled(); }); it('advertises Langfuse connection access by default in single-tenant mode', async () => { mockGetAppConfig.mockResolvedValue(baseAppConfig); - mockHasCapability.mockResolvedValue(true); - mockHasConfigCapability.mockResolvedValue(true); + mockGetHeldCapabilities.mockResolvedValue(fullLangfuseAccess()); const app = createApp(mockUser); const response = await request(app).get('/api/config'); @@ -900,8 +954,7 @@ describe('GET /api/config', () => { it('hides single-tenant connection settings when environment credentials are configured', async () => { mockGetAppConfig.mockResolvedValue(baseAppConfig); - mockHasCapability.mockResolvedValue(true); - mockHasConfigCapability.mockResolvedValue(true); + mockGetHeldCapabilities.mockResolvedValue(fullLangfuseAccess()); process.env.LANGFUSE_PUBLIC_KEY = 'pk-env'; process.env.LANGFUSE_SECRET_KEY = 'sk-env'; const app = createApp(mockUser); @@ -909,8 +962,7 @@ describe('GET /api/config', () => { const response = await request(app).get('/api/config'); expect(response.body.langfuseConnectionAccess).toBe(false); - expect(mockHasCapability).not.toHaveBeenCalled(); - expect(mockHasConfigCapability).not.toHaveBeenCalled(); + expect(mockGetHeldCapabilities).not.toHaveBeenCalled(); }); it.each([ @@ -918,15 +970,14 @@ describe('GET /api/config', () => { ['LANGFUSE_SAMPLE_RATE', '0'], ])('hides Langfuse connection settings when %s=%s', async (key, value) => { mockGetAppConfig.mockResolvedValue(baseAppConfig); - mockHasCapability.mockResolvedValue(true); - mockHasConfigCapability.mockResolvedValue(true); + mockGetHeldCapabilities.mockResolvedValue(fullLangfuseAccess()); 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(); + expect(mockGetHeldCapabilities).not.toHaveBeenCalled(); }); it.each([ @@ -945,14 +996,13 @@ describe('GET /api/config', () => { ])('advertises prompt access with %s without advertising trace access', async (_mode, env) => { Object.assign(process.env, env, { LANGFUSE_PROMPT_SYNC_AVAILABLE: 'true' }); mockGetAppConfig.mockResolvedValue(baseAppConfig); - mockHasCapability.mockResolvedValue(true); - mockHasConfigCapability.mockResolvedValue(true); + mockGetHeldCapabilities.mockResolvedValue(fullLangfuseAccess()); const response = await request(createApp(mockUser)).get('/api/config'); expect(response.status).toBe(200); expect(response.body.langfusePromptSyncAccess).toBe(true); expect(response.body.langfuseConnectionAccess).toBe(false); - expect(mockHasCapability).toHaveBeenCalledTimes(1); - expect(mockHasConfigCapability).toHaveBeenCalledTimes(1); + expect(response.body.langfusePromptManageAccess).toBe(true); + expect(mockGetHeldCapabilities).toHaveBeenCalledTimes(1); }); it.each([ @@ -962,13 +1012,33 @@ describe('GET /api/config', () => { process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; process.env.LANGFUSE_TRACING_ENABLED = 'false'; mockGetAppConfig.mockResolvedValue(baseAppConfig); - mockHasCapability.mockResolvedValue(admin); - mockHasConfigCapability.mockResolvedValue(manage); + mockGetHeldCapabilities.mockResolvedValue( + new Set([ + ...(admin ? [SystemCapabilities.ACCESS_ADMIN] : []), + ...(manage ? [configCapability('langfuse')] : []), + SystemCapabilities.MANAGE_PROMPTS, + ]), + ); const response = await request(createApp(mockUser)).get('/api/config'); expect(response.body.langfusePromptSyncAccess).toBe(false); expect(response.body.langfuseConnectionAccess).toBe(false); }); + it('shows prompt settings without sync permission when tracing is disabled', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + process.env.LANGFUSE_TRACING_ENABLED = 'false'; + mockGetAppConfig.mockResolvedValue(baseAppConfig); + mockGetHeldCapabilities.mockResolvedValue( + new Set([SystemCapabilities.ACCESS_ADMIN, configCapability('langfuse')]), + ); + + const response = await request(createApp(mockUser)).get('/api/config'); + + expect(response.body.langfusePromptSyncAccess).toBe(true); + expect(response.body.langfuseConnectionAccess).toBe(false); + expect(response.body.langfusePromptManageAccess).toBe(false); + }); + it('should include post-login informational fields', async () => { process.env.ANALYTICS_GTM_ID = 'GTM-XYZ'; process.env.CUSTOM_FOOTER = 'authenticated footer text'; diff --git a/api/server/routes/admin/langfuse.js b/api/server/routes/admin/langfuse.js index 10e91ea4443..0102bf6658a 100644 --- a/api/server/routes/admin/langfuse.js +++ b/api/server/routes/admin/langfuse.js @@ -5,6 +5,7 @@ const { hasConfigCapability, requireCapability, } = require('~/server/middleware/roles/capabilities'); +const { syncLangfusePromptCatalog } = require('~/server/services/LangfusePromptSync'); const { invalidateConfigCaches } = require('~/server/services/Config'); const configMiddleware = require('~/server/middleware/config/app'); const { requireJwtAuth } = require('~/server/middleware'); @@ -13,6 +14,7 @@ const db = require('~/models'); const router = express.Router(); const requireAdminAccess = requireCapability(SystemCapabilities.ACCESS_ADMIN); +const requireManagePrompts = requireCapability(SystemCapabilities.MANAGE_PROMPTS); async function requireLangfuseManage(req, res, next) { try { @@ -40,6 +42,7 @@ const handlers = createAdminLangfuseHandlers({ patchConfigFields: db.patchConfigFields, toggleConfigActive: db.toggleConfigActive, getMessages: db.getMessages, + syncPromptCatalog: syncLangfusePromptCatalog, invalidateConfigCaches, recordAuditEntry: db.recordAuditEntry, }); @@ -55,6 +58,7 @@ router.put('/connection', handlers.updateConnection); router.post('/connection/test', handlers.testConnection); router.put('/prompt-sync', handlers.updatePromptSync); router.get('/prompts', handlers.listPrompts); +router.post('/prompts/sync', requireManagePrompts, handlers.syncPrompts); router.get('/prompts/:name', handlers.getPrompt); module.exports = router; diff --git a/api/server/routes/admin/langfuse.test.js b/api/server/routes/admin/langfuse.test.js index fde9b5be969..4ff8a83ea3b 100644 --- a/api/server/routes/admin/langfuse.test.js +++ b/api/server/routes/admin/langfuse.test.js @@ -24,19 +24,27 @@ const mockHandlers = { testConnection: jest.fn((_req, res) => res.status(200).json({ handler: 'test' })), updatePromptSync: jest.fn((_req, res) => res.status(200).json({ handler: 'prompt-sync' })), listPrompts: jest.fn((_req, res) => res.status(200).json({ handler: 'prompts' })), + syncPrompts: jest.fn((_req, res) => res.status(200).json({ handler: 'sync' })), getPrompt: jest.fn((req, res) => res.status(200).json({ handler: 'prompt', name: req.params.name }), ), }; +const mockSyncPromptCatalog = jest.fn(); +const mockLogger = { warn: jest.fn(), error: jest.fn() }; jest.mock('@librechat/data-schemas', () => ({ - SystemCapabilities: { ACCESS_ADMIN: 'access:admin' }, + logger: mockLogger, + SystemCapabilities: { ACCESS_ADMIN: 'access:admin', MANAGE_PROMPTS: 'manage:prompts' }, })); jest.mock('@librechat/api', () => ({ createAdminLangfuseHandlers: jest.fn(() => mockHandlers), })); +jest.mock('~/server/services/LangfusePromptSync', () => ({ + syncLangfusePromptCatalog: mockSyncPromptCatalog, +})); + jest.mock('~/server/middleware/roles/capabilities', () => ({ requireCapability: mockRequireCapability, hasConfigCapability: mockHasConfigCapability, @@ -78,6 +86,18 @@ describe('admin Langfuse routes', () => { return app; } + // The router module builds its handler dependencies once, at `require` + // time, not per request. `createApp()` only reloads the module the first + // time it runs in this file (later `beforeEach` clears already took effect + // by the time any other test observes it), so this captures that one + // load's dependency wiring before anything can clear it. + let handlerDepsAtLoad; + beforeAll(() => { + createApp(); + const { createAdminLangfuseHandlers } = require('@librechat/api'); + handlerDepsAtLoad = createAdminLangfuseHandlers.mock.calls[0]?.[0]; + }); + beforeEach(() => { deniedCapability = undefined; canManageLangfuse = true; @@ -138,6 +158,7 @@ describe('admin Langfuse routes', () => { ['PUT', '/api/admin/langfuse/prompt-sync', 'updatePromptSync'], ['GET', '/api/admin/langfuse/prompts', 'listPrompts'], ['GET', '/api/admin/langfuse/prompts/greeting', 'getPrompt'], + ['POST', '/api/admin/langfuse/prompts/sync', 'syncPrompts'], ])('blocks %s %s without Langfuse manage access', async (method, path, handlerName) => { canManageLangfuse = false; @@ -149,6 +170,7 @@ describe('admin Langfuse routes', () => { it.each([ ['GET', '/api/admin/langfuse/prompts', 'listPrompts'], ['GET', '/api/admin/langfuse/prompts/greeting', 'getPrompt'], + ['POST', '/api/admin/langfuse/prompts/sync', 'syncPrompts'], ])('rejects %s %s for an unauthenticated caller', async (method, path, handlerName) => { mockRequireJwtAuth.mockImplementationOnce((_req, res) => { middlewareCalls.push('jwt'); @@ -163,6 +185,7 @@ describe('admin Langfuse routes', () => { it.each([ ['GET', '/api/admin/langfuse/prompts', 'listPrompts'], ['GET', '/api/admin/langfuse/prompts/greeting', 'getPrompt'], + ['POST', '/api/admin/langfuse/prompts/sync', 'syncPrompts'], ])('rejects %s %s for a non-admin caller', async (method, path, handlerName) => { deniedCapability = 'access:admin'; @@ -204,4 +227,32 @@ describe('admin Langfuse routes', () => { expect(mockConfigMiddleware).not.toHaveBeenCalled(); }); + + it('passes the composed Langfuse prompt-sync runner to the handler factory, not its dependencies', () => { + expect(handlerDepsAtLoad).toMatchObject({ syncPromptCatalog: mockSyncPromptCatalog }); + expect(handlerDepsAtLoad).not.toHaveProperty('lock'); + expect(handlerDepsAtLoad).not.toHaveProperty('grantPermission'); + expect(handlerDepsAtLoad).not.toHaveProperty('findLangfusePromptGroup'); + expect(handlerDepsAtLoad).not.toHaveProperty('createLangfusePromptGroup'); + expect(handlerDepsAtLoad).not.toHaveProperty('deletePromptGroup'); + }); + + it('requires manage:prompts access, in addition to Langfuse manage access, for POST /prompts/sync', async () => { + const response = await request(createApp()) + .post('/api/admin/langfuse/prompts/sync') + .send({}) + .expect(200); + + expect(response.body).toEqual({ handler: 'sync' }); + expect(middlewareCalls).toEqual(['jwt', 'access:admin', 'config', 'manage:prompts']); + expect(mockHandlers.syncPrompts).toHaveBeenCalledTimes(1); + }); + + it('blocks POST /prompts/sync for a caller without manage:prompts access', async () => { + deniedCapability = 'manage:prompts'; + + await request(createApp()).post('/api/admin/langfuse/prompts/sync').send({}).expect(403); + + expect(mockHandlers.syncPrompts).not.toHaveBeenCalled(); + }); }); diff --git a/api/server/routes/config.js b/api/server/routes/config.js index b7908e24eac..a7d32cd15d8 100644 --- a/api/server/routes/config.js +++ b/api/server/routes/config.js @@ -34,8 +34,8 @@ const { defaultSocialLogins, resolveMCPAppsPolicy, } = require('librechat-data-provider'); -const { logger, SystemCapabilities } = require('@librechat/data-schemas'); -const { hasCapability, hasConfigCapability } = require('~/server/middleware/roles/capabilities'); +const { logger, SystemCapabilities, configCapability } = require('@librechat/data-schemas'); +const { hasCapability, getHeldCapabilities } = require('~/server/middleware/roles/capabilities'); const { getLdapConfig } = require('~/server/services/Config/ldap'); const { getRumConfig } = require('~/server/services/Config/rum'); const { getAppConfig } = require('~/server/services/Config/app'); @@ -278,6 +278,8 @@ router.get('/', async function (req, res) { const langfuseFanoutEnabled = isLangfuseFanoutEnabled(); const langfuseSettingsAvailable = isLangfuseSettingsAvailable(); let canManageLangfuse = false; + // Running a sync also requires the prompt-management capability. + let langfusePromptManageAccess = false; if (langfuseSettingsAvailable) { try { @@ -289,13 +291,18 @@ router.get('/', async function (req, res) { tenantId: req.user.tenantId, idOnTheSource: req.user.idOnTheSource ?? null, }; - const hasAdminAccess = await hasCapability( - capabilityUser, + const held = await getHeldCapabilities(capabilityUser, [ SystemCapabilities.ACCESS_ADMIN, - ); - if (hasAdminAccess) { - canManageLangfuse = await hasConfigCapability(capabilityUser, 'langfuse'); - } + SystemCapabilities.MANAGE_CONFIGS, + configCapability('langfuse'), + SystemCapabilities.MANAGE_PROMPTS, + ]); + const hasAdminAccess = held.has(SystemCapabilities.ACCESS_ADMIN); + canManageLangfuse = + hasAdminAccess && + (held.has(SystemCapabilities.MANAGE_CONFIGS) || held.has(configCapability('langfuse'))); + langfusePromptManageAccess = + canManageLangfuse && held.has(SystemCapabilities.MANAGE_PROMPTS); } } catch (err) { logger.warn(`[config] Langfuse capability check failed: ${err.message}`); @@ -331,6 +338,7 @@ router.get('/', async function (req, res) { : 0, langfuseFanoutEnabled, ...getLangfuseSettingsAccess(canManageLangfuse), + langfusePromptManageAccess, insightsEnabled: isEnabled(process.env.ENABLE_INSIGHTS), /** Lets the client skip the pull request lookup entirely when the feature is off. */ ...resolvePullRequestCapabilities(appConfig, process.env), diff --git a/api/server/services/LangfusePromptSync.js b/api/server/services/LangfusePromptSync.js new file mode 100644 index 00000000000..b672cec7303 --- /dev/null +++ b/api/server/services/LangfusePromptSync.js @@ -0,0 +1,17 @@ +const { createLangfusePromptSyncRunner } = require('@librechat/api'); +const { logger } = require('@librechat/data-schemas'); +const { distributedLock } = require('~/cache'); +const { grantPermission } = require('~/server/services/PermissionService'); +const db = require('~/models'); + +/** The one Langfuse prompt-sync runner for this process, bound to the shared `distributedLock`. */ +const syncLangfusePromptCatalog = createLangfusePromptSyncRunner({ + lock: distributedLock, + findLangfusePromptGroup: db.findLangfusePromptGroup, + createLangfusePromptGroup: db.createLangfusePromptGroup, + deletePromptGroup: db.deletePromptGroup, + grantPermission, + logger, +}); + +module.exports = { syncLangfusePromptCatalog }; diff --git a/api/server/services/LangfusePromptSync.spec.js b/api/server/services/LangfusePromptSync.spec.js new file mode 100644 index 00000000000..c899fa66239 --- /dev/null +++ b/api/server/services/LangfusePromptSync.spec.js @@ -0,0 +1,45 @@ +const mockLock = { acquire: jest.fn(), extend: jest.fn(), release: jest.fn() }; +const mockSyncPromptCatalog = jest.fn(); +const mockCreateLangfusePromptSyncRunner = jest.fn(() => mockSyncPromptCatalog); +const mockGrantPermission = jest.fn(); +const mockLogger = { warn: jest.fn(), error: jest.fn() }; + +jest.mock('@librechat/api', () => ({ + createLangfusePromptSyncRunner: mockCreateLangfusePromptSyncRunner, +})); + +jest.mock('@librechat/data-schemas', () => ({ + logger: mockLogger, +})); + +jest.mock('~/cache', () => ({ + distributedLock: mockLock, +})); + +jest.mock('~/server/services/PermissionService', () => ({ + grantPermission: mockGrantPermission, +})); + +jest.mock('~/models', () => ({ + findLangfusePromptGroup: jest.fn(), + createLangfusePromptGroup: jest.fn(), + deletePromptGroup: jest.fn(), +})); + +describe('LangfusePromptSync', () => { + it('composes the sync runner once with the shared lock and the db methods', () => { + const db = require('~/models'); + const { syncLangfusePromptCatalog } = require('./LangfusePromptSync'); + + expect(mockCreateLangfusePromptSyncRunner).toHaveBeenCalledTimes(1); + expect(mockCreateLangfusePromptSyncRunner).toHaveBeenCalledWith({ + lock: mockLock, + findLangfusePromptGroup: db.findLangfusePromptGroup, + createLangfusePromptGroup: db.createLangfusePromptGroup, + deletePromptGroup: db.deletePromptGroup, + grantPermission: mockGrantPermission, + logger: mockLogger, + }); + expect(syncLangfusePromptCatalog).toBe(mockSyncPromptCatalog); + }); +}); diff --git a/client/src/components/Nav/SettingsTabs/Integrations/LangfuseConnection.tsx b/client/src/components/Nav/SettingsTabs/Integrations/LangfuseConnection.tsx index ca0117e842c..09026fbbc48 100644 --- a/client/src/components/Nav/SettingsTabs/Integrations/LangfuseConnection.tsx +++ b/client/src/components/Nav/SettingsTabs/Integrations/LangfuseConnection.tsx @@ -18,12 +18,15 @@ import type { } from 'librechat-data-provider'; import type { TranslationKeys } from '~/hooks'; import { + useGetStartupConfig, useGetLangfuseConnectionQuery, useUpdateLangfuseConnectionMutation, useTestLangfuseConnectionMutation, useUpdateLangfusePromptSyncMutation, + useSyncLangfusePromptsMutation, } from '~/data-provider'; import LangfusePromptsDialog from './LangfusePromptsDialog'; +import { getPromptErrorMessage } from './promptErrors'; import { useLocalize } from '~/hooks'; type ConnectionTestState = 'idle' | 'unverified' | 'checking' | 'connected' | 'failed'; @@ -104,6 +107,11 @@ function getDisplayPublicKey(publicKey: string): string { export default function LangfuseConnection() { const localize = useLocalize(); const { showToast } = useToastContext(); + const { data: startupConfig } = useGetStartupConfig(); + // The server's `POST /prompts/sync` route requires `manage:prompts` in addition to the + // `langfuse` config capability that gates this whole tab, so an admin can see the + // connection without being able to run the sync. + const canManagePrompts = startupConfig?.langfusePromptManageAccess === true; const { data: status, isLoading: isConnectionLoading, @@ -114,6 +122,7 @@ export default function LangfuseConnection() { const updateMutation = useUpdateLangfuseConnectionMutation(); const testMutation = useTestLangfuseConnectionMutation(); const promptSyncMutation = useUpdateLangfusePromptSyncMutation(); + const syncPromptsMutation = useSyncLangfusePromptsMutation(); const [connectionStatus, setConnectionStatus] = useState(); const [destination, setDestination] = useState(''); @@ -197,7 +206,7 @@ export default function LangfuseConnection() { trimmedPublicKey !== '' && ((!connectionCredentialsChanged && secretConfigured) || trimmedSecretKey !== ''); const isSavingStatus = updateMutation.isLoading || promptSyncMutation.isLoading; - const busy = testMutation.isLoading || isSavingStatus; + const busy = testMutation.isLoading || isSavingStatus || syncPromptsMutation.isLoading; useEffect(() => { const storedConnectionTestKey = getStoredConnectionTestKey(connectionStatus); @@ -391,6 +400,35 @@ export default function LangfuseConnection() { ); }; + const handleSyncPrompts = () => { + syncPromptsMutation.mutate(undefined, { + onSuccess: (result) => { + if (result.complete === false) { + showToast({ + message: localize('com_ui_langfuse_prompt_sync_incomplete', { + 0: result.created, + 1: result.unchanged, + 2: result.skipped, + }), + status: 'warning', + }); + return; + } + showToast({ + message: localize('com_ui_langfuse_prompt_sync_result', { + 0: result.created, + 1: result.unchanged, + 2: result.skipped, + }), + status: 'success', + }); + }, + onError: (error) => { + showToast({ message: getPromptErrorMessage(error, localize), status: 'error' }); + }, + }); + }; + if (isConnectionLoading && connectionStatus == null) { return (
@@ -604,8 +642,25 @@ export default function LangfuseConnection() {

)} {connectionStatus.promptSync.enabled && ( -
- + )} +
diff --git a/client/src/components/Nav/SettingsTabs/Integrations/LangfusePromptsDialog.tsx b/client/src/components/Nav/SettingsTabs/Integrations/LangfusePromptsDialog.tsx index 8f1862c91bb..e54fcd5d8fb 100644 --- a/client/src/components/Nav/SettingsTabs/Integrations/LangfusePromptsDialog.tsx +++ b/client/src/components/Nav/SettingsTabs/Integrations/LangfusePromptsDialog.tsx @@ -1,5 +1,4 @@ import { useId, useState } from 'react'; -import axios from 'axios'; import { Button, Input, @@ -18,16 +17,10 @@ import { TableRow, Textarea, } from '@librechat/client'; -import type { - TLangfusePromptErrorBody, - TLangfusePromptErrorCode, - TLangfusePromptGetParams, - TLangfusePromptListParams, -} from 'librechat-data-provider'; +import type { TLangfusePromptGetParams, TLangfusePromptListParams } from 'librechat-data-provider'; import type { FormEvent } from 'react'; -import type { TranslationKeys } from '~/hooks'; import { useLangfusePromptQuery, useLangfusePromptsQuery } from '~/data-provider'; -import { getResponseErrorCode } from '~/utils'; +import { getPromptErrorMessage } from './promptErrors'; import { useLocalize } from '~/hooks'; type LangfusePromptsDialogProps = { @@ -45,40 +38,6 @@ function isGetMode(value: string): value is GetMode { return value === 'production' || value === 'version'; } -/** Codes whose whole copy is one localized sentence; `upstream` is handled separately below - * because its sentence needs the Langfuse HTTP status when the server sent one. */ -const PROMPT_ERROR_MESSAGE_KEYS: Partial> = { - not_configured: 'com_ui_langfuse_prompt_error_not_configured', - unauthorized: 'com_ui_langfuse_prompt_error_unauthorized', - timeout: 'com_ui_langfuse_prompt_error_timeout', - invalid_response: 'com_ui_langfuse_prompt_error_invalid_response', - not_found: 'com_ui_langfuse_prompt_error_not_found', - unsupported_type: 'com_ui_langfuse_prompt_error_unsupported_type', - invalid_request: 'com_ui_langfuse_prompt_error_invalid_request', -}; - -function getUpstreamStatus(error: unknown): number | undefined { - if (!axios.isAxiosError(error)) { - return undefined; - } - const status = error.response?.data?.status; - return typeof status === 'number' ? status : undefined; -} - -/** Maps a failed list or get call to one short localized line, with a generic fallback for an - * unmapped or missing code (for example a bare 404 with no body, which means the feature is off). */ -function getPromptErrorMessage(error: unknown, localize: ReturnType): string { - const code = getResponseErrorCode(error); - if (code === 'upstream') { - const status = getUpstreamStatus(error); - return status != null - ? localize('com_ui_langfuse_prompt_error_upstream_status', { 0: status }) - : localize('com_ui_langfuse_prompt_error_upstream'); - } - const messageKey = code != null ? PROMPT_ERROR_MESSAGE_KEYS[code] : undefined; - return localize(messageKey ?? 'com_ui_langfuse_prompt_error_fallback'); -} - /** A whole, positive integer, the way page, limit and version numbers must read. Rejects * fractional input (`"1.5"`) and blanks instead of silently rounding or defaulting. */ function parsePositiveInt(value: string): number | undefined { diff --git a/client/src/components/Nav/SettingsTabs/Integrations/__tests__/LangfuseConnection.spec.tsx b/client/src/components/Nav/SettingsTabs/Integrations/__tests__/LangfuseConnection.spec.tsx index b1759c4902b..9c6550e0de2 100644 --- a/client/src/components/Nav/SettingsTabs/Integrations/__tests__/LangfuseConnection.spec.tsx +++ b/client/src/components/Nav/SettingsTabs/Integrations/__tests__/LangfuseConnection.spec.tsx @@ -3,44 +3,59 @@ import { act, render, screen, fireEvent, waitFor } from '@testing-library/react' import LangfuseConnection from '../LangfuseConnection'; const mockGet = jest.fn(); +const mockGetStartupConfig = jest.fn(); const mockUpdate = jest.fn(); const mockTest = jest.fn(); const mockUpdatePromptSync = jest.fn(); +const mockSyncPrompts = jest.fn(); const mockRefetch = jest.fn(); +const mockShowToast = jest.fn(); // Mutable so a test can simulate an in-flight mutation; `jest.mock` factories may only // reference identifiers whose name starts with `mock`. let mockUpdateIsLoading = false; let mockPromptSyncIsLoading = false; +let mockSyncIsLoading = false; +let mockTestIsLoading = false; const destinationLabels = { eu: 'eu - https://cloud.langfuse.com', us: 'us - https://us.cloud.langfuse.com', }; +function axiosError(outerStatus: number, data: { code: string; status?: number }) { + return { isAxiosError: true, response: { status: outerStatus, data } }; +} + async function selectDestination(destination: keyof typeof destinationLabels) { await userEvent.click(screen.getByTestId('langfuse-destination')); await userEvent.click(screen.getByRole('option', { name: destinationLabels[destination] })); } jest.mock('~/data-provider', () => ({ + useGetStartupConfig: () => mockGetStartupConfig(), useGetLangfuseConnectionQuery: () => mockGet(), useUpdateLangfuseConnectionMutation: () => ({ mutate: mockUpdate, isLoading: mockUpdateIsLoading, }), - useTestLangfuseConnectionMutation: () => ({ mutate: mockTest, isLoading: false }), + useTestLangfuseConnectionMutation: () => ({ mutate: mockTest, isLoading: mockTestIsLoading }), useUpdateLangfusePromptSyncMutation: () => ({ mutate: mockUpdatePromptSync, isLoading: mockPromptSyncIsLoading, }), + useSyncLangfusePromptsMutation: () => ({ + mutate: mockSyncPrompts, + isLoading: mockSyncIsLoading, + }), })); jest.mock('~/hooks', () => ({ - useLocalize: () => (key: string) => key, + useLocalize: () => (key: string, vars?: Record) => + vars ? `${key} ${JSON.stringify(vars)}` : key, })); jest.mock('@librechat/client', () => ({ ...jest.requireActual('@librechat/client'), - useToastContext: () => ({ showToast: jest.fn() }), + useToastContext: () => ({ showToast: mockShowToast }), })); // Covered on its own in LangfusePromptsDialog.spec.tsx; stubbed here so these tests do not @@ -59,16 +74,24 @@ beforeEach(() => { disconnect = jest.fn(); }; mockGet.mockReset(); + mockGetStartupConfig.mockReset(); mockUpdate.mockReset(); mockTest.mockReset(); mockUpdatePromptSync.mockReset(); + mockSyncPrompts.mockReset(); mockRefetch.mockReset(); + mockShowToast.mockReset(); mockPromptsDialog.mockClear(); mockUpdateIsLoading = false; mockPromptSyncIsLoading = false; + mockSyncIsLoading = false; + mockTestIsLoading = false; mockTest.mockImplementation((_payload, options) => { options?.onSuccess?.({ success: true }); }); + // Defaults to access so the many pre-existing tests that exercise the Sync button do not + // each have to opt in; the dedicated capability tests below set this to false. + mockGetStartupConfig.mockReturnValue({ data: { langfusePromptManageAccess: true } }); mockGet.mockReturnValue({ isLoading: false, isError: false, @@ -874,6 +897,228 @@ describe('LangfuseConnection', () => { expect(screen.getByRole('button', { name: 'com_ui_langfuse_enable' })).toBeDisabled(); }); + + function withPromptSyncEnabled() { + mockGet.mockReturnValue({ + isLoading: false, + isError: false, + isFetching: false, + refetch: mockRefetch, + data: { + configured: false, + enabled: false, + destinations: [], + promptSync: { available: true, enabled: true }, + }, + }); + } + + it('hides the Sync prompts button when prompt sync is disabled', () => { + mockGet.mockReturnValue({ + isLoading: false, + isError: false, + isFetching: false, + refetch: mockRefetch, + data: { + configured: false, + enabled: false, + destinations: [], + promptSync: { available: true, enabled: false }, + }, + }); + + render(); + + expect( + screen.queryByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' }), + ).not.toBeInTheDocument(); + }); + + it('shows the Sync prompts button when prompt sync is enabled', () => { + withPromptSyncEnabled(); + + render(); + + expect( + screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' }), + ).toBeInTheDocument(); + }); + + it('hides the Sync prompts button for an admin without manage:prompts access', () => { + mockGetStartupConfig.mockReturnValue({ data: { langfusePromptManageAccess: false } }); + withPromptSyncEnabled(); + + render(); + + expect( + screen.queryByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' }), + ).not.toBeInTheDocument(); + // The route behind this button does not require manage:prompts, so it stays available. + expect( + screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_open' }), + ).toBeInTheDocument(); + }); + + it('calls the sync mutation when the Sync prompts button is clicked', async () => { + withPromptSyncEnabled(); + + render(); + await userEvent.click(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' })); + + expect(mockSyncPrompts).toHaveBeenCalledWith(undefined, expect.any(Object)); + }); + + it('shows a success toast with the three counts after a sync completes', async () => { + mockSyncPrompts.mockImplementation((_payload, options) => { + options?.onSuccess?.({ created: 1, unchanged: 2, skipped: 3 }); + }); + withPromptSyncEnabled(); + + render(); + await userEvent.click(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' })); + + expect(mockShowToast).toHaveBeenCalledWith({ + message: 'com_ui_langfuse_prompt_sync_result {"0":1,"1":2,"2":3}', + status: 'success', + }); + }); + + it('shows a warning toast, not the success toast, when the sync stops at the page cap', async () => { + mockSyncPrompts.mockImplementation((_payload, options) => { + options?.onSuccess?.({ created: 1, unchanged: 2, skipped: 3, complete: false }); + }); + withPromptSyncEnabled(); + + render(); + await userEvent.click(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' })); + + expect(mockShowToast).toHaveBeenCalledWith({ + message: 'com_ui_langfuse_prompt_sync_incomplete {"0":1,"1":2,"2":3}', + status: 'warning', + }); + expect(mockShowToast).not.toHaveBeenCalledWith(expect.objectContaining({ status: 'success' })); + }); + + it('shows a mapped error toast for an already-running sync', async () => { + mockSyncPrompts.mockImplementation((_payload, options) => { + options?.onError?.(axiosError(409, { code: 'already_running' })); + }); + withPromptSyncEnabled(); + + render(); + await userEvent.click(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' })); + + expect(mockShowToast).toHaveBeenCalledWith({ + message: 'com_ui_langfuse_prompt_error_already_running', + status: 'error', + }); + }); + + it('falls back to a generic error toast for an unmapped sync error code', async () => { + mockSyncPrompts.mockImplementation((_payload, options) => { + options?.onError?.(axiosError(500, { code: 'something_unexpected' })); + }); + withPromptSyncEnabled(); + + render(); + await userEvent.click(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' })); + + expect(mockShowToast).toHaveBeenCalledWith({ + message: 'com_ui_langfuse_prompt_error_fallback', + status: 'error', + }); + }); + + it('disables the Sync prompts button and the switch while a sync is in flight, announcing the loading state', () => { + withPromptSyncEnabled(); + mockSyncIsLoading = true; + + render(); + + // The accessible name tracks the visible label (no static aria-label pins it to the + // idle text), so assistive tech hears the loading state too. + expect( + screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_running' }), + ).toBeDisabled(); + expect( + screen.queryByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' }), + ).not.toBeInTheDocument(); + expect(screen.getByRole('switch')).toBeDisabled(); + }); + + it('disables the Save button and the destination and key inputs while a sync is in flight', async () => { + const { rerender } = render(); + await selectDestination('us'); + fireEvent.change(screen.getByLabelText('com_ui_langfuse_public_key'), { + target: { value: 'pk-lf-1' }, + }); + fireEvent.change(screen.getByLabelText(/com_ui_langfuse_secret_key/), { + target: { value: 'sk-lf-secret' }, + }); + expect(screen.getByRole('button', { name: 'com_ui_save' })).toBeEnabled(); + + mockSyncIsLoading = true; + rerender(); + + expect(screen.getByRole('button', { name: 'com_ui_save' })).toBeDisabled(); + expect(screen.getByTestId('langfuse-destination')).toBeDisabled(); + expect(screen.getByLabelText('com_ui_langfuse_public_key')).toBeDisabled(); + expect(screen.getByLabelText(/com_ui_langfuse_secret_key/)).toBeDisabled(); + }); + + it('disables the Sync prompts and Preview prompts buttons while the connection test is running', () => { + withPromptSyncEnabled(); + mockTestIsLoading = true; + + render(); + + expect(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' })).toBeDisabled(); + expect(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_open' })).toBeDisabled(); + }); + + it('disables the Sync prompts and Preview prompts buttons while a connection save is in flight', () => { + withPromptSyncEnabled(); + mockUpdateIsLoading = true; + + render(); + + expect(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' })).toBeDisabled(); + expect(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_open' })).toBeDisabled(); + }); + + it('disables the Sync prompts and Preview prompts buttons while a prompt-sync setting save is in flight', () => { + withPromptSyncEnabled(); + mockPromptSyncIsLoading = true; + + render(); + + expect(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' })).toBeDisabled(); + expect(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_open' })).toBeDisabled(); + }); + + it('disables the Preview prompts button while a sync is running', () => { + withPromptSyncEnabled(); + mockSyncIsLoading = true; + + render(); + + expect(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_open' })).toBeDisabled(); + }); + + it('shows the forbidden copy in the toast when the sync mutation rejects with a 403', async () => { + mockSyncPrompts.mockImplementation((_payload, options) => { + options?.onError?.(axiosError(403, { code: 'unauthorized' })); + }); + withPromptSyncEnabled(); + + render(); + await userEvent.click(screen.getByRole('button', { name: 'com_ui_langfuse_prompt_sync_run' })); + + expect(mockShowToast).toHaveBeenCalledWith({ + message: 'com_ui_langfuse_prompt_error_forbidden', + status: 'error', + }); + }); }); describe('prompt-only Langfuse settings', () => { diff --git a/client/src/components/Nav/SettingsTabs/Integrations/__tests__/promptErrors.spec.ts b/client/src/components/Nav/SettingsTabs/Integrations/__tests__/promptErrors.spec.ts new file mode 100644 index 00000000000..4b2262bcbc4 --- /dev/null +++ b/client/src/components/Nav/SettingsTabs/Integrations/__tests__/promptErrors.spec.ts @@ -0,0 +1,58 @@ +import { getPromptErrorMessage } from '../promptErrors'; + +/** An axios-style rejection the way the real backend sends one: an outer HTTP status plus a + * `{ code, status? }` body. */ +function axiosError(outerStatus: number, data: { code?: string; status?: number } = {}) { + return { isAxiosError: true, response: { status: outerStatus, data } }; +} + +const localize = (key: string, options?: Record) => + options ? `${key}:${Object.values(options).join(',')}` : key; + +describe('getPromptErrorMessage', () => { + it('returns the forbidden copy for a 403, ahead of any body code', () => { + const error = axiosError(403, { message: 'Forbidden' } as { code?: string }); + + expect(getPromptErrorMessage(error, localize)).toBe('com_ui_langfuse_prompt_error_forbidden'); + }); + + it('returns the forbidden copy for a 403 even when the body carries an unrelated code', () => { + const error = axiosError(403, { code: 'unauthorized' }); + + expect(getPromptErrorMessage(error, localize)).toBe('com_ui_langfuse_prompt_error_forbidden'); + }); + + it('maps the internal code to its localized copy', () => { + const error = axiosError(500, { code: 'internal' }); + + expect(getPromptErrorMessage(error, localize)).toBe('com_ui_langfuse_prompt_error_internal'); + }); + + it('maps a known code to its localized copy', () => { + const error = axiosError(409, { code: 'already_running' }); + + expect(getPromptErrorMessage(error, localize)).toBe( + 'com_ui_langfuse_prompt_error_already_running', + ); + }); + + it('includes the Langfuse HTTP status for an upstream failure', () => { + const error = axiosError(502, { code: 'upstream', status: 400 }); + + expect(getPromptErrorMessage(error, localize)).toBe( + 'com_ui_langfuse_prompt_error_upstream_status:400', + ); + }); + + it('falls back to the generic copy for an unmapped code', () => { + const error = axiosError(500, { code: 'something_unexpected' }); + + expect(getPromptErrorMessage(error, localize)).toBe('com_ui_langfuse_prompt_error_fallback'); + }); + + it('falls back to the generic copy for a non-axios error', () => { + expect(getPromptErrorMessage(new Error('boom'), localize)).toBe( + 'com_ui_langfuse_prompt_error_fallback', + ); + }); +}); diff --git a/client/src/components/Nav/SettingsTabs/Integrations/promptErrors.ts b/client/src/components/Nav/SettingsTabs/Integrations/promptErrors.ts new file mode 100644 index 00000000000..18754cfde26 --- /dev/null +++ b/client/src/components/Nav/SettingsTabs/Integrations/promptErrors.ts @@ -0,0 +1,50 @@ +import axios from 'axios'; +import type { TLangfusePromptErrorBody, TLangfusePromptErrorCode } from 'librechat-data-provider'; +import type { TranslationKeys, useLocalize } from '~/hooks'; +import { getResponseErrorCode, getResponseStatus } from '~/utils'; + +/** Codes whose whole copy is one localized sentence; `upstream` is handled separately below + * because its sentence needs the Langfuse HTTP status when the server sent one. */ +export const PROMPT_ERROR_MESSAGE_KEYS: Partial> = + { + not_configured: 'com_ui_langfuse_prompt_error_not_configured', + unauthorized: 'com_ui_langfuse_prompt_error_unauthorized', + timeout: 'com_ui_langfuse_prompt_error_timeout', + invalid_response: 'com_ui_langfuse_prompt_error_invalid_response', + not_found: 'com_ui_langfuse_prompt_error_not_found', + unsupported_type: 'com_ui_langfuse_prompt_error_unsupported_type', + invalid_request: 'com_ui_langfuse_prompt_error_invalid_request', + already_running: 'com_ui_langfuse_prompt_error_already_running', + lock_lost: 'com_ui_langfuse_prompt_error_lock_lost', + internal: 'com_ui_langfuse_prompt_error_internal', + }; + +function getUpstreamStatus(error: unknown): number | undefined { + if (!axios.isAxiosError(error)) { + return undefined; + } + const status = error.response?.data?.status; + return typeof status === 'number' ? status : undefined; +} + +/** Maps a failed Langfuse prompt call to one short localized line, with a generic fallback for + * an unmapped or missing code (for example a bare 404 with no body, which means the feature is off). + * A 403 is checked ahead of any body code: it means the caller lacks permission regardless of + * what the body says, so it always wins. */ +export function getPromptErrorMessage( + error: unknown, + localize: ReturnType, +): string { + if (getResponseStatus(error) === 403) { + return localize('com_ui_langfuse_prompt_error_forbidden'); + } + const code = getResponseErrorCode(error); + if (code === 'upstream') { + const status = getUpstreamStatus(error); + return status != null + ? localize('com_ui_langfuse_prompt_error_upstream_status', { 0: status }) + : localize('com_ui_langfuse_prompt_error_upstream'); + } + const messageKey = code != null ? PROMPT_ERROR_MESSAGE_KEYS[code] : undefined; + return localize(messageKey ?? 'com_ui_langfuse_prompt_error_fallback'); +} diff --git a/client/src/data-provider/Langfuse/__tests__/sync.spec.tsx b/client/src/data-provider/Langfuse/__tests__/sync.spec.tsx new file mode 100644 index 00000000000..b3cafb67efd --- /dev/null +++ b/client/src/data-provider/Langfuse/__tests__/sync.spec.tsx @@ -0,0 +1,84 @@ +import { act, renderHook } from '@testing-library/react'; +import { QueryKeys, ResourceType } from 'librechat-data-provider'; +import { QueryClient, QueryClientProvider } from '@tanstack/react-query'; +import type { ReactNode } from 'react'; +import { useSyncLangfusePromptsMutation } from '../index'; + +const mockSyncLangfusePrompts = jest.fn(); + +jest.mock('librechat-data-provider', () => { + const actual = jest.requireActual('librechat-data-provider'); + return { + ...actual, + dataService: { + ...actual.dataService, + syncLangfusePrompts: (...args: unknown[]) => mockSyncLangfusePrompts(...args), + }, + }; +}); + +describe('useSyncLangfusePromptsMutation', () => { + let queryClient: QueryClient; + let wrapper: ({ children }: { children: ReactNode }) => JSX.Element; + let invalidateSpy: jest.SpyInstance; + + beforeEach(() => { + jest.resetAllMocks(); + queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false }, mutations: { retry: false } }, + }); + invalidateSpy = jest.spyOn(queryClient, 'invalidateQueries'); + wrapper = ({ children }: { children: ReactNode }) => ( + {children} + ); + }); + + it('invalidates prompt groups, all prompt groups, and prompt-group effective permissions on success', async () => { + mockSyncLangfusePrompts.mockResolvedValueOnce({ created: 1, unchanged: 2, skipped: 0 }); + const { result } = renderHook(() => useSyncLangfusePromptsMutation(), { wrapper }); + + await act(async () => { + await result.current.mutateAsync(); + }); + + expect(invalidateSpy).toHaveBeenCalledWith([QueryKeys.promptGroups]); + expect(invalidateSpy).toHaveBeenCalledWith([QueryKeys.allPromptGroups]); + expect(invalidateSpy).toHaveBeenCalledWith([ + QueryKeys.effectivePermissions, + 'all', + ResourceType.PROMPTGROUP, + ]); + }); + + it('invalidates prompt groups, all prompt groups, and prompt-group effective permissions when the sync call fails', async () => { + // An earlier page's groups can already be committed when a later page throws, so a + // failed sync still invalidates the catalog's queries, not only a successful one. + mockSyncLangfusePrompts.mockRejectedValueOnce(new Error('upstream')); + const { result } = renderHook(() => useSyncLangfusePromptsMutation(), { wrapper }); + + await act(async () => { + await result.current.mutateAsync().catch(() => undefined); + }); + + expect(invalidateSpy).toHaveBeenCalledWith([QueryKeys.promptGroups]); + expect(invalidateSpy).toHaveBeenCalledWith([QueryKeys.allPromptGroups]); + expect(invalidateSpy).toHaveBeenCalledWith([ + QueryKeys.effectivePermissions, + 'all', + ResourceType.PROMPTGROUP, + ]); + }); + + it('still calls a caller-supplied onError when the sync call fails', async () => { + mockSyncLangfusePrompts.mockRejectedValueOnce(new Error('upstream')); + const { result } = renderHook(() => useSyncLangfusePromptsMutation(), { wrapper }); + const onError = jest.fn(); + + await act(async () => { + await result.current.mutateAsync(undefined, { onError }).catch(() => undefined); + }); + + expect(onError).toHaveBeenCalledTimes(1); + expect(invalidateSpy).toHaveBeenCalledWith([QueryKeys.promptGroups]); + }); +}); diff --git a/client/src/data-provider/Langfuse/index.ts b/client/src/data-provider/Langfuse/index.ts index 47dc3f38225..178ef100755 100644 --- a/client/src/data-provider/Langfuse/index.ts +++ b/client/src/data-provider/Langfuse/index.ts @@ -1,5 +1,5 @@ import { useQuery, useMutation, useQueryClient } from '@tanstack/react-query'; -import { dataService, QueryKeys, MutationKeys } from 'librechat-data-provider'; +import { dataService, QueryKeys, MutationKeys, ResourceType } from 'librechat-data-provider'; import type { TLangfuseConnectionStatus, TUpdateLangfuseConnectionRequest, @@ -11,6 +11,7 @@ import type { TLangfusePromptListResponse, TLangfusePromptGetParams, TLangfusePromptGetResponse, + TLangfusePromptSyncResult, } from 'librechat-data-provider'; import type { UseQueryResult, UseMutationResult } from '@tanstack/react-query'; @@ -123,3 +124,30 @@ export const useLangfusePromptQuery = ( }, { enabled: params != null, retry: false, refetchOnWindowFocus: false }, ); + +/** + * Runs the real Langfuse prompt sync. The catalog's own prompt-group and permissions queries + * go stale whether the call settles in success or failure: a synced group is a new + * PromptGroup with a PUBLIC viewer grant, and a page that throws partway through still + * leaves any earlier page's groups committed, so both the list views and the sidebar's + * effective-permissions map must refetch on every outcome, not only on success. + */ +export const useSyncLangfusePromptsMutation = (): UseMutationResult< + TLangfusePromptSyncResult, + unknown, + void +> => { + const queryClient = useQueryClient(); + return useMutation(() => dataService.syncLangfusePrompts(), { + mutationKey: [MutationKeys.syncLangfusePrompts], + onSettled: () => { + queryClient.invalidateQueries([QueryKeys.promptGroups]); + queryClient.invalidateQueries([QueryKeys.allPromptGroups]); + queryClient.invalidateQueries([ + QueryKeys.effectivePermissions, + 'all', + ResourceType.PROMPTGROUP, + ]); + }, + }); +}; diff --git a/client/src/locales/en/translation.json b/client/src/locales/en/translation.json index e3e2fe560c7..ed1a12d90f0 100644 --- a/client/src/locales/en/translation.json +++ b/client/src/locales/en/translation.json @@ -1940,9 +1940,13 @@ "com_ui_langfuse_disable": "Disable", "com_ui_langfuse_enable": "Enable", "com_ui_langfuse_load_error": "Failed to load the Langfuse connection", + "com_ui_langfuse_prompt_error_already_running": "A prompt sync is already running.", "com_ui_langfuse_prompt_error_fallback": "Something went wrong. Try again.", + "com_ui_langfuse_prompt_error_forbidden": "You do not have permission to manage prompts.", + "com_ui_langfuse_prompt_error_internal": "The prompt sync failed on the server. Check the server logs and try again.", "com_ui_langfuse_prompt_error_invalid_request": "Check the values and try again.", "com_ui_langfuse_prompt_error_invalid_response": "Langfuse returned an unexpected response.", + "com_ui_langfuse_prompt_error_lock_lost": "The sync stopped early. Run it again.", "com_ui_langfuse_prompt_error_not_configured": "Connect a Langfuse project first.", "com_ui_langfuse_prompt_error_not_found": "No prompt with this name and version.", "com_ui_langfuse_prompt_error_timeout": "Langfuse did not respond in time.", @@ -1951,15 +1955,19 @@ "com_ui_langfuse_prompt_error_upstream": "Could not reach Langfuse.", "com_ui_langfuse_prompt_error_upstream_status": "Langfuse returned HTTP {{0}}.", "com_ui_langfuse_prompt_sync": "Langfuse prompts", - "com_ui_langfuse_prompt_sync_description": "Try listing and resolving prompts from the connected Langfuse project. Nothing is written.", + "com_ui_langfuse_prompt_sync_description": "Adds new text prompts from the connected Langfuse project to the catalog as read-only library entries. Never removes or changes an existing entry.", "com_ui_langfuse_prompt_sync_error": "Failed to update the Langfuse prompts setting", - "com_ui_langfuse_prompt_sync_open": "Sync prompts (test)", + "com_ui_langfuse_prompt_sync_incomplete": "Synced {{0}} new, {{1}} already in the catalog, {{2}} skipped. The sync stopped at the page limit, so some prompts were not imported. Increase the prompt sync page limit and run the sync again.", + "com_ui_langfuse_prompt_sync_open": "Preview prompts (dry run)", + "com_ui_langfuse_prompt_sync_result": "Synced {{0}} new, {{1}} already in the catalog, {{2}} skipped", + "com_ui_langfuse_prompt_sync_run": "Sync prompts", + "com_ui_langfuse_prompt_sync_running": "Syncing…", "com_ui_langfuse_prompts_column_labels": "Labels", "com_ui_langfuse_prompts_column_name": "Name", "com_ui_langfuse_prompts_column_tags": "Tags", "com_ui_langfuse_prompts_column_type": "Type", "com_ui_langfuse_prompts_column_versions": "Versions", - "com_ui_langfuse_prompts_dialog_title": "Sync prompts (test)", + "com_ui_langfuse_prompts_dialog_title": "Preview prompts (dry run)", "com_ui_langfuse_prompts_dry_run_notice": "This is a dry run. Nothing is written.", "com_ui_langfuse_prompts_get_button": "Get", "com_ui_langfuse_prompts_get_name": "Name", diff --git a/config/__tests__/delete-user.spec.js b/config/__tests__/delete-user.spec.js index 92cb7f04eb5..ec2b480c479 100644 --- a/config/__tests__/delete-user.spec.js +++ b/config/__tests__/delete-user.spec.js @@ -26,6 +26,7 @@ const mockMethods = { deleteAgentTriggerDeliveriesByUser: jest.fn(), cancelAgentTriggerUserPurge: jest.fn(), cancelAgentTriggerUserDeletion: jest.fn(), + deleteUserPrompts: jest.fn(), }; const mockGetCleanupBlockingJobIdsForUser = jest.fn(); const mockAbortJob = jest.fn(); @@ -129,6 +130,7 @@ describe('Delete user CLI', () => { mockMethods.deleteAgentTriggerDeliveriesByUser.mockReset().mockResolvedValue(undefined); mockMethods.cancelAgentTriggerUserPurge.mockReset().mockResolvedValue(true); mockMethods.cancelAgentTriggerUserDeletion.mockReset().mockResolvedValue(true); + mockMethods.deleteUserPrompts.mockReset().mockResolvedValue(undefined); mockGetCleanupBlockingJobIdsForUser.mockReset().mockResolvedValue([]); mockAbortJob.mockReset().mockResolvedValue({ success: true }); mockDestroy.mockReset().mockResolvedValue(undefined); @@ -389,6 +391,13 @@ describe('Delete user CLI', () => { expect(mockMethods.cancelAgentTriggerUserDeletion).toHaveBeenCalledWith(USER_ID, deletionFence); }); + it('passes deleteUserPrompts through to the shared cascade, which keeps a synced group with a synthetic author untouched', async () => { + expect(await runCli()).toBe(0); + + const [, deps] = mockEraseUserData.mock.calls[0]; + expect(deps.deleteUserPrompts).toBe(mockMethods.deleteUserPrompts); + }); + it('deletes nothing and releases both fences when provider drain fails', async () => { mockGetCleanupBlockingJobIdsForUser.mockResolvedValueOnce(['stream-1']); mockAbortJob.mockRejectedValueOnce(new Error('provider drain failed')); diff --git a/config/__tests__/migrate-prompt-permissions.spec.js b/config/__tests__/migrate-prompt-permissions.spec.js index 2d5b2cb4b0a..a0271f79461 100644 --- a/config/__tests__/migrate-prompt-permissions.spec.js +++ b/config/__tests__/migrate-prompt-permissions.spec.js @@ -276,4 +276,66 @@ describe('PromptGroup Migration Script', () => { }); expect(promptEntries).toHaveLength(0); }); + + it('should skip a Langfuse-source promptGroup in dry run', async () => { + await PromptGroup.create({ + name: 'Native Group', + author: testOwner._id, + authorName: testOwner.name, + productionId: new ObjectId(), + }); + + await PromptGroup.create({ + name: 'Langfuse Group', + author: testOwner._id, + authorName: 'Langfuse Sync', + source: 'langfuse', + sourcePromptName: 'langfuse-prompt', + sourceProjectId: 'project-1', + sourceDestination: 'destination-1', + }); + + const result = await migrateToPromptGroupPermissions({ dryRun: true }); + + expect(result.summary.total).toBe(1); + expect(result.details.privateGroups).toHaveLength(1); + expect(result.details.privateGroups[0].name).toBe('Native Group'); + }); + + it('should skip a Langfuse-source promptGroup during a live run and write no owner grant for it', async () => { + const nativeGroup = await PromptGroup.create({ + name: 'Native Group', + author: testOwner._id, + authorName: testOwner.name, + productionId: new ObjectId(), + }); + + const langfuseGroup = await PromptGroup.create({ + name: 'Langfuse Group', + author: testOwner._id, + authorName: 'Langfuse Sync', + source: 'langfuse', + sourcePromptName: 'langfuse-prompt', + sourceProjectId: 'project-1', + sourceDestination: 'destination-1', + }); + + const result = await migrateToPromptGroupPermissions({ dryRun: false }); + + expect(result.migrated).toBe(1); + expect(result.ownerGrants).toBe(1); + + const nativeOwnerEntry = await AclEntry.findOne({ + resourceType: ResourceType.PROMPTGROUP, + resourceId: nativeGroup._id, + principalType: PrincipalType.USER, + }); + expect(nativeOwnerEntry).toBeTruthy(); + + const langfuseOwnerEntry = await AclEntry.findOne({ + resourceType: ResourceType.PROMPTGROUP, + resourceId: langfuseGroup._id, + }); + expect(langfuseOwnerEntry).toBeNull(); + }); }); diff --git a/config/migrate-prompt-permissions.js b/config/migrate-prompt-permissions.js index 1127c44cebe..b9ed28746f5 100644 --- a/config/migrate-prompt-permissions.js +++ b/config/migrate-prompt-permissions.js @@ -52,6 +52,7 @@ async function migrateToPromptGroupPermissions({ dryRun = true, batchSize = 100 const promptGroupsToMigrate = await PromptGroup.find({ _id: { $nin: migratedGroupIds }, author: { $exists: true, $ne: null }, + source: { $in: [null, 'native'] }, }) .select('_id name author authorName category') .lean(); diff --git a/packages/api/src/admin/index.ts b/packages/api/src/admin/index.ts index 19e2c40611b..00279a672db 100644 --- a/packages/api/src/admin/index.ts +++ b/packages/api/src/admin/index.ts @@ -21,7 +21,7 @@ export { resolveConfigSecret, redactConfigSecretMaps } from './secrets'; export type { AdminConfigDeps } from './config'; export type { AdminConfigReadDeps } from './resolved'; export type { AdminConfigCatalogDeps, CatalogScopeRequest, CatalogScopeUser } from './catalog'; -export type { AdminLangfuseDeps } from './langfuse'; +export type { AdminLangfuseDeps, LangfuseAdminConfig } from './langfuse'; export type { AdminGrantsDeps, GrantPrincipalType } from './grants'; export type { AdminGroupsDeps } from './groups'; export type { diff --git a/packages/api/src/admin/langfuse.handler.spec.ts b/packages/api/src/admin/langfuse.handler.spec.ts index 69265096f8a..231ae9357fd 100644 --- a/packages/api/src/admin/langfuse.handler.spec.ts +++ b/packages/api/src/admin/langfuse.handler.spec.ts @@ -114,6 +114,25 @@ function promptListBody() { }; } +function syncPromptItem(name: string, overrides: Record = {}) { + return { + name, + type: 'text', + versions: [1], + labels: ['production'], + tags: [], + lastUpdatedAt: '2026-01-01T00:00:00.000Z', + ...overrides, + }; +} + +function syncListBody( + items: Array>, + meta: { page: number; limit: number; totalItems: number; totalPages: number }, +) { + return { data: items, meta }; +} + function mockReq(overrides = {}) { return { user: { id: 'u1', role: 'ADMIN', tenantId: 't1' }, @@ -174,6 +193,9 @@ function createHandlers(overrides = {}) { }), ), getMessages: jest.fn().mockResolvedValue([]), + syncPromptCatalog: jest + .fn() + .mockResolvedValue({ ok: true, value: { created: 0, unchanged: 0, skipped: 0 } }), invalidateConfigCaches: jest.fn().mockResolvedValue(undefined), recordConnectionUpdate: jest.fn(), ...overrides, @@ -1846,7 +1868,7 @@ describe('createAdminLangfuseHandlers', () => { await handlers.listPrompts(mockReq({ config: { langfuse: withPromptSync(true) } }), res); expect(res.statusCode).toBe(500); - expect(res.body).toEqual({ code: 'upstream' }); + expect(res.body).toEqual({ code: 'internal' }); expect(JSON.stringify(res.body)).not.toContain('MARKER_SECRET_LEAK_TOKEN'); }); @@ -2352,4 +2374,305 @@ describe('createAdminLangfuseHandlers', () => { ).toBe('env-public:env-secret'); }); }); + + describe('syncPrompts', () => { + function tenantSyncConnection(overrides: Record = {}) { + return withPromptSync(true, { projectId: 'project-1', ...overrides }); + } + + function createSyncHandlers(overrides: Record = {}) { + return createHandlers({ + findConfigByPrincipal: jest.fn().mockResolvedValue(baseConfigDoc(tenantSyncConnection())), + ...overrides, + }); + } + + function syncReq(overrides: Record = {}) { + return mockReq({ config: { langfuse: tenantSyncConnection() }, ...overrides }); + } + + /** The runner invocation the handler made; the handler calls it once per + * request that passes every gate. */ + function runnerRequest(deps: { syncPromptCatalog: jest.Mock }) { + expect(deps.syncPromptCatalog).toHaveBeenCalledTimes(1); + return deps.syncPromptCatalog.mock.calls[0][0] as { + source: { destination: string; projectId: string; tenantId?: string }; + listPrompts: (query: { type: 'text'; limit: number; page: number }) => Promise; + lockLeaseMs: number; + maxPages: number; + }; + } + + it('returns 404 when prompt sync is not available at all', async () => { + const { handlers, deps } = createHandlers(); + const res = mockRes(); + + await handlers.syncPrompts(mockReq(), res); + + expect(res.statusCode).toBe(404); + expect(global.fetch).not.toHaveBeenCalled(); + expect(deps.syncPromptCatalog).not.toHaveBeenCalled(); + }); + + it('returns 404 when available but the tenant switch is off', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const { handlers, deps } = createHandlers({ + findConfigByPrincipal: jest + .fn() + .mockResolvedValue(baseConfigDoc({ promptSync: { enabled: false } })), + }); + const res = mockRes(); + + await handlers.syncPrompts(syncReq(), res); + + expect(res.statusCode).toBe(404); + expect(deps.syncPromptCatalog).not.toHaveBeenCalled(); + }); + + it('returns 409 not_configured when no connection resolves', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const { handlers, deps } = createHandlers({ + findConfigByPrincipal: jest + .fn() + .mockResolvedValue(baseConfigDoc({ promptSync: { enabled: true } })), + }); + const res = mockRes(); + + await handlers.syncPrompts( + mockReq({ config: { langfuse: { promptSync: { enabled: true } } } }), + res, + ); + + expect(res.statusCode).toBe(409); + expect(res.body).toEqual({ code: 'not_configured' }); + expect(deps.syncPromptCatalog).not.toHaveBeenCalled(); + }); + + it('returns 409 not_configured when the connection resolves but the source identity has no project id', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const { handlers, deps } = createHandlers({ + findConfigByPrincipal: jest.fn().mockResolvedValue(baseConfigDoc(withPromptSync(true))), + }); + const res = mockRes(); + + await handlers.syncPrompts(mockReq({ config: { langfuse: withPromptSync(true) } }), res); + + expect(res.statusCode).toBe(409); + expect(res.body).toEqual({ code: 'not_configured' }); + expect(global.fetch).not.toHaveBeenCalled(); + expect(deps.syncPromptCatalog).not.toHaveBeenCalled(); + }); + + it('runs the sync with the source identity, the default lease and page cap, and returns its counts', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const { handlers, deps } = createSyncHandlers({ + syncPromptCatalog: jest + .fn() + .mockResolvedValue({ ok: true, value: { created: 2, unchanged: 1, skipped: 0 } }), + }); + const res = mockRes(); + + await handlers.syncPrompts(syncReq(), res); + + expect(res.statusCode).toBe(200); + expect(res.body).toEqual({ created: 2, unchanged: 1, skipped: 0 }); + expect(deps.syncPromptCatalog).toHaveBeenCalledWith({ + source: { destination: 'eu', projectId: 'project-1', tenantId: 't1' }, + listPrompts: expect.any(Function), + lockLeaseMs: 300_000, + maxPages: 50, + }); + }); + + it("passes the runner's complete: false through unchanged when the page cap stopped the run", async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const { handlers } = createSyncHandlers({ + syncPromptCatalog: jest.fn().mockResolvedValue({ + ok: true, + value: { created: 2, unchanged: 1, skipped: 0, complete: false }, + }), + }); + const res = mockRes(); + + await handlers.syncPrompts(syncReq(), res); + + expect(res.statusCode).toBe(200); + expect(res.body).toEqual({ created: 2, unchanged: 1, skipped: 0, complete: false }); + }); + + it('forwards the configured lock lease and page cap from req.config', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const { handlers, deps } = createSyncHandlers(); + const res = mockRes(); + + await handlers.syncPrompts( + syncReq({ + config: { + langfuse: { + ...tenantSyncConnection(), + promptSync: { enabled: true, lockLeaseMs: 120_000, maxPages: 3 }, + }, + }, + }), + res, + ); + + expect(res.statusCode).toBe(200); + expect(runnerRequest(deps)).toMatchObject({ lockLeaseMs: 120_000, maxPages: 3 }); + }); + + it('hands the runner a page lister bound to the resolved connection, with the text filter', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const { handlers, deps } = createSyncHandlers(); + const res = mockRes(); + + await handlers.syncPrompts(syncReq(), res); + expect(global.fetch).not.toHaveBeenCalled(); + + global.fetch = jest.fn().mockResolvedValue( + fetchJsonResponse( + syncListBody([syncPromptItem('a')], { + page: 2, + limit: 100, + totalItems: 2, + totalPages: 2, + }), + ), + ) as unknown as typeof fetch; + const page = await runnerRequest(deps).listPrompts({ type: 'text', limit: 100, page: 2 }); + + expect(page).toMatchObject({ items: [expect.objectContaining({ name: 'a' })] }); + expect(global.fetch).toHaveBeenCalledTimes(1); + const [url] = (global.fetch as unknown as jest.Mock).mock.calls[0] as [string]; + expect(url).toContain('limit=100'); + expect(url).toContain('page=2'); + expect(decodeURIComponent(url)).toContain('"column":"type"'); + expect(decodeURIComponent(url)).toContain('"value":"text"'); + }); + + it('returns 409 already_running when the runner reports the lock is already held', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const { handlers } = createSyncHandlers({ + syncPromptCatalog: jest + .fn() + .mockResolvedValue({ ok: false, error: { code: 'already_running' } }), + }); + const res = mockRes(); + + await handlers.syncPrompts(syncReq(), res); + + expect(res.statusCode).toBe(409); + expect(res.body).toEqual({ code: 'already_running' }); + }); + + it('returns 503 lock_lost when the runner loses the lock mid-run', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const { handlers } = createSyncHandlers({ + syncPromptCatalog: jest.fn().mockResolvedValue({ ok: false, error: { code: 'lock_lost' } }), + }); + const res = mockRes(); + + await handlers.syncPrompts(syncReq(), res); + + expect(res.statusCode).toBe(503); + expect(res.body).toEqual({ code: 'lock_lost' }); + }); + + /** Runs the bound page lister once so a Langfuse failure reaches the + * handler through the same `LangfusePromptRequestError` path the real + * runner surfaces it on. */ + function runnerThatListsOnce() { + return jest + .fn() + .mockImplementation( + async ({ + listPrompts, + }: { + listPrompts: (query: { type: 'text'; limit: number; page: number }) => Promise; + }) => { + await listPrompts({ type: 'text', limit: 100, page: 1 }); + return { ok: true, value: { created: 0, unchanged: 0, skipped: 0 } }; + }, + ); + } + + it('maps an upstream 500 from Langfuse to 502 upstream without leaking the body', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + global.fetch = jest + .fn() + .mockResolvedValue( + fetchJsonResponse({ message: 'MARKER_SECRET_LEAK_TOKEN' }, 500), + ) as unknown as typeof fetch; + const { handlers } = createSyncHandlers({ syncPromptCatalog: runnerThatListsOnce() }); + const res = mockRes(); + + await handlers.syncPrompts(syncReq(), res); + + expect(res.statusCode).toBe(502); + expect(res.body).toEqual({ code: 'upstream', status: 500 }); + expect(JSON.stringify(res.body)).not.toContain('MARKER_SECRET_LEAK_TOKEN'); + }); + + it('maps a Langfuse timeout to 504 timeout', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const timeoutError = new Error('The operation was aborted due to timeout'); + timeoutError.name = 'TimeoutError'; + global.fetch = jest.fn().mockRejectedValue(timeoutError) as unknown as typeof fetch; + const { handlers } = createSyncHandlers({ syncPromptCatalog: runnerThatListsOnce() }); + const res = mockRes(); + + await handlers.syncPrompts(syncReq(), res); + + expect(res.statusCode).toBe(504); + expect(res.body).toEqual({ code: 'timeout' }); + }); + + it('maps an unexpected runner error to 500 internal without leaking its message', async () => { + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + const { handlers } = createSyncHandlers({ + syncPromptCatalog: jest + .fn() + .mockRejectedValue(new Error('db down MARKER_SECRET_LEAK_TOKEN')), + }); + const res = mockRes(); + + await handlers.syncPrompts(syncReq(), res); + + expect(res.statusCode).toBe(500); + expect(res.body).toEqual({ code: 'internal' }); + expect(JSON.stringify(res.body)).not.toContain('MARKER_SECRET_LEAK_TOKEN'); + }); + + it('syncs in env mode with the "env" source destination and no tenant', async () => { + delete process.env.TENANT_ISOLATION_STRICT; + delete process.env.LANGFUSE_FANOUT_ENABLED; + delete process.env.LANGFUSE_FANOUT_COLLECTOR_URL; + process.env.LANGFUSE_PROMPT_SYNC_AVAILABLE = 'true'; + process.env.LANGFUSE_PUBLIC_KEY = 'env-public'; + process.env.LANGFUSE_SECRET_KEY = 'env-secret'; + process.env.LANGFUSE_PROJECT_ID = 'env-project-1'; + const { handlers, deps } = createHandlers({ + findConfigByPrincipal: jest + .fn() + .mockResolvedValue(baseConfigDoc({ promptSync: { enabled: true } })), + }); + const res = mockRes(); + + await handlers.syncPrompts( + mockReq({ + user: { id: 'u1', role: 'ADMIN' }, + config: { langfuse: { promptSync: { enabled: true } } }, + }), + res, + ); + + expect(res.statusCode).toBe(200); + expect(runnerRequest(deps).source).toEqual({ + destination: 'env', + projectId: 'env-project-1', + tenantId: undefined, + }); + delete process.env.LANGFUSE_PROJECT_ID; + }); + }); }); diff --git a/packages/api/src/admin/langfuse.ts b/packages/api/src/admin/langfuse.ts index 8f7e6409a17..f80116712a7 100644 --- a/packages/api/src/admin/langfuse.ts +++ b/packages/api/src/admin/langfuse.ts @@ -17,12 +17,15 @@ import type { TUpdateLangfusePromptSyncRequest, TLangfusePromptListResponse, TLangfusePromptGetResponse, + TLangfusePromptSyncResult, TLangfusePromptErrorBody, } from 'librechat-data-provider'; -import type { ConfigRecord, MessageMethods } from '@librechat/data-schemas'; +import type { MessageMethods } from '@librechat/data-schemas'; import type { Response } from 'express'; import type { LangfusePromptConnection, LangfuseTextPromptSelector } from '~/langfuse/prompts'; import type { LangfuseTenantDestination } from '~/langfuse/tenantDestinations'; +import type { LangfusePromptSyncRunner } from '~/langfuse/promptSyncRunner'; +import type { StoredBaseConfig } from '~/langfuse/promptSync'; import type { ServerRequest } from '~/types/http'; import type { RecordAuditEntry } from './audit'; import { @@ -30,11 +33,14 @@ import { isLangfusePromptSyncAvailable, getLangfusePromptSyncTimeoutMs, getLangfuseConnectionAvailability, + getLangfusePromptSyncLockLeaseMs, + getLangfusePromptSyncMaxPages, } from '~/langfuse/policy'; import { readStoredLangfuse, buildPromptSyncConnection, toLangfusePromptErrorResponse, + getPromptSyncSourceIdentity, } from '~/langfuse/promptSync'; import { getLangfuseTenantDestinations, @@ -76,12 +82,19 @@ export interface LangfuseConnectionEvent { verification_result: 'skipped' | 'success' | 'failed'; } +/** Stored base-config fields the Langfuse admin handlers read. The data-schemas `ConfigRecord` satisfies it. */ +export interface LangfuseAdminConfig extends StoredBaseConfig { + priority?: number; + isActive?: boolean; + updatedAt?: Date; +} + export interface AdminLangfuseDeps { findConfigByPrincipal: ( principalType: PrincipalType, principalId: string, options?: { includeInactive?: boolean }, - ) => Promise; + ) => Promise; /** * Sets the fields in one compare-and-swap write, re-reading on a concurrent change. An * omitted priority keeps the stored one (Base merges first whatever its priority). @@ -92,13 +105,14 @@ export interface AdminLangfuseDeps { principalModel: PrincipalModel, fields: Record, priority?: number, - ) => Promise; + ) => Promise; toggleConfigActive: ( principalType: PrincipalType, principalId: string, isActive: boolean, - ) => Promise; + ) => Promise; getMessages: MessageMethods['getMessages']; + syncPromptCatalog: LangfusePromptSyncRunner; invalidateConfigCaches?: (tenantId?: string) => Promise; recordConnectionUpdate?: (event: LangfuseConnectionEvent) => void; /** Optional audit emission for connection and prompt sync updates; fail-open unless `auditFailClosed`. */ @@ -110,7 +124,7 @@ function getTenantId(req: ServerRequest): string | undefined { return (req.user as { tenantId?: string } | undefined)?.tenantId; } -function buildStatus(config: ConfigRecord | null): TLangfuseConnectionStatus { +function buildStatus(config: LangfuseAdminConfig | null): TLangfuseConnectionStatus { const stored = readStoredLangfuse(config); const configured = Boolean(stored?.publicKey && stored?.secretKey); const promptSyncAvailable = isLangfusePromptSyncAvailable(); @@ -212,7 +226,7 @@ function respondToPromptRequestError(res: Response, error: unknown, logPrefix: s } logger.error(logPrefix, error); - return res.status(500).json({ code: 'upstream' } satisfies TLangfusePromptErrorBody); + return res.status(500).json({ code: 'internal' } satisfies TLangfusePromptErrorBody); } type LangfuseVerificationFailure = { @@ -367,12 +381,14 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): { updatePromptSync: (req: ServerRequest, res: Response) => Promise; listPrompts: (req: ServerRequest, res: Response) => Promise; getPrompt: (req: ServerRequest, res: Response) => Promise; + syncPrompts: (req: ServerRequest, res: Response) => Promise; } { const { findConfigByPrincipal, patchConfigFields, toggleConfigActive, getMessages, + syncPromptCatalog, invalidateConfigCaches, recordConnectionUpdate = (event) => logger.info({ message: '[adminLangfuse] Connection updated', ...event }), @@ -385,7 +401,9 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): { logPrefix: '[adminLangfuse]', }); - function findBaseConfig(options?: { includeInactive?: boolean }): Promise { + function findBaseConfig(options?: { + includeInactive?: boolean; + }): Promise { return options ? findConfigByPrincipal(PrincipalType.ROLE, BASE_CONFIG_PRINCIPAL_ID, options) : findConfigByPrincipal(PrincipalType.ROLE, BASE_CONFIG_PRINCIPAL_ID); @@ -843,6 +861,73 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): { } } + /** Syncs every Langfuse text prompt in the tenant's project into the local + * catalog as read-only pointer groups. Gated the same way as `listPrompts`/ + * `getPrompt` (deployment, tenant switch, resolvable connection), plus the + * source identity the runner records on each new group. Runs under the + * per-tenant distributed lock the runner itself acquires; a lock already + * held or lost mid-run never reaches this handler as a thrown error, so + * those map to `409`/`503` instead of the generic catch-all. + * + * The sync runs inside this HTTP request. The page cap (default 50 pages + * of 100 prompts each, 5,000 prompts) is the request-sized ceiling; a + * larger catalog is truncated at the cap with a warning log, not an + * error — the response's `complete: false` tells the caller the run + * stopped before every upstream page was read. The lock lease (default + * 300 s) is renewed after every page but the last, and only between + * pages, so a page whose own processing outlasts the lease can still let + * a second run start concurrently. The sync is append-only: it creates + * pointers for new Langfuse text prompts and never deletes, re-grants or + * reconciles existing groups, so a prompt deleted in Langfuse keeps its + * local pointer; a truncated run is not resumed from where it stopped — + * a later run re-lists from page 1. */ + async function syncPrompts(req: ServerRequest, res: Response): Promise { + const gate = await gatePromptSyncTenantSwitch(res); + if (gate.response) { + return gate.response; + } + + const connection = resolvePromptSyncConnection(req, gate.stored); + if (!connection) { + return res.status(409).json({ code: 'not_configured' } satisfies TLangfusePromptErrorBody); + } + + const timeoutMs = getLangfusePromptSyncTimeoutMs(); + try { + const identity = await getPromptSyncSourceIdentity( + req.config, + gate.stored, + getTenantId(req), + timeoutMs, + ); + if (!identity) { + return res.status(409).json({ code: 'not_configured' } satisfies TLangfusePromptErrorBody); + } + + const result = await syncPromptCatalog({ + source: { + destination: identity.destination, + projectId: identity.projectId, + tenantId: getTenantId(req), + }, + listPrompts: (query) => listLangfusePrompts(connection, query, { timeoutMs }), + lockLeaseMs: getLangfusePromptSyncLockLeaseMs(req.config), + maxPages: getLangfusePromptSyncMaxPages(req.config), + }); + + if (!result.ok) { + const status = result.error.code === 'already_running' ? 409 : 503; + return res + .status(status) + .json({ code: result.error.code } satisfies TLangfusePromptErrorBody); + } + + return res.status(200).json(result.value satisfies TLangfusePromptSyncResult); + } catch (error) { + return respondToPromptRequestError(res, error, '[adminLangfuse] syncPrompts error:'); + } + } + return { getConnection, getSessionLink, @@ -851,5 +936,6 @@ export function createAdminLangfuseHandlers(deps: AdminLangfuseDeps): { updatePromptSync, listPrompts, getPrompt, + syncPrompts, }; } diff --git a/packages/api/src/langfuse/index.ts b/packages/api/src/langfuse/index.ts index 63074258e03..af96a758c19 100644 --- a/packages/api/src/langfuse/index.ts +++ b/packages/api/src/langfuse/index.ts @@ -1,6 +1,8 @@ export * from './destinations'; export * from './feedback'; export * from './policy'; +export * from './promptCatalogSync'; +export * from './promptSyncRunner'; export * from './prompts'; export * from './reader'; export * from './trace'; diff --git a/packages/api/src/langfuse/policy.spec.ts b/packages/api/src/langfuse/policy.spec.ts index ed2255f26c7..0cfd5b7f172 100644 --- a/packages/api/src/langfuse/policy.spec.ts +++ b/packages/api/src/langfuse/policy.spec.ts @@ -1,3 +1,11 @@ +import type { AppConfig } from '@librechat/data-schemas'; + +/** A minimal AppConfig stub carrying only the langfuse.promptSync fields the + * lock-lease and page-cap readers consult. */ +function langfuseAppConfig(promptSync: Record): AppConfig { + return { langfuse: { promptSync } } as unknown as AppConfig; +} + const envKeys = [ 'LANGFUSE_PUBLIC_KEY', 'LANGFUSE_SECRET_KEY', @@ -170,6 +178,50 @@ describe('Langfuse policy', () => { expect(getLangfusePromptSyncTimeoutMs()).toBe(10000); }); + + it('defaults the Langfuse prompt sync lock lease to 300000ms when unset', async () => { + const { getLangfusePromptSyncLockLeaseMs } = await import('./policy'); + + expect(getLangfusePromptSyncLockLeaseMs(undefined)).toBe(300000); + expect(getLangfusePromptSyncLockLeaseMs(langfuseAppConfig({}))).toBe(300000); + }); + + it('uses a configured Langfuse prompt sync lock lease', async () => { + const { getLangfusePromptSyncLockLeaseMs } = await import('./policy'); + + expect(getLangfusePromptSyncLockLeaseMs(langfuseAppConfig({ lockLeaseMs: 120000 }))).toBe( + 120000, + ); + }); + + it('clamps the Langfuse prompt sync lock lease to 60000-3600000', async () => { + const { getLangfusePromptSyncLockLeaseMs } = await import('./policy'); + + expect(getLangfusePromptSyncLockLeaseMs(langfuseAppConfig({ lockLeaseMs: 1 }))).toBe(60000); + expect(getLangfusePromptSyncLockLeaseMs(langfuseAppConfig({ lockLeaseMs: 10000000 }))).toBe( + 3600000, + ); + }); + + it('defaults the Langfuse prompt sync page cap to 50 when unset', async () => { + const { getLangfusePromptSyncMaxPages } = await import('./policy'); + + expect(getLangfusePromptSyncMaxPages(undefined)).toBe(50); + expect(getLangfusePromptSyncMaxPages(langfuseAppConfig({}))).toBe(50); + }); + + it('uses a configured Langfuse prompt sync page cap', async () => { + const { getLangfusePromptSyncMaxPages } = await import('./policy'); + + expect(getLangfusePromptSyncMaxPages(langfuseAppConfig({ maxPages: 5 }))).toBe(5); + }); + + it('clamps the Langfuse prompt sync page cap to 1-1000', async () => { + const { getLangfusePromptSyncMaxPages } = await import('./policy'); + + expect(getLangfusePromptSyncMaxPages(langfuseAppConfig({ maxPages: 0 }))).toBe(1); + expect(getLangfusePromptSyncMaxPages(langfuseAppConfig({ maxPages: 100000 }))).toBe(1000); + }); }); describe('Langfuse settings availability', () => { diff --git a/packages/api/src/langfuse/policy.ts b/packages/api/src/langfuse/policy.ts index 13aeab0f387..1c885c026f0 100644 --- a/packages/api/src/langfuse/policy.ts +++ b/packages/api/src/langfuse/policy.ts @@ -118,6 +118,40 @@ export function getLangfusePromptSyncTimeoutMs(): number { ); } +const DEFAULT_PROMPT_SYNC_LOCK_LEASE_MS = 300_000; +const MIN_PROMPT_SYNC_LOCK_LEASE_MS = 60_000; +const MAX_PROMPT_SYNC_LOCK_LEASE_MS = 3_600_000; + +/** Prompt sync lock lease, in milliseconds. Configured on librechat.yaml's + * langfuse.promptSync.lockLeaseMs; clamps to the supported range and falls + * back to the default when unset. */ +export function getLangfusePromptSyncLockLeaseMs(appConfig: AppConfig | undefined): number { + const leaseMs = appConfig?.langfuse?.promptSync?.lockLeaseMs; + if (leaseMs == null) { + return DEFAULT_PROMPT_SYNC_LOCK_LEASE_MS; + } + return Math.round( + Math.min(MAX_PROMPT_SYNC_LOCK_LEASE_MS, Math.max(MIN_PROMPT_SYNC_LOCK_LEASE_MS, leaseMs)), + ); +} + +const DEFAULT_PROMPT_SYNC_MAX_PAGES = 50; +const MIN_PROMPT_SYNC_MAX_PAGES = 1; +const MAX_PROMPT_SYNC_MAX_PAGES = 1000; + +/** Page cap for one sync request, in pages of up to 100 Langfuse prompts + * each. Configured on librechat.yaml's langfuse.promptSync.maxPages; + * clamps to the supported range and falls back to the default when unset. */ +export function getLangfusePromptSyncMaxPages(appConfig: AppConfig | undefined): number { + const maxPages = appConfig?.langfuse?.promptSync?.maxPages; + if (maxPages == null) { + return DEFAULT_PROMPT_SYNC_MAX_PAGES; + } + return Math.round( + Math.min(MAX_PROMPT_SYNC_MAX_PAGES, Math.max(MIN_PROMPT_SYNC_MAX_PAGES, maxPages)), + ); +} + /** Whether the deployment offers Langfuse connection settings, and if not, why. */ export type LangfuseConnectionAvailability = | { available: true } diff --git a/packages/api/src/langfuse/promptCatalogSync.spec.ts b/packages/api/src/langfuse/promptCatalogSync.spec.ts new file mode 100644 index 00000000000..c9033fb0865 --- /dev/null +++ b/packages/api/src/langfuse/promptCatalogSync.spec.ts @@ -0,0 +1,436 @@ +import type { TLangfusePromptListItem, TLangfusePromptListResponse } from 'librechat-data-provider'; +import type { PromptCatalogSyncDeps, PromptCatalogSyncSource } from './promptCatalogSync'; +import { syncLangfusePromptCatalog } from './promptCatalogSync'; +import { makeSourceAuthorId } from '~/utils/author'; + +function textItem(name: string): TLangfusePromptListItem { + return { name, type: 'text', versions: [1], labels: [], tags: [], lastUpdatedAt: '2026-01-01' }; +} + +function chatItem(name: string): TLangfusePromptListItem { + return { name, type: 'chat', versions: [1], labels: [], tags: [], lastUpdatedAt: '2026-01-01' }; +} + +function listResult( + items: TLangfusePromptListItem[], + totalPages: number, + page = 1, +): TLangfusePromptListResponse { + return { items, meta: { page, limit: 100, totalItems: items.length, totalPages } }; +} + +function createDeps(overrides: Partial = {}): PromptCatalogSyncDeps { + return { + listPrompts: jest.fn().mockResolvedValue(listResult([], 0)), + findPromptGroup: jest.fn().mockResolvedValue(null), + createPromptGroup: jest.fn().mockResolvedValue({ _id: 'group-1' }), + isDuplicateKeyError: jest.fn().mockReturnValue(false), + isValidationError: jest.fn().mockReturnValue(false), + grantPublicViewer: jest.fn().mockResolvedValue(undefined), + deletePromptGroup: jest.fn().mockResolvedValue(undefined), + lock: { + acquire: jest.fn().mockResolvedValue('token'), + extend: jest.fn().mockResolvedValue(true), + release: jest.fn().mockResolvedValue(undefined), + }, + lockLeaseMs: 300000, + logger: { warn: jest.fn(), error: jest.fn() }, + ...overrides, + }; +} + +const source: PromptCatalogSyncSource = { destination: 'us', projectId: 'project-1' }; + +describe('syncLangfusePromptCatalog', () => { + it('returns already_running and never lists when the lock is held', async () => { + const deps = createDeps({ + lock: { ...createDeps().lock, acquire: jest.fn().mockResolvedValue(null) }, + }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ ok: false, error: { code: 'already_running' } }); + expect(deps.listPrompts).not.toHaveBeenCalled(); + expect(deps.lock.release).not.toHaveBeenCalled(); + }); + + it('extends the lock once per page that is followed by another page', async () => { + const listPrompts = jest + .fn() + .mockResolvedValueOnce(listResult([textItem('a')], 3, 1)) + .mockResolvedValueOnce(listResult([textItem('b')], 3, 2)) + .mockResolvedValueOnce(listResult([textItem('c')], 3, 3)); + const deps = createDeps({ listPrompts }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 3, unchanged: 0, skipped: 0, complete: true }, + }); + expect(listPrompts).toHaveBeenCalledTimes(3); + expect(listPrompts).toHaveBeenNthCalledWith(1, { type: 'text', limit: 100, page: 1 }); + expect(listPrompts).toHaveBeenNthCalledWith(2, { type: 'text', limit: 100, page: 2 }); + expect(listPrompts).toHaveBeenNthCalledWith(3, { type: 'text', limit: 100, page: 3 }); + // Pages 1 and 2 are each followed by another page, so each extends once; + // page 3 is the last, so no extend follows it. + expect(deps.lock.extend).toHaveBeenCalledTimes(2); + expect(deps.lock.release).toHaveBeenCalledTimes(1); + }); + + it('does not extend the lock after the only page', async () => { + const listPrompts = jest.fn().mockResolvedValueOnce(listResult([textItem('a')], 1, 1)); + const deps = createDeps({ listPrompts }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 1, unchanged: 0, skipped: 0, complete: true }, + }); + expect(deps.lock.extend).not.toHaveBeenCalled(); + expect(deps.lock.release).toHaveBeenCalledTimes(1); + }); + + it('stops with lock_lost once extend fails, releasing the lock', async () => { + const listPrompts = jest + .fn() + .mockResolvedValueOnce(listResult([textItem('a')], 3, 1)) + .mockResolvedValueOnce(listResult([textItem('b')], 3, 2)); + const extend = jest.fn().mockResolvedValueOnce(true).mockResolvedValueOnce(false); + const deps = createDeps({ listPrompts, lock: { ...createDeps().lock, extend } }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ ok: false, error: { code: 'lock_lost' } }); + expect(listPrompts).toHaveBeenCalledTimes(2); + expect(deps.lock.release).toHaveBeenCalledTimes(1); + }); + + it('stops on an empty page before reaching totalPages', async () => { + const listPrompts = jest + .fn() + .mockResolvedValueOnce(listResult([textItem('a')], 5, 1)) + .mockResolvedValueOnce(listResult([], 5, 2)); + const deps = createDeps({ listPrompts }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 1, unchanged: 0, skipped: 0, complete: true }, + }); + expect(listPrompts).toHaveBeenCalledTimes(2); + expect(deps.logger.warn).not.toHaveBeenCalled(); + }); + + it('stops at the maxPages cap even when more pages are reported', async () => { + const listPrompts = jest + .fn() + .mockImplementation((query: { page: number }) => + Promise.resolve(listResult([textItem(`item-${query.page}`)], 1000, query.page)), + ); + const deps = createDeps({ listPrompts, maxPages: 2 }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 2, unchanged: 0, skipped: 0, complete: false }, + }); + expect(listPrompts).toHaveBeenCalledTimes(2); + }); + + it('warns once when the page cap stops the loop before totalPages', async () => { + const listPrompts = jest + .fn() + .mockImplementation((query: { page: number }) => + Promise.resolve(listResult([textItem(`item-${query.page}`)], 5, query.page)), + ); + const deps = createDeps({ listPrompts, maxPages: 2 }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 2, unchanged: 0, skipped: 0, complete: false }, + }); + expect(listPrompts).toHaveBeenCalledTimes(2); + expect(deps.logger.warn).toHaveBeenCalledTimes(1); + expect(deps.logger.warn).toHaveBeenCalledWith( + '[langfuse] Langfuse prompt sync stopped at the page cap', + { maxPages: 2, totalPages: 5 }, + ); + }); + + it('does not extend the lock after the page cap is reached', async () => { + const listPrompts = jest + .fn() + .mockImplementation((query: { page: number }) => + Promise.resolve(listResult([textItem(`item-${query.page}`)], 3, query.page)), + ); + const deps = createDeps({ listPrompts, maxPages: 1 }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 1, unchanged: 0, skipped: 0, complete: false }, + }); + expect(deps.lock.extend).not.toHaveBeenCalled(); + expect(deps.logger.warn).toHaveBeenCalledTimes(1); + expect(deps.logger.warn).toHaveBeenCalledWith( + '[langfuse] Langfuse prompt sync stopped at the page cap', + { maxPages: 1, totalPages: 3 }, + ); + }); + + it('respects an injected pageLimit', async () => { + const listPrompts = jest.fn().mockResolvedValue(listResult([], 0)); + const deps = createDeps({ listPrompts, pageLimit: 25 }); + + await syncLangfusePromptCatalog(source, deps); + + expect(listPrompts).toHaveBeenCalledWith({ type: 'text', limit: 25, page: 1 }); + }); + + it('counts a found match as unchanged without creating or granting', async () => { + const findPromptGroup = jest.fn().mockResolvedValue({ _id: 'existing' }); + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + findPromptGroup, + }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 0, unchanged: 1, skipped: 0, complete: true }, + }); + expect(deps.createPromptGroup).not.toHaveBeenCalled(); + expect(deps.grantPublicViewer).not.toHaveBeenCalled(); + }); + + it('creates and grants a new group on a miss', async () => { + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 1, unchanged: 0, skipped: 0, complete: true }, + }); + expect(deps.createPromptGroup).toHaveBeenCalledWith({ + sourceDestination: 'us', + sourceProjectId: 'project-1', + sourcePromptName: 'a', + name: 'a', + author: makeSourceAuthorId('langfuse', 'project-1', undefined), + authorName: 'Langfuse Sync', + }); + expect(deps.grantPublicViewer).toHaveBeenCalledWith('group-1'); + }); + + it('counts a duplicate-key create as unchanged without granting', async () => { + const createPromptGroup = jest.fn().mockRejectedValue(new Error('E11000')); + const isDuplicateKeyError = jest.fn().mockReturnValue(true); + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + createPromptGroup, + isDuplicateKeyError, + }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 0, unchanged: 1, skipped: 0, complete: true }, + }); + expect(deps.grantPublicViewer).not.toHaveBeenCalled(); + }); + + it('rejects when create fails with an unexpected error, releasing the lock once', async () => { + const createPromptGroup = jest.fn().mockRejectedValueOnce(new Error('boom')); + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + createPromptGroup, + }); + + await expect(syncLangfusePromptCatalog(source, deps)).rejects.toThrow('boom'); + expect(deps.lock.release).toHaveBeenCalledTimes(1); + }); + + it('counts a validation error on create as skipped and continues the loop', async () => { + const createPromptGroup = jest + .fn() + .mockRejectedValueOnce(new Error('invalid')) + .mockResolvedValueOnce({ _id: 'group-2' }); + const isValidationError = jest.fn().mockReturnValue(true); + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a'), textItem('b')], 1)), + createPromptGroup, + isValidationError, + }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 1, unchanged: 0, skipped: 1, complete: true }, + }); + expect(deps.logger.warn).toHaveBeenCalledWith( + expect.stringContaining('validation'), + 'a', + expect.any(Error), + ); + }); + + it('rolls back and rejects with the original grant error when the grant throws', async () => { + const grantPublicViewer = jest.fn().mockRejectedValue(new Error('grant failed')); + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + grantPublicViewer, + }); + + await expect(syncLangfusePromptCatalog(source, deps)).rejects.toThrow('grant failed'); + expect(deps.deletePromptGroup).toHaveBeenCalledWith('group-1'); + expect(deps.lock.release).toHaveBeenCalledTimes(1); + }); + + it('still rejects with the original grant error and error-logs when the rollback delete also throws', async () => { + const grantPublicViewer = jest.fn().mockRejectedValue(new Error('grant failed')); + const deletePromptGroup = jest.fn().mockRejectedValue(new Error('delete failed')); + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + grantPublicViewer, + deletePromptGroup, + }); + + await expect(syncLangfusePromptCatalog(source, deps)).rejects.toThrow('grant failed'); + expect(deps.logger.error).toHaveBeenCalledTimes(2); + }); + + it('warns and counts a non-text item as skipped', async () => { + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([chatItem('chat-1')], 1)), + }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 0, unchanged: 0, skipped: 1, complete: true }, + }); + expect(deps.logger.warn).toHaveBeenCalledWith(expect.stringContaining('chat-1')); + expect(deps.findPromptGroup).not.toHaveBeenCalled(); + }); + + it('counts a name repeated across pages only once', async () => { + const listPrompts = jest + .fn() + .mockResolvedValueOnce(listResult([textItem('dup')], 2, 1)) + .mockResolvedValueOnce(listResult([textItem('dup')], 2, 2)); + const deps = createDeps({ listPrompts }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 1, unchanged: 0, skipped: 0, complete: true }, + }); + expect(deps.findPromptGroup).toHaveBeenCalledTimes(1); + }); + + it('releases the lock when listPrompts throws, and the error still propagates', async () => { + class LangfusePromptRequestError extends Error { + code = 'upstream'; + } + const listPrompts = jest.fn().mockRejectedValue(new LangfusePromptRequestError('boom')); + const deps = createDeps({ listPrompts }); + + await expect(syncLangfusePromptCatalog(source, deps)).rejects.toThrow('boom'); + expect(deps.lock.release).toHaveBeenCalledTimes(1); + }); + + it('resolves ok with counts when lock.release rejects after a successful run', async () => { + const release = jest.fn().mockRejectedValue(new Error('release failed')); + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + lock: { ...createDeps().lock, release }, + }); + + const result = await syncLangfusePromptCatalog(source, deps); + + expect(result).toEqual({ + ok: true, + value: { created: 1, unchanged: 0, skipped: 0, complete: true }, + }); + expect(deps.logger.warn).toHaveBeenCalledWith( + expect.stringContaining('Could not release'), + expect.any(Error), + ); + }); + + it('seeds the author id from the project id alone when no tenant is given', async () => { + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + }); + + await syncLangfusePromptCatalog({ destination: 'us', projectId: 'project-1' }, deps); + + expect(deps.createPromptGroup).toHaveBeenCalledWith( + expect.objectContaining({ author: makeSourceAuthorId('langfuse', 'project-1', undefined) }), + ); + }); + + it('seeds a different author id per tenant for the same project', async () => { + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + }); + + await syncLangfusePromptCatalog( + { destination: 'us', projectId: 'project-1', tenantId: 'tenant-a' }, + deps, + ); + + expect(deps.createPromptGroup).toHaveBeenCalledWith( + expect.objectContaining({ + author: makeSourceAuthorId('langfuse', 'project-1', 'tenant-a'), + }), + ); + expect(deps.createPromptGroup).not.toHaveBeenCalledWith( + expect.objectContaining({ author: makeSourceAuthorId('langfuse', 'project-1', undefined) }), + ); + }); + + it('uses the lock key prompt-sync:langfuse:, empty for single-tenant', async () => { + const deps = createDeps(); + + await syncLangfusePromptCatalog({ destination: 'us', projectId: 'project-1' }, deps); + expect(deps.lock.acquire).toHaveBeenCalledWith('prompt-sync:langfuse:', deps.lockLeaseMs); + + const tenantDeps = createDeps(); + await syncLangfusePromptCatalog( + { destination: 'us', projectId: 'project-1', tenantId: 'tenant-a' }, + tenantDeps, + ); + expect(tenantDeps.lock.acquire).toHaveBeenCalledWith( + 'prompt-sync:langfuse:tenant-a', + tenantDeps.lockLeaseMs, + ); + }); + + it('always passes the Langfuse Sync author name', async () => { + const deps = createDeps({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + }); + + await syncLangfusePromptCatalog(source, deps); + + expect(deps.createPromptGroup).toHaveBeenCalledWith( + expect.objectContaining({ authorName: 'Langfuse Sync' }), + ); + }); +}); diff --git a/packages/api/src/langfuse/promptCatalogSync.ts b/packages/api/src/langfuse/promptCatalogSync.ts new file mode 100644 index 00000000000..0f7dc1f6c69 --- /dev/null +++ b/packages/api/src/langfuse/promptCatalogSync.ts @@ -0,0 +1,237 @@ +import type { TLangfusePromptListItem, TLangfusePromptListResponse } from 'librechat-data-provider'; +import type { DistributedLock } from '~/cache/lock'; +import { makeSourceAuthorId } from '~/utils/author'; + +const LANGFUSE_SYNC_AUTHOR_NAME = 'Langfuse Sync'; +const PAGE_LIMIT = 100; +const MAX_PAGES = 50; + +export type PromptCatalogSyncSource = { + destination: string; + projectId: string; + tenantId?: string; +}; + +export type PromptCatalogSyncCounts = { + created: number; + unchanged: number; + skipped: number; + /** False when the page cap stopped the run before every upstream page was read; the counts + * above cover only the pages the run actually fetched. */ + complete: boolean; +}; + +export type PromptCatalogSyncResult = + | { ok: true; value: PromptCatalogSyncCounts } + | { ok: false; error: { code: 'already_running' | 'lock_lost' } }; + +/** The identity a synced group is found or created by. */ +export type PromptGroupSourceIdentity = { + sourceDestination: string; + sourceProjectId: string; + sourcePromptName: string; +}; + +export type PromptCatalogSyncCreateInput = PromptGroupSourceIdentity & { + name: string; + author: string; + authorName: string; +}; + +export type PromptCatalogSyncDeps = { + listPrompts: (query: { + type: 'text'; + limit: number; + page: number; + }) => Promise; + findPromptGroup: (identity: PromptGroupSourceIdentity) => Promise<{ _id: string } | null>; + createPromptGroup: (input: PromptCatalogSyncCreateInput) => Promise<{ _id: string }>; + isDuplicateKeyError: (error: unknown) => boolean; + isValidationError: (error: unknown) => boolean; + grantPublicViewer: (groupId: string) => Promise; + deletePromptGroup: (groupId: string) => Promise; + lock: DistributedLock; + lockLeaseMs: number; + logger: { + warn: (message: string, ...args: unknown[]) => void; + error: (message: string, ...args: unknown[]) => void; + }; + pageLimit?: number; + maxPages?: number; +}; + +/** Finds-or-creates one Langfuse text prompt as a pointer-only group, granting + * a PUBLIC viewer entry only when it was just created. A match is left as is: + * no ACL write and no update statement run against an existing group. */ +async function syncOne( + item: TLangfusePromptListItem, + identity: PromptGroupSourceIdentity, + author: string, + deps: PromptCatalogSyncDeps, +): Promise<'created' | 'unchanged' | 'skipped'> { + // The group insert and its initial PUBLIC grant below are separate writes. If an earlier + // run inserted the group and then exited before granting PUBLIC access, this run finds + // the group here and returns `unchanged` without retrying the grant, so the group keeps + // no PUBLIC viewer entry. + const existing = await deps.findPromptGroup(identity); + if (existing) { + return 'unchanged'; + } + + let created: { _id: string }; + try { + created = await deps.createPromptGroup({ + ...identity, + name: item.name, + author, + authorName: LANGFUSE_SYNC_AUTHOR_NAME, + }); + } catch (error) { + if (deps.isDuplicateKeyError(error)) { + return 'unchanged'; + } + if (deps.isValidationError(error)) { + deps.logger.warn( + '[langfuse] Could not save a synced prompt group that failed validation:', + item.name, + error, + ); + return 'skipped'; + } + throw error; + } + + try { + await deps.grantPublicViewer(created._id); + } catch (error) { + deps.logger.error('[langfuse] Could not grant public access to a synced prompt group:', error); + try { + await deps.deletePromptGroup(created._id); + } catch (deleteError) { + deps.logger.error( + '[langfuse] Could not roll back an ungranted synced prompt group:', + deleteError, + ); + } + throw error; + } + + return 'created'; +} + +/** + * Syncs every Langfuse text prompt in `source`'s project into the local + * prompt catalog as a read-only, pointer-only group: name and source identity + * only, never content. A synced group is never modified once inserted; + * `unchanged` covers both an existing match and a duplicate-key race lost to + * a concurrent insert — expected when the lock below is lost mid-page and a + * second run starts while the first is still saving that page's items. A + * chat prompt that slips past the server-side `type: 'text'` filter is + * warned and counted `skipped`, never saved. + * + * Runs under a per-tenant lock (`prompt-sync:langfuse:`): a + * lock already held returns `already_running` without listing a single + * prompt. The lock is extended only when another page will actually be + * fetched — the current page is not the last one (`page < totalPages`) and + * the page cap has not been reached (`page < maxPages`) — so a fully + * processed sync, and a run that ends because it hit the page cap, never + * fails on a spurious lock loss it no longer needs the lock for. The + * renewal happens only between pages, never during one: a page whose own + * processing (listing and saving its items) outlasts the lease can still + * let the lease expire, so a second run can acquire the lock and start + * while the first is still partway through that page. The lock reduces + * overlapping runs; it does not guarantee only one runs at a time. A lost + * lock stops the loop with `lock_lost` before saving further pages — pages + * already saved stay, so a re-run completes the sync. The lock is released + * on every exit path, including when the upstream client throws. + * + * The result's `complete` flag is `false` when the page cap (`maxPages`) + * stopped the run before every upstream page was read; the counts then + * cover only the pages actually fetched. This sync is append-only and + * always re-lists from page 1, so a re-run with the same page limit reads + * the same pages again and makes no further progress. Importing the rest + * requires raising the `langfuse.promptSync.maxPages` config. + */ +export async function syncLangfusePromptCatalog( + source: PromptCatalogSyncSource, + deps: PromptCatalogSyncDeps, +): Promise { + const lockKey = `prompt-sync:langfuse:${source.tenantId ?? ''}`; + const token = await deps.lock.acquire(lockKey, deps.lockLeaseMs); + if (token == null) { + return { ok: false, error: { code: 'already_running' } }; + } + + try { + const author = makeSourceAuthorId('langfuse', source.projectId, source.tenantId); + const counts: Omit = { + created: 0, + unchanged: 0, + skipped: 0, + }; + const seen = new Set(); + const pageLimit = deps.pageLimit ?? PAGE_LIMIT; + const maxPages = deps.maxPages ?? MAX_PAGES; + // True once the loop breaks because the upstream pages ran out (an empty page or the + // last page), as opposed to running out of the page cap first. + let exhaustedUpstream = false; + let lastTotalPages = 0; + + for (let page = 1; page <= maxPages; page++) { + const result = await deps.listPrompts({ type: 'text', limit: pageLimit, page }); + lastTotalPages = result.meta.totalPages; + + for (const item of result.items) { + if (item.type !== 'text') { + deps.logger.warn( + `[langfuse] Skipped a non-text prompt "${item.name}" past the server-side filter`, + ); + counts.skipped += 1; + continue; + } + if (seen.has(item.name)) { + continue; + } + seen.add(item.name); + + const identity: PromptGroupSourceIdentity = { + sourceDestination: source.destination, + sourceProjectId: source.projectId, + sourcePromptName: item.name, + }; + const outcome = await syncOne(item, identity, author, deps); + counts[outcome] += 1; + } + + if (result.items.length === 0 || page >= result.meta.totalPages) { + exhaustedUpstream = true; + break; + } + + if (page < maxPages) { + const extended = await deps.lock.extend(lockKey, token, deps.lockLeaseMs); + if (!extended) { + return { ok: false, error: { code: 'lock_lost' } }; + } + } + } + + if (!exhaustedUpstream && lastTotalPages > maxPages) { + deps.logger.warn('[langfuse] Langfuse prompt sync stopped at the page cap', { + maxPages, + totalPages: lastTotalPages, + }); + } + + return { ok: true, value: { ...counts, complete: exhaustedUpstream } }; + } finally { + try { + await deps.lock.release(lockKey, token); + } catch (error) { + deps.logger.warn( + '[langfuse] Could not release the prompt sync lock; the lease will expire', + error, + ); + } + } +} diff --git a/packages/api/src/langfuse/promptSyncRunner.spec.ts b/packages/api/src/langfuse/promptSyncRunner.spec.ts new file mode 100644 index 00000000000..5293d9503ed --- /dev/null +++ b/packages/api/src/langfuse/promptSyncRunner.spec.ts @@ -0,0 +1,249 @@ +import type { TLangfusePromptListItem, TLangfusePromptListResponse } from 'librechat-data-provider'; +import type { LangfusePromptSyncRunnerDeps, LangfusePromptSyncRequest } from './promptSyncRunner'; +import type { DistributedLock } from '~/cache/lock'; +import { createLangfusePromptSyncRunner } from './promptSyncRunner'; +import { makeSourceAuthorId } from '~/utils/author'; + +function textItem(name: string): TLangfusePromptListItem { + return { name, type: 'text', versions: [1], labels: [], tags: [], lastUpdatedAt: '2026-01-01' }; +} + +function listResult( + items: TLangfusePromptListItem[], + totalPages: number, + page = 1, +): TLangfusePromptListResponse { + return { items, meta: { page, limit: 100, totalItems: items.length, totalPages } }; +} + +function createFakeLock(overrides: Partial = {}): DistributedLock { + return { + acquire: jest.fn().mockResolvedValue('token-1'), + extend: jest.fn().mockResolvedValue(true), + release: jest.fn().mockResolvedValue(undefined), + ...overrides, + }; +} + +function createDeps(overrides: Partial = {}) { + return { + lock: createFakeLock(), + findLangfusePromptGroup: jest.fn().mockResolvedValue(null), + createLangfusePromptGroup: jest.fn().mockResolvedValue({ _id: 'group-1' }), + deletePromptGroup: jest.fn().mockResolvedValue({ message: 'Prompt group deleted' }), + grantPermission: jest.fn().mockResolvedValue(undefined), + logger: { warn: jest.fn(), error: jest.fn() }, + ...overrides, + }; +} + +function createRequest( + overrides: Partial = {}, +): LangfusePromptSyncRequest { + return { + source: { destination: 'eu', projectId: 'project-1', tenantId: 't1' }, + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('greeting')], 1)), + lockLeaseMs: 300_000, + maxPages: 50, + ...overrides, + }; +} + +describe('createLangfusePromptSyncRunner', () => { + it('creates a pointer group for a new prompt and grants the PUBLIC viewer as the system user', async () => { + const deps = createDeps(); + const run = createLangfusePromptSyncRunner(deps); + + const result = await run(createRequest()); + + expect(result).toEqual({ + ok: true, + value: { created: 1, unchanged: 0, skipped: 0, complete: true }, + }); + expect(deps.findLangfusePromptGroup).toHaveBeenCalledWith({ + sourceDestination: 'eu', + sourceProjectId: 'project-1', + sourcePromptName: 'greeting', + }); + expect(deps.createLangfusePromptGroup).toHaveBeenCalledWith({ + name: 'greeting', + author: makeSourceAuthorId('langfuse', 'project-1', 't1'), + authorName: 'Langfuse Sync', + sourceDestination: 'eu', + sourceProjectId: 'project-1', + sourcePromptName: 'greeting', + }); + expect(deps.grantPermission).toHaveBeenCalledTimes(1); + expect(deps.grantPermission).toHaveBeenCalledWith({ + principalType: 'public', + principalId: null, + resourceType: 'promptGroup', + resourceId: 'group-1', + accessRoleId: 'promptGroup_viewer', + grantedBy: '000000000000000000000000', + }); + expect(deps.deletePromptGroup).not.toHaveBeenCalled(); + expect(deps.lock.release).toHaveBeenCalledTimes(1); + }); + + it('counts an existing group as unchanged without creating or granting', async () => { + const deps = createDeps({ + findLangfusePromptGroup: jest.fn().mockResolvedValue({ _id: 'existing-id' }), + }); + const run = createLangfusePromptSyncRunner(deps); + + const result = await run(createRequest()); + + expect(result).toEqual({ + ok: true, + value: { created: 0, unchanged: 1, skipped: 0, complete: true }, + }); + expect(deps.createLangfusePromptGroup).not.toHaveBeenCalled(); + expect(deps.grantPermission).not.toHaveBeenCalled(); + }); + + it('maps a duplicate-key error (code 11000) on create to unchanged without granting', async () => { + const duplicateError = Object.assign(new Error('duplicate key'), { code: 11000 }); + const deps = createDeps({ + createLangfusePromptGroup: jest.fn().mockRejectedValue(duplicateError), + }); + const run = createLangfusePromptSyncRunner(deps); + + const result = await run(createRequest()); + + expect(result).toEqual({ + ok: true, + value: { created: 0, unchanged: 1, skipped: 0, complete: true }, + }); + expect(deps.grantPermission).not.toHaveBeenCalled(); + expect(deps.logger.warn).not.toHaveBeenCalled(); + }); + + it('maps a ValidationError on create to skipped with a warning', async () => { + const validationError = new Error('invalid document'); + validationError.name = 'ValidationError'; + const deps = createDeps({ + createLangfusePromptGroup: jest.fn().mockRejectedValue(validationError), + }); + const run = createLangfusePromptSyncRunner(deps); + + const result = await run(createRequest()); + + expect(result).toEqual({ + ok: true, + value: { created: 0, unchanged: 0, skipped: 1, complete: true }, + }); + expect(deps.grantPermission).not.toHaveBeenCalled(); + expect(deps.logger.warn).toHaveBeenCalledTimes(1); + expect(deps.logger.warn).toHaveBeenCalledWith( + expect.stringContaining('failed validation'), + 'greeting', + validationError, + ); + }); + + it('rethrows an unexpected create error, which is neither a duplicate key nor a validation failure', async () => { + const deps = createDeps({ + createLangfusePromptGroup: jest.fn().mockRejectedValue(new Error('boom')), + }); + const run = createLangfusePromptSyncRunner(deps); + + await expect(run(createRequest())).rejects.toThrow('boom'); + expect(deps.grantPermission).not.toHaveBeenCalled(); + expect(deps.lock.release).toHaveBeenCalledTimes(1); + }); + + it('deletes the created group and rethrows when the grant fails', async () => { + const grantError = new Error('grant failed'); + const deps = createDeps({ grantPermission: jest.fn().mockRejectedValue(grantError) }); + const run = createLangfusePromptSyncRunner(deps); + + await expect(run(createRequest())).rejects.toBe(grantError); + expect(deps.deletePromptGroup).toHaveBeenCalledWith({ _id: 'group-1' }); + expect(deps.logger.error).toHaveBeenCalledWith( + expect.stringContaining('Could not grant public access'), + grantError, + ); + expect(deps.lock.release).toHaveBeenCalledTimes(1); + }); + + it('acquires and extends the lock with the per-request lease', async () => { + const deps = createDeps(); + const run = createLangfusePromptSyncRunner(deps); + const listPrompts = jest + .fn() + .mockResolvedValueOnce(listResult([textItem('a')], 2, 1)) + .mockResolvedValueOnce(listResult([textItem('b')], 2, 2)); + + await run(createRequest({ listPrompts, lockLeaseMs: 120_000 })); + + expect(deps.lock.acquire).toHaveBeenCalledWith('prompt-sync:langfuse:t1', 120_000); + expect(deps.lock.extend).toHaveBeenCalledWith('prompt-sync:langfuse:t1', 'token-1', 120_000); + }); + + it('stops paging at the per-request maxPages', async () => { + const deps = createDeps(); + const run = createLangfusePromptSyncRunner(deps); + const listPrompts = jest + .fn() + .mockResolvedValueOnce(listResult([textItem('a')], 5, 1)) + .mockResolvedValueOnce(listResult([textItem('b')], 5, 2)) + .mockResolvedValueOnce(listResult([textItem('c')], 5, 3)); + + const result = await run(createRequest({ listPrompts, maxPages: 2 })); + + expect(result).toEqual({ + ok: true, + value: { created: 2, unchanged: 0, skipped: 0, complete: false }, + }); + expect(listPrompts).toHaveBeenCalledTimes(2); + expect(listPrompts).toHaveBeenNthCalledWith(1, { type: 'text', limit: 100, page: 1 }); + expect(listPrompts).toHaveBeenNthCalledWith(2, { type: 'text', limit: 100, page: 2 }); + expect(deps.logger.warn).toHaveBeenCalledWith( + expect.stringContaining('page cap'), + expect.objectContaining({ maxPages: 2, totalPages: 5 }), + ); + }); + + it('uses the per-request listPrompts for every page and keeps the lock across requests', async () => { + const deps = createDeps(); + const run = createLangfusePromptSyncRunner(deps); + const first = createRequest({ + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('a')], 1)), + }); + const second = createRequest({ + source: { destination: 'us', projectId: 'project-2' }, + listPrompts: jest.fn().mockResolvedValue(listResult([textItem('b')], 1)), + }); + + await run(first); + await run(second); + + expect(first.listPrompts).toHaveBeenCalledTimes(1); + expect(second.listPrompts).toHaveBeenCalledTimes(1); + expect(deps.lock.acquire).toHaveBeenNthCalledWith(1, 'prompt-sync:langfuse:t1', 300_000); + expect(deps.lock.acquire).toHaveBeenNthCalledWith(2, 'prompt-sync:langfuse:', 300_000); + expect(deps.createLangfusePromptGroup).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + sourceDestination: 'us', + sourceProjectId: 'project-2', + author: makeSourceAuthorId('langfuse', 'project-2'), + }), + ); + }); + + it('returns already_running without listing when the lock is held', async () => { + const deps = createDeps({ + lock: createFakeLock({ acquire: jest.fn().mockResolvedValue(null) }), + }); + const run = createLangfusePromptSyncRunner(deps); + const request = createRequest(); + + const result = await run(request); + + expect(result).toEqual({ ok: false, error: { code: 'already_running' } }); + expect(request.listPrompts).not.toHaveBeenCalled(); + expect(deps.lock.release).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/api/src/langfuse/promptSyncRunner.ts b/packages/api/src/langfuse/promptSyncRunner.ts new file mode 100644 index 00000000000..bc5fcc16feb --- /dev/null +++ b/packages/api/src/langfuse/promptSyncRunner.ts @@ -0,0 +1,100 @@ +import { PrincipalType, ResourceType, AccessRoleIds } from 'librechat-data-provider'; +import type { PromptMethods } from '@librechat/data-schemas'; +import type { + PromptCatalogSyncDeps, + PromptCatalogSyncResult, + PromptCatalogSyncSource, +} from './promptCatalogSync'; +import type { DistributedLock } from '~/cache/lock'; +import { syncLangfusePromptCatalog } from './promptCatalogSync'; +import { CONSTANTS } from '~/mcp/enum'; + +/** Process-lifetime dependencies: the shared lock, the Mongo-backed prompt + * group methods, the ACL grant and the logger. */ +export interface LangfusePromptSyncRunnerDeps { + lock: DistributedLock; + findLangfusePromptGroup: PromptMethods['findLangfusePromptGroup']; + createLangfusePromptGroup: PromptMethods['createLangfusePromptGroup']; + deletePromptGroup: PromptMethods['deletePromptGroup']; + grantPermission: (params: { + principalType: PrincipalType; + principalId: string | null; + resourceType: ResourceType; + resourceId: string; + accessRoleId: string; + grantedBy: string; + }) => Promise; + logger: PromptCatalogSyncDeps['logger']; +} + +/** Per-request inputs: the source identity, a connection-bound page lister + * and the config-resolved lock lease and page cap. */ +export interface LangfusePromptSyncRequest { + source: PromptCatalogSyncSource; + listPrompts: PromptCatalogSyncDeps['listPrompts']; + lockLeaseMs: number; + maxPages: number; +} + +export type LangfusePromptSyncRunner = ( + request: LangfusePromptSyncRequest, +) => Promise; + +/** The AI-2272 partial unique index is defense in depth: this maps the + * resulting duplicate-key race to the runner's `unchanged` outcome. */ +function isDuplicateKeyError(error: unknown): boolean { + return (error as { code?: number } | undefined)?.code === 11000; +} + +/** A document validation failure on `createLangfusePromptGroup`, mapped to the + * runner's `skipped` outcome. */ +function isValidationError(error: unknown): boolean { + return error instanceof Error && error.name === 'ValidationError'; +} + +/** Binds the storage-agnostic `syncLangfusePromptCatalog` to the Mongo-backed + * prompt-group methods and the ACL grant, so the Mongo error predicates live + * here and `promptCatalogSync.ts` stays storage-agnostic. Every new group is + * granted a PUBLIC `PROMPTGROUP_VIEWER` entry by the system user; a failed + * grant deletes the group again before the error propagates. */ +export function createLangfusePromptSyncRunner( + deps: LangfusePromptSyncRunnerDeps, +): LangfusePromptSyncRunner { + const { + lock, + findLangfusePromptGroup, + createLangfusePromptGroup, + deletePromptGroup, + grantPermission, + logger, + } = deps; + + return async function syncPromptCatalog({ source, listPrompts, lockLeaseMs, maxPages }) { + const syncDeps: PromptCatalogSyncDeps = { + listPrompts, + findPromptGroup: findLangfusePromptGroup, + createPromptGroup: createLangfusePromptGroup, + isDuplicateKeyError, + isValidationError, + grantPublicViewer: async (groupId) => { + await grantPermission({ + principalType: PrincipalType.PUBLIC, + principalId: null, + resourceType: ResourceType.PROMPTGROUP, + resourceId: groupId, + accessRoleId: AccessRoleIds.PROMPTGROUP_VIEWER, + grantedBy: CONSTANTS.SYSTEM_USER_ID, + }); + }, + deletePromptGroup: async (groupId) => { + await deletePromptGroup({ _id: groupId }); + }, + lock, + lockLeaseMs, + maxPages, + logger, + }; + + return syncLangfusePromptCatalog(source, syncDeps); + }; +} diff --git a/packages/api/src/langfuse/prompts.spec.ts b/packages/api/src/langfuse/prompts.spec.ts index d5b8b520052..3e54cf43d70 100644 --- a/packages/api/src/langfuse/prompts.spec.ts +++ b/packages/api/src/langfuse/prompts.spec.ts @@ -146,6 +146,25 @@ describe('listLangfusePrompts', () => { expect(url).toBe('https://cloud.langfuse.test/api/public/v2/prompts'); }); + it('sends a text filter as URL-encoded JSON that decodes to the exact array', async () => { + const conn = makeConnection(); + fetchMock.mockResolvedValue( + jsonResponse({ data: [], meta: { page: 1, limit: 100, totalItems: 0, totalPages: 0 } }), + ); + + await listLangfusePrompts(conn, { type: 'text', limit: 100, page: 1 }, { timeoutMs: 5000 }); + + const [url] = fetchMock.mock.calls[0]; + const parsedUrl = new URL(url as string); + expect(parsedUrl.searchParams.get('limit')).toBe('100'); + expect(parsedUrl.searchParams.get('page')).toBe('1'); + const filter = parsedUrl.searchParams.get('filter'); + expect(filter).not.toBeNull(); + expect(JSON.parse(filter as string)).toEqual([ + { type: 'string', column: 'type', operator: '=', value: 'text' }, + ]); + }); + it('sends tag, fromUpdatedAt and toUpdatedAt as given', async () => { const conn = makeConnection(); fetchMock.mockResolvedValue( diff --git a/packages/api/src/langfuse/prompts.ts b/packages/api/src/langfuse/prompts.ts index ace4628e43a..3aaff7cf425 100644 --- a/packages/api/src/langfuse/prompts.ts +++ b/packages/api/src/langfuse/prompts.ts @@ -26,6 +26,7 @@ export type LangfusePromptListQuery = { name?: string; label?: string; tag?: string; + type?: 'text' | 'chat'; page?: number; limit?: number; fromUpdatedAt?: string; @@ -239,8 +240,10 @@ function appendDefined( /** * Lists prompt metadata from Langfuse. Query params are sent as-is, only those - * defined; `filter` is never sent (deferred to the sync slice), and every - * prompt type — including `chat` — is returned. + * defined; every prompt type — including `chat` — is returned unless `type` + * is given. `type` is not a Langfuse query param itself: it is serialized as + * the `filter` parameter, the shape the Langfuse list endpoint expects for a + * column filter. `filter` is omitted entirely when `type` is unset. */ export async function listLangfusePrompts( conn: LangfusePromptConnection, @@ -255,6 +258,12 @@ export async function listLangfusePrompts( appendDefined(params, 'limit', query.limit); appendDefined(params, 'fromUpdatedAt', query.fromUpdatedAt); appendDefined(params, 'toUpdatedAt', query.toUpdatedAt); + if (query.type != null) { + params.set( + 'filter', + JSON.stringify([{ type: 'string', column: 'type', operator: '=', value: query.type }]), + ); + } const search = params.size > 0 ? `?${params.toString()}` : ''; const url = `${conn.baseUrl.replace(/\/+$/, '')}${PROMPTS_PATH}${search}`; diff --git a/packages/api/src/prompts/migration.spec.ts b/packages/api/src/prompts/migration.spec.ts new file mode 100644 index 00000000000..8fd84ccd82d --- /dev/null +++ b/packages/api/src/prompts/migration.spec.ts @@ -0,0 +1,122 @@ +import mongoose, { Types } from 'mongoose'; +import { MongoMemoryServer } from 'mongodb-memory-server'; +import { createModels, createMethods } from '@librechat/data-schemas'; +import { + AccessRoleIds, + PrincipalType, + PrincipalModel, + ResourceType, +} from 'librechat-data-provider'; +import type { IPromptGroupDocument } from '@librechat/data-schemas'; +import type { Model } from 'mongoose'; +import { checkPromptPermissionsMigration } from './migration'; + +let mongo: MongoMemoryServer; +let methods: ReturnType; +let PromptGroupModel: Model; + +beforeAll(async () => { + mongo = await MongoMemoryServer.create({ instance: { args: ['--nounixsocket'] } }); + await mongoose.connect(mongo.getUri()); + createModels(mongoose); + methods = createMethods(mongoose); + await methods.seedDefaultRoles(); + PromptGroupModel = mongoose.models.PromptGroup as Model; +}); + +afterEach(async () => { + await mongoose.models.PromptGroup.deleteMany({}); + await mongoose.models.AclEntry.deleteMany({}); +}); + +afterAll(async () => { + await mongoose.disconnect(); + await mongo?.stop(); +}); + +function runCheck() { + return checkPromptPermissionsMigration({ methods, mongoose, PromptGroupModel }); +} + +describe('checkPromptPermissionsMigration', () => { + it('counts a native group without a USER ACL entry', async () => { + const group = await PromptGroupModel.create({ + name: 'Native group', + author: new Types.ObjectId(), + authorName: 'Author', + productionId: new Types.ObjectId(), + }); + + const result = await runCheck(); + + expect(result.totalToMigrate).toBe(1); + expect(result.privateGroups).toBe(1); + expect(result.details?.privateGroups).toEqual([ + { name: 'Native group', _id: group._id.toString(), category: 'uncategorized' }, + ]); + }); + + it('does not count a Langfuse-source group', async () => { + await PromptGroupModel.create({ + name: 'Langfuse group', + author: new Types.ObjectId(), + authorName: 'Langfuse Sync', + source: 'langfuse', + sourcePromptName: 'langfuse-prompt', + sourceProjectId: 'project-1', + sourceDestination: 'destination-1', + }); + + const result = await runCheck(); + + expect(result.totalToMigrate).toBe(0); + expect(result.details).toBeUndefined(); + }); + + it('counts a legacy group with no source field, inserted without the schema default', async () => { + const db = mongoose.connection.db; + if (!db) { + throw new Error('No database connection'); + } + const inserted = await db.collection('promptgroups').insertOne({ + name: 'Legacy group', + author: new Types.ObjectId(), + authorName: 'Legacy author', + }); + const stored = await db + .collection('promptgroups') + .findOne({ _id: inserted.insertedId }, { projection: { source: 1 } }); + expect(stored).not.toHaveProperty('source'); + + const result = await runCheck(); + + expect(result.totalToMigrate).toBe(1); + expect(result.details?.privateGroups).toEqual([ + { name: 'Legacy group', _id: inserted.insertedId.toString(), category: 'uncategorized' }, + ]); + }); + + it('excludes a native group that already has a USER ACL entry', async () => { + const author = new Types.ObjectId(); + const group = await PromptGroupModel.create({ + name: 'Migrated group', + author, + authorName: 'Author', + productionId: new Types.ObjectId(), + }); + const ownerRole = await methods.findRoleByIdentifier(AccessRoleIds.PROMPTGROUP_OWNER); + await mongoose.models.AclEntry.create({ + principalType: PrincipalType.USER, + principalId: author, + principalModel: PrincipalModel.USER, + resourceType: ResourceType.PROMPTGROUP, + resourceId: group._id, + roleId: ownerRole?._id, + grantedBy: author, + }); + + const result = await runCheck(); + + expect(result.totalToMigrate).toBe(0); + }); +}); diff --git a/packages/api/src/prompts/migration.ts b/packages/api/src/prompts/migration.ts index be2b32e26dc..e0680da7082 100644 --- a/packages/api/src/prompts/migration.ts +++ b/packages/api/src/prompts/migration.ts @@ -1,8 +1,8 @@ import { logger } from '@librechat/data-schemas'; import { AccessRoleIds, ResourceType, PrincipalType } from 'librechat-data-provider'; -import { ensureRequiredCollectionsExist } from '../db/utils'; import type { AccessRoleMethods, IPromptGroupDocument } from '@librechat/data-schemas'; import type { Model, Mongoose } from 'mongoose'; +import { ensureRequiredCollectionsExist } from '../db/utils'; const GLOBAL_PROJECT_NAME = 'instance'; @@ -85,6 +85,7 @@ export async function checkPromptPermissionsMigration({ const promptGroupsToMigrate = (await PromptGroupModel.find({ _id: { $nin: migratedGroupIds }, author: { $exists: true, $ne: null }, + source: { $in: [null, 'native'] }, }) .select('_id name author authorName category') .lean()) as unknown as PromptGroupMigrationData[]; diff --git a/packages/api/src/user/erasure.spec.ts b/packages/api/src/user/erasure.spec.ts index 901dd492321..da3937a3136 100644 --- a/packages/api/src/user/erasure.spec.ts +++ b/packages/api/src/user/erasure.spec.ts @@ -499,6 +499,17 @@ describe('eraseUserData', () => { expect(await count('File', { user: userId })).toBe(0); }); + it('calls deleteUserPrompts with the user id during erasure', async () => { + const userId = new mongoose.Types.ObjectId(); + await seedUser(userId); + const { deps } = createDeps(); + const deleteUserPrompts = jest.fn(async () => undefined); + + await eraseUserData({ userId: userId.toString() }, { ...deps, deleteUserPrompts }); + + expect(deleteUserPrompts).toHaveBeenCalledWith(userId.toString()); + }); + for (const modelName of ['Agent', 'PromptGroup']) { it(`keeps the account erasable when the ${modelName} lookup fails`, async () => { const userId = new mongoose.Types.ObjectId(); diff --git a/packages/data-provider/src/api-endpoints.ts b/packages/data-provider/src/api-endpoints.ts index e281f0c3104..2fea074752e 100644 --- a/packages/data-provider/src/api-endpoints.ts +++ b/packages/data-provider/src/api-endpoints.ts @@ -604,6 +604,7 @@ export const adminLangfuseSessionLink = (conversationId: string) => export const adminLangfusePromptSync = () => `${BASE_URL}/api/admin/langfuse/prompt-sync`; export const adminLangfusePrompts = (params: t.TLangfusePromptListParams = {}) => `${BASE_URL}/api/admin/langfuse/prompts${buildQuery(params)}`; +export const adminLangfuseSyncPrompts = () => `${BASE_URL}/api/admin/langfuse/prompts/sync`; export const adminLangfusePrompt = (name: string, version?: number) => `${BASE_URL}/api/admin/langfuse/prompts/${encodeURIComponent(name)}${buildQuery({ version })}`; diff --git a/packages/data-provider/src/config.ts b/packages/data-provider/src/config.ts index e002bfe3151..59be86ea64c 100644 --- a/packages/data-provider/src/config.ts +++ b/packages/data-provider/src/config.ts @@ -3614,6 +3614,10 @@ export type TStartupConfig = { langfuseFanoutEnabled?: boolean; langfuseConnectionAccess?: boolean; langfusePromptSyncAccess?: boolean; + /** Whether this admin also holds `manage:prompts`, the capability the server's + * `POST /prompts/sync` route requires in addition to Langfuse settings access. Gates the + * "Sync prompts" action in the connection tab; absent (falsy) for an older server. */ + langfusePromptManageAccess?: boolean; insightsEnabled?: boolean; /** Whether pull requests are active (on unless turned off, with a token and a repository scope); * the header does not ask for a pull request without it. */ @@ -4268,10 +4272,17 @@ export const langfuseConfigSchema = z.object({ /** Trace user identity and allowlisted user/request metadata. */ trace: langfuseTraceConfigSchema.optional(), /** Tenant admin switch for Langfuse prompt sync. Effective only when the - * deployment sets LANGFUSE_PROMPT_SYNC_AVAILABLE. */ + * deployment sets LANGFUSE_PROMPT_SYNC_AVAILABLE. + * - enabled: the tenant admin switch itself. + * - lockLeaseMs: the per-tenant sync lock lease, in milliseconds (default + * 300000, clamped to 60000-3600000). + * - maxPages: the page cap for one sync request, in pages of up to 100 + * Langfuse prompts each (default 50, clamped to 1-1000). */ promptSync: z .object({ enabled: z.boolean().optional(), + lockLeaseMs: z.number().int().min(60000).max(3600000).optional(), + maxPages: z.number().int().min(1).max(1000).optional(), }) .optional(), }); diff --git a/packages/data-provider/src/data-service.ts b/packages/data-provider/src/data-service.ts index af7a8cccaba..253b1b0e750 100644 --- a/packages/data-provider/src/data-service.ts +++ b/packages/data-provider/src/data-service.ts @@ -114,6 +114,10 @@ export function getLangfusePrompt( return request.get(endpoints.adminLangfusePrompt(params.name, params.version)); } +export function syncLangfusePrompts(): Promise { + return request.post(endpoints.adminLangfuseSyncPrompts(), {}); +} + export function revokeUserKey(name: string): Promise { return request.delete(endpoints.revokeUserKey(name)); } diff --git a/packages/data-provider/src/keys.ts b/packages/data-provider/src/keys.ts index 7d2e91f5f19..7e1eda2fcf7 100644 --- a/packages/data-provider/src/keys.ts +++ b/packages/data-provider/src/keys.ts @@ -150,6 +150,7 @@ export enum MutationKeys { updateLangfuseConnection = 'updateLangfuseConnection', testLangfuseConnection = 'testLangfuseConnection', updateLangfusePromptSync = 'updateLangfusePromptSync', + syncLangfusePrompts = 'syncLangfusePrompts', createAgentApiKey = 'createAgentApiKey', deleteAgentApiKey = 'deleteAgentApiKey', fileUpload = 'fileUpload', diff --git a/packages/data-provider/src/types.ts b/packages/data-provider/src/types.ts index d6cceccd7ef..c4bc145da58 100644 --- a/packages/data-provider/src/types.ts +++ b/packages/data-provider/src/types.ts @@ -1509,9 +1509,25 @@ export type TLangfusePromptErrorCode = | 'invalid_response' | 'not_found' | 'unsupported_type' - | 'invalid_request'; + | 'invalid_request' + | 'already_running' + | 'lock_lost' + | 'internal'; export type TLangfusePromptErrorBody = { code: TLangfusePromptErrorCode; status?: number; }; + +export type TLangfusePromptSyncResult = { + created: number; + unchanged: number; + skipped: number; + /** False when the server's page cap stopped the run before every upstream page was read, + * so the counts above are partial. A re-run with the same page limit re-lists from page 1 + * and reads the same pages again; the server's `langfuse.promptSync.maxPages` config must + * be raised to import the rest. Optional for compatibility with an older server that has + * not added this field yet: treat a missing value as `true`, matching that server's only + * possible outcome. */ + complete?: boolean; +}; diff --git a/packages/data-schemas/src/methods/prompt.spec.ts b/packages/data-schemas/src/methods/prompt.spec.ts index 0b3e5adc8b6..73b94da8678 100644 --- a/packages/data-schemas/src/methods/prompt.spec.ts +++ b/packages/data-schemas/src/methods/prompt.spec.ts @@ -1093,3 +1093,250 @@ describe('PromptGroup list projections', () => { expect(fromAccess?.sourceDestination).toBe('eu'); }); }); + +describe('createLangfusePromptGroup and findLangfusePromptGroup', () => { + beforeAll(async () => { + await PromptGroup.syncIndexes(); + }); + + afterEach(async () => { + // Unscoped delete: the tenant-isolation plugin applies no filter outside a + // tenant context, so this clears the fixtures regardless of their tenantId. + await PromptGroup.deleteMany({ source: 'langfuse', sourceProjectId: /^sync-test-/ }); + }); + + const identity = (suffix: string) => ({ + sourceDestination: 'eu', + sourceProjectId: `sync-test-${suffix}`, + sourcePromptName: 'welcome-prompt', + }); + + it('finds nothing before create, then the created id after create', async () => { + const id = identity('find-then-create'); + + await expect(methods.findLangfusePromptGroup(id)).resolves.toBeNull(); + + const created = await methods.createLangfusePromptGroup({ + ...id, + name: 'Welcome Prompt', + author: String(testUsers.owner._id), + authorName: 'Langfuse Sync', + }); + + await expect(methods.findLangfusePromptGroup(id)).resolves.toEqual({ _id: created._id }); + }); + + it('creates a doc with source "langfuse", the given author/authorName, and no productionId', async () => { + const id = identity('doc-fields'); + const author = String(testUsers.owner._id); + + const created = await methods.createLangfusePromptGroup({ + ...id, + name: 'Doc Fields Prompt', + author, + authorName: 'Langfuse Sync', + }); + + const doc = (await PromptGroup.findById(created._id).lean()) as unknown as LeanPromptGroup & { + source?: string; + productionId?: unknown; + }; + expect(doc.source).toBe('langfuse'); + expect(String(doc.author)).toBe(author); + expect(doc.authorName).toBe('Langfuse Sync'); + expect(doc.productionId).toBeUndefined(); + }); + + it('stamps the active tenant on create, and isolates lookups and creates by tenant', async () => { + const id = identity('tenant-isolation'); + const author = String(testUsers.owner._id); + + const createdInA = await tenantStorage.run({ tenantId: 'a' }, () => + methods.createLangfusePromptGroup({ + ...id, + name: 'Tenant A Prompt', + author, + authorName: 'Langfuse Sync', + }), + ); + + const docInA = (await PromptGroup.findById( + createdInA._id, + ).lean()) as unknown as LeanPromptGroup & { tenantId?: string }; + expect(docInA.tenantId).toBe('a'); + + await tenantStorage.run({ tenantId: 'b' }, async () => { + await expect(methods.findLangfusePromptGroup(id)).resolves.toBeNull(); + + const createdInB = await methods.createLangfusePromptGroup({ + ...id, + name: 'Tenant B Prompt', + author, + authorName: 'Langfuse Sync', + }); + + expect(createdInB._id).not.toBe(createdInA._id); + }); + }); + + it('throws a duplicate-key error (code 11000) on a repeat create at the same identity', async () => { + const id = identity('duplicate'); + + await methods.createLangfusePromptGroup({ + ...id, + name: 'Duplicate Prompt', + author: String(testUsers.owner._id), + authorName: 'Langfuse Sync', + }); + + await expect( + methods.createLangfusePromptGroup({ + ...id, + name: 'Duplicate Prompt Again', + author: String(testUsers.owner._id), + authorName: 'Langfuse Sync', + }), + ).rejects.toMatchObject({ code: 11000 }); + }); + + it('returns the source fields from getAllPromptGroups and getListPromptGroupsByAccess', async () => { + const id = identity('listing'); + const created = await methods.createLangfusePromptGroup({ + ...id, + name: 'Listing Sync Prompt', + author: String(testUsers.owner._id), + authorName: 'Langfuse Sync', + }); + + const allGroups = (await methods.getAllPromptGroups({ + name: 'Listing Sync Prompt', + })) as Array>; + const fromGetAll = allGroups.find((g) => String(g._id) === created._id); + expect(fromGetAll).toMatchObject({ source: 'langfuse', ...id }); + + const byAccess = await methods.getListPromptGroupsByAccess({ accessibleIds: [created._id] }); + const fromAccess = byAccess.data.find((g) => String(g._id) === created._id); + expect(fromAccess).toMatchObject({ source: 'langfuse', ...id }); + }); + + it('serves a lookup with no tenant context via an index scan, not a collection scan', async () => { + const id = identity('ixscan'); + await methods.createLangfusePromptGroup({ + ...id, + name: 'IXSCAN Prompt', + author: String(testUsers.owner._id), + authorName: 'Langfuse Sync', + }); + + const explain = (await PromptGroup.collection + .find(id) + .explain('queryPlanner')) as unknown as Record; + + const hasIndexScan = (stage: unknown): boolean => { + if (stage == null || typeof stage !== 'object') { + return false; + } + const node = stage as Record; + if (node.stage === 'IXSCAN') { + return true; + } + const children = [ + node.inputStage, + node.queryPlan, + ...(Array.isArray(node.inputStages) ? node.inputStages : []), + ]; + return children.some(hasIndexScan); + }; + + const queryPlanner = explain.queryPlanner as Record; + const winningPlan = queryPlanner.winningPlan as Record; + expect(hasIndexScan(winningPlan)).toBe(true); + }); +}); + +describe('deleteUserPrompts and synced groups', () => { + afterEach(async () => { + await AclEntry.deleteMany({}); + await PromptGroup.deleteMany({}); + await Prompt.deleteMany({}); + }); + + it('deletes a native group the admin solely owns, but leaves a Langfuse-synced group and its ACL intact', async () => { + const admin = ( + await User.create({ + name: 'Sync Admin', + email: 'sync-admin@example.com', + role: SystemRoles.ADMIN, + }) + ).toObject() as unknown as LeanUser; + + const nativeGroup = ( + await PromptGroup.create({ + name: 'Admin Native Group', + author: admin._id, + authorName: admin.name, + productionId: new ObjectId(), + }) + ).toObject() as unknown as LeanPromptGroup; + + await grantPermission({ + principalType: PrincipalType.USER, + principalId: admin._id, + resourceType: ResourceType.PROMPTGROUP, + resourceId: nativeGroup._id, + accessRoleId: AccessRoleIds.PROMPTGROUP_OWNER, + grantedBy: admin._id, + }); + + // Synthetic author distinct from the admin, as `makeSourceAuthorId` produces. + const syntheticAuthor = new ObjectId(); + const langfuseGroup = await methods.createLangfusePromptGroup({ + name: 'Synced Prompt', + author: String(syntheticAuthor), + authorName: 'Langfuse Sync', + sourceDestination: 'eu', + sourceProjectId: 'sync-test-delete-user', + sourcePromptName: 'synced-prompt', + }); + + await aclMethods.grantPermission( + PrincipalType.PUBLIC, + null, + ResourceType.PROMPTGROUP, + langfuseGroup._id, + PermissionBits.VIEW, + syntheticAuthor, + ); + + await methods.deleteUserPrompts(String(admin._id)); + + await expect(PromptGroup.findById(nativeGroup._id).lean()).resolves.toBeNull(); + const nativeAcl = await AclEntry.find({ + resourceType: ResourceType.PROMPTGROUP, + resourceId: nativeGroup._id, + }).lean(); + expect(nativeAcl).toHaveLength(0); + + const survivingGroup = await PromptGroup.findById(langfuseGroup._id).lean(); + expect(survivingGroup).not.toBeNull(); + + const survivingAcl = (await AclEntry.find({ + resourceType: ResourceType.PROMPTGROUP, + resourceId: langfuseGroup._id, + }).lean()) as unknown as LeanAclEntry[]; + expect(survivingAcl).toHaveLength(1); + expect(survivingAcl[0]).toMatchObject({ + principalType: PrincipalType.PUBLIC, + permBits: PermissionBits.VIEW, + }); + + /** + * Documented, not changed: `deleteUserPrompts` only removes groups the user + * authored, or solely owns via ACL. A synced group's author is the synthetic + * Langfuse identity, not an admin, so running the sync — or deleting the admin + * who ran it — never deletes it. If an admin were to grant themselves OWNER on + * this same synced group, they would become its sole owner, and it would then + * be deleted like any native group. + */ + }); +}); diff --git a/packages/data-schemas/src/methods/prompt.ts b/packages/data-schemas/src/methods/prompt.ts index 388ffd5b029..335844619e7 100644 --- a/packages/data-schemas/src/methods/prompt.ts +++ b/packages/data-schemas/src/methods/prompt.ts @@ -165,6 +165,19 @@ export interface PromptMethods { ): Promise; makePromptProduction(promptId: string): Promise<{ message: string }>; updatePromptLabels(_id: string, labels: unknown): Promise<{ message: string }>; + findLangfusePromptGroup(identity: { + sourceDestination: string; + sourceProjectId: string; + sourcePromptName: string; + }): Promise<{ _id: string } | null>; + createLangfusePromptGroup(input: { + name: string; + author: string; + authorName: string; + sourceDestination: string; + sourceProjectId: string; + sourcePromptName: string; + }): Promise<{ _id: string }>; } export function createPromptMethods( @@ -578,6 +591,67 @@ export function createPromptMethods( } } + /** + * Finds the pointer-only PromptGroup mirrored from a Langfuse prompt, if one + * already exists for this identity. `null` means no prior sync found it, not + * a query failure — the caller lets a database error propagate. + */ + async function findLangfusePromptGroup({ + sourceDestination, + sourceProjectId, + sourcePromptName, + }: { + sourceDestination: string; + sourceProjectId: string; + sourcePromptName: string; + }): Promise<{ _id: string } | null> { + const PromptGroup = mongoose.models.PromptGroup as Model; + const doc = await PromptGroup.findOne( + { source: 'langfuse', sourceDestination, sourceProjectId, sourcePromptName }, + { _id: 1 }, + ).lean<{ _id: Types.ObjectId }>(); + return doc ? { _id: String(doc._id) } : null; + } + + /** + * Creates a pointer-only PromptGroup mirrored from a Langfuse prompt: name and + * source identity only, never content. No `productionId` — a synced group has + * no production prompt. No `tenantId` here; the tenant-isolation plugin stamps + * it from the active request context. Throws on failure, including a + * duplicate-key error from the partial unique source-identity index, which the + * caller maps to `unchanged`. + */ + async function createLangfusePromptGroup({ + name, + author, + authorName, + sourceDestination, + sourceProjectId, + sourcePromptName, + }: { + name: string; + author: string; + authorName: string; + sourceDestination: string; + sourceProjectId: string; + sourcePromptName: string; + }): Promise<{ _id: string }> { + const PromptGroup = mongoose.models.PromptGroup as Model; + const doc = await PromptGroup.create({ + source: 'langfuse', + sourceDestination, + sourceProjectId, + sourcePromptName, + name, + author: new ObjectId(author), + authorName, + category: '', + oneliner: '', + numberOfGenerations: 0, + }); + return { _id: String(doc._id) }; + } + /** * Save a prompt. */ @@ -1078,6 +1152,8 @@ export function createPromptMethods( getListPromptGroupsByAccess, incrementPromptGroupUsage, createPromptGroup, + findLangfusePromptGroup, + createLangfusePromptGroup, savePrompt, getPrompts, getPrompt, diff --git a/packages/data-schemas/src/schema/promptGroup.ts b/packages/data-schemas/src/schema/promptGroup.ts index 3fcdbb493d7..745d46e87bd 100644 --- a/packages/data-schemas/src/schema/promptGroup.ts +++ b/packages/data-schemas/src/schema/promptGroup.ts @@ -105,5 +105,10 @@ promptGroupSchema.index( partialFilterExpression: { sourcePromptName: { $exists: true } }, }, ); +// Serves the same lookup when no tenant context is active (env-mode sync, tenantId unset). +promptGroupSchema.index( + { sourceDestination: 1, sourceProjectId: 1, sourcePromptName: 1 }, + { partialFilterExpression: { sourcePromptName: { $exists: true } } }, +); export default promptGroupSchema;