🚏 feat: Central Trace Destination Opt-Out for Langfuse (#14838)

* feat(langfuse): let callers opt out of the central trace destination

Adds `centralTraceExportEnabled` to the score-destination options and threads
it through `getScoreDestinations`, `getLangfuseTraceDestinationIds` and
`getLangfuseTraceMessageFields`.

Deployments that route traces per tenant may want a given turn's spans to reach
only the tenant destination — for example when central export is a per-tenant
setting rather than a deployment-wide one. Today the central project is included
whenever env credentials exist, with no way for a caller to decline it for a
single trace.

Defaults to `true` everywhere, so existing callers are unaffected: the option is
additive and every current call site resolves exactly as before. The three
public helpers gain an optional trailing parameter and nothing else.

While here, `getScoreDestinations` destructures its options with defaults
instead of repeating `options?.waitForCentralProjectId !== false` at both call
sites, which is what made adding a second flag awkward.

Verified: no new tsc errors (one pre-existing cacheFactory error is unchanged),
99 langfuse tests pass, ESLint and Prettier clean.

* fix(langfuse): keep the central opt-out intact when destinations resolve

Addresses two review findings on the new `centralTraceExportEnabled` option.
Both are cases where opting out of central export was silently discarded by
destination resolution, letting later feedback reach a project the trace never
went to.

1. Non-fanout deployments with no central env credentials still returned the
   configured connection, because only the central-credential branch was gated.
   `resolveLangfuseExportPlan` reports `disabled` for that shape — without a
   fanout route there is nowhere for a central-suppressed trace to go — so
   return no destinations and match it.

2. `getLangfuseTraceDestinationIds` returned `undefined` whenever any
   destination lacked an id, and `sendFeedbackScore` reads `undefined` as
   unrestricted. A tenant destination has no id when its optional `projectId`
   is unset, so a suppressed-central trace could resolve back to the central
   project at feedback time. Fail closed with an empty list, which stays
   restricted, instead.

Both paths now have regression tests, each verified to fail without the
corresponding change: the first returned 3 destination ids, the second returned
`undefined`.

* fix(langfuse): carry the central opt-out into the feedback path

`getLangfuseTraceDestinationIds` returns `undefined` when an eligible
destination has no stable id, which a tenant route hits whenever the
optional `langfuse.projectId` is unset. `sendFeedbackScore` read that as
"unrestricted" and re-resolved destinations with central export enabled,
so a suppressed-central trace still drew central feedback.

Persisting an empty list instead only traded the leak for a drop: the
destination filter rejects every id-less destination, discarding ratings
the tenant should receive. The id list restricts feedback to destinations
that survived reconfiguration; it cannot also encode deployment policy.

Thread `centralTraceExportEnabled` through `sendFeedbackScore` so the
policy is evaluated the same way at trace time and feedback time, and
restore `undefined` for unidentifiable destinations.

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<LangfuseScoreDestination[]> {
   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<string, LangfuseScoreDestination>();
@@ -251,11 +273,21 @@ export async function getLangfuseTraceDestinationIds(
   appConfig: AppConfig | undefined,
   traceId: string,
   sampled?: boolean,
+  {
+    centralTraceExportEnabled = true,
+  }: Pick<LangfuseScoreDestinationOptions, 'centralTraceExportEnabled'> = {},
 ): Promise<string[] | undefined> {
   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<LangfuseScoreDestinationOptions, 'centralTraceExportEnabled'> = {},
 ): 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 },
     ),
   };
 }

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(),
+        }),
+      }),
+    );
+  });
 });

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<void> {
   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;
   }