diff --git a/packages/api/src/langfuse/destinations.ts b/packages/api/src/langfuse/destinations.ts index 10dc1adf2d..ac5f6a7d36 100644 --- a/packages/api/src/langfuse/destinations.ts +++ b/packages/api/src/langfuse/destinations.ts @@ -30,6 +30,16 @@ export type LangfuseScoreDestination = { authorization: string; }; +export type LangfuseScoreDestinationOptions = { + waitForCentralProjectId?: boolean; + /** + * Deployments that route traces per tenant may want a turn's spans to reach + * only the tenant destination. Setting this false drops the central project + * from resolution without disturbing tenant or connection destinations. + */ + centralTraceExportEnabled?: boolean; +}; + export function getLangfuseDestinationId(baseUrl: string, projectId: string): string { return createHash('sha256') .update(`${baseUrl.replace(/\/+$/, '')}\n${projectId}`) @@ -204,7 +214,10 @@ export async function getScoreDestinations( appConfig: AppConfig | undefined, traceId: string, sampled?: boolean, - options?: { waitForCentralProjectId?: boolean }, + { + waitForCentralProjectId = true, + centralTraceExportEnabled = true, + }: LangfuseScoreDestinationOptions = {}, ): Promise { if ( !isLangfuseTracingEnabled() || @@ -215,8 +228,15 @@ export async function getScoreDestinations( } if (!usesLangfuseMultiTenantRouting()) { + /** Mirrors `resolveLangfuseExportPlan`: without fanout there is no tenant + * route to fall back on, so suppressing central export disables the trace + * outright. Capturing a destination here would let later feedback reach a + * project the trace never went to. */ + if (!centralTraceExportEnabled) { + return []; + } return hasLangfuseEnvCredentials() - ? [await getCentralScoreDestination(options?.waitForCentralProjectId !== false)].filter( + ? [await getCentralScoreDestination(waitForCentralProjectId)].filter( (destination): destination is LangfuseScoreDestination => Boolean(destination), ) : [getConfiguredScoreDestination(appConfig)].filter( @@ -225,7 +245,9 @@ export async function getScoreDestinations( } const destinations = [ - await getCentralScoreDestination(options?.waitForCentralProjectId !== false), + centralTraceExportEnabled + ? await getCentralScoreDestination(waitForCentralProjectId) + : undefined, getTenantScoreDestination(appConfig), ].filter((destination): destination is LangfuseScoreDestination => Boolean(destination)); const unique = new Map(); @@ -251,11 +273,21 @@ export async function getLangfuseTraceDestinationIds( appConfig: AppConfig | undefined, traceId: string, sampled?: boolean, + { + centralTraceExportEnabled = true, + }: Pick = {}, ): Promise { const destinations = await getScoreDestinations(appConfig, traceId, sampled, { waitForCentralProjectId: false, + centralTraceExportEnabled, }); if (destinations.some(({ id }) => id == null)) { + /** A tenant destination's `projectId` is optional, so an eligible project can + * have no stable id to record. `undefined` defers to whatever policy the + * feedback path resolves — an empty list would instead reject every + * destination, silently dropping feedback the tenant should receive. + * `sendFeedbackScore` must be given the same `centralTraceExportEnabled` + * for that deferral to honor a central opt-out. */ return undefined; } return destinations.map(({ id }) => id as string); @@ -264,6 +296,9 @@ export async function getLangfuseTraceDestinationIds( export async function getLangfuseTraceMessageFields( appConfig: AppConfig | undefined, messageId: string, + { + centralTraceExportEnabled = true, + }: Pick = {}, ): Promise<{ langfuseSampled: boolean; langfuseDestinationIds?: string[] }> { const traceId = traceIdForMessage(messageId); const langfuseSampled = isLangfuseTraceSampled(traceId); @@ -273,6 +308,7 @@ export async function getLangfuseTraceMessageFields( appConfig, traceId, langfuseSampled, + { centralTraceExportEnabled }, ), }; } diff --git a/packages/api/src/langfuse/feedback.spec.ts b/packages/api/src/langfuse/feedback.spec.ts index e114cbc967..9c4be391b1 100644 --- a/packages/api/src/langfuse/feedback.spec.ts +++ b/packages/api/src/langfuse/feedback.spec.ts @@ -1203,4 +1203,85 @@ describe('Langfuse feedback scores', () => { expect(getFetchMock()).not.toHaveBeenCalled(); }); + it('captures no destinations when central export is suppressed without a fanout route', async () => { + delete process.env.LANGFUSE_PUBLIC_KEY; + delete process.env.LANGFUSE_SECRET_KEY; + process.env.LANGFUSE_FANOUT_TENANT_DESTINATIONS = 'eu=http://tenant-langfuse:3000'; + await loadFeedback(); + const { getLangfuseTraceDestinationIds } = await import('./destinations'); + + /** `resolveLangfuseExportPlan` reports `disabled` for this shape — no fanout + * route to fall back on — so capturing the configured connection would let + * later feedback reach a project the trace never went to. */ + await expect( + getLangfuseTraceDestinationIds( + appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: encryptedTenantSecret(), + destination: 'eu', + }), + 'trace-id', + true, + { centralTraceExportEnabled: false }, + ), + ).resolves.toEqual([]); + }); + + it('records no restriction when a suppressed-central trace has no identifiable destination', async () => { + enableTenantFanout(); + process.env.LANGFUSE_FANOUT_TENANT_DESTINATIONS = 'eu=http://tenant-langfuse:3000'; + await loadFeedback(); + const { getLangfuseTraceDestinationIds } = await import('./destinations'); + + /** Tenant `projectId` is optional, so this eligible destination carries no + * stable id. An empty list would reject it at feedback time and drop the + * rating; the opt-out is re-asserted through `sendFeedbackScore` instead. */ + await expect( + getLangfuseTraceDestinationIds( + appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: encryptedTenantSecret(), + destination: 'eu', + projectId: undefined, + }), + 'trace-id', + true, + { centralTraceExportEnabled: false }, + ), + ).resolves.toBeUndefined(); + }); + + it('sends unrestricted suppressed-central feedback to the tenant only', async () => { + enableTenantFanout(); + process.env.LANGFUSE_BASE_URL = 'http://central-langfuse:3000'; + process.env.LANGFUSE_FANOUT_TENANT_DESTINATIONS = 'eu=http://tenant-langfuse:3000'; + const { sendFeedbackScore } = await loadFeedback(); + + /** The trace reached the tenant through its fanout route while central export + * was suppressed, and left no destination ids to filter on. The rating has to + * follow the same policy: tenant receives it, central stays excluded. */ + await sendFeedbackScore({ + traceId: 'trace-id', + feedback: { rating: 'thumbsUp' }, + destinationIds: undefined, + centralTraceExportEnabled: false, + appConfig: appConfigWithLangfuse({ + publicKey: 'tenant-public-key', + secretKey: encryptedTenantSecret(), + destination: 'eu', + projectId: undefined, + }), + }); + + expect(getFetchMock()).toHaveBeenCalledTimes(1); + expect(getFetchMock()).toHaveBeenCalledWith( + 'http://tenant-langfuse:3000/api/public/scores', + expect.objectContaining({ + method: 'POST', + headers: expect.objectContaining({ + Authorization: getTenantAuthorization(), + }), + }), + ); + }); }); diff --git a/packages/api/src/langfuse/feedback.ts b/packages/api/src/langfuse/feedback.ts index 6fffa64da4..ca34ac0750 100644 --- a/packages/api/src/langfuse/feedback.ts +++ b/packages/api/src/langfuse/feedback.ts @@ -18,6 +18,13 @@ export type SendFeedbackScoreParams = { metadata?: LangfuseFeedbackMetadata; observationId?: string; appConfig?: AppConfig; + /** + * Must mirror the value used when the trace was generated. `destinationIds` + * only restricts feedback to destinations that carry a stable id, so a caller + * that opts out of central export has to re-assert it here for destinations + * whose project identity is unknown. + */ + centralTraceExportEnabled?: boolean; }; const ENVIRONMENT = process.env.LANGFUSE_TRACING_ENVIRONMENT; @@ -110,15 +117,16 @@ export async function sendFeedbackScore({ metadata = {}, observationId, appConfig, + centralTraceExportEnabled, }: SendFeedbackScoreParams): Promise { if (!traceId) { return; } const destinationIdSet = destinationIds == null ? undefined : new Set(destinationIds); - const destinations = (await getScoreDestinations(appConfig, traceId, sampled)).filter( - ({ id }) => destinationIdSet == null || (id != null && destinationIdSet.has(id)), - ); + const destinations = ( + await getScoreDestinations(appConfig, traceId, sampled, { centralTraceExportEnabled }) + ).filter(({ id }) => destinationIdSet == null || (id != null && destinationIdSet.has(id))); if (destinations.length === 0) { return; }