fix(node): Preserve CallbackManager handlers in LangChain instrumentation (#20849)

## Summary

`augmentCallbackHandlers` in
`packages/node/src/integrations/tracing/langchain/instrumentation.ts`
previously fell into its `typeof handlers === 'object'` branch whenever
`options.callbacks` was a single object and wrapped it into `[handlers,
sentryHandler]`. When that object was a LangChain `CallbackManager`,
downstream code treats the whole manager as one opaque handler — its
inheritable children (notably LangGraph's `StreamMessagesHandler`
installed by `streamMode: ['messages']`, and the LangSmith tracer) are
never unpacked, so per-token streaming events and nested tracing
silently disappear.

This PR detects `CallbackManager` via duck-typing (avoids coupling to a
specific `@langchain/core` resolution) and registers Sentry's handler as
an inheritable child on a copy via `.addHandler(handler, true)`. The
manager's existing children continue to receive `handleLLMNewToken` and
friends.

## Repro

A LangGraph compiled graph + `ChatOpenAI` (or any LangChain provider
with `bindTools(...)`), driven through `@ai-sdk/langchain`'s
`toUIMessageStream`:

```ts
const stream = await graph.stream(
  { messages },
  { streamMode: ['values', 'messages'] },
);
return createUIMessageStreamResponse({ stream: toUIMessageStream(stream) });
```

- **Without this fix**: the SSE output collapses to a single aggregated
`text-delta` at end of run. Probing inside `_streamResponseChunks` shows
the per-token chunks fire `runManager.handleLLMNewToken` correctly, but
the LLM-level run-manager's handlers list contains only
`[CallbackManager, SentryCallbackHandler, langchain_tracer]` —
`StreamMessagesHandler` is missing.
- **With this fix**: every token is delivered as the model produces it
(verified end-to-end against `gpt-5.5` + `useResponsesApi: true` +
`bindTools` and against `gpt-4o-mini` with no tools).

## Tests

Added `packages/node/test/integrations/tracing/langchain.test.ts` with 7
unit tests covering:
- `undefined` callbacks → single-handler array
- pre-existing array → append (idempotent)
- `CallbackManager` → preserves children, copies rather than mutates,
deduplicates
- opaque non-manager object → returned unchanged

## Verify

- [x] Added unit tests for the fixed behaviour
- [x] `yarn build` succeeds
- [x] `yarn test:unit` passes (381/381 in `packages/node`)
- [ ] No related issue currently open — happy to open one if preferred.

---------

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: Andrei Borza <andrei.borza@sentry.io>

Author: 3 people

packages/core/src/shared-exports.ts

@@ -173,6 +173,7 @@ export { instrumentGoogleGenAIClient } from './tracing/google-genai';
 export { GOOGLE_GENAI_INTEGRATION_NAME } from './tracing/google-genai/constants';
 export type { GoogleGenAIResponse } from './tracing/google-genai/types';
 export { createLangChainCallbackHandler, instrumentLangChainEmbeddings } from './tracing/langchain';
+export { _INTERNAL_mergeLangChainCallbackHandler } from './tracing/langchain/utils';
 export { LANGCHAIN_INTEGRATION_NAME } from './tracing/langchain/constants';
 export type { LangChainOptions, LangChainIntegration } from './tracing/langchain/types';
 export { instrumentStateGraphCompile, instrumentCreateReactAgent, instrumentLangGraph } from './tracing/langgraph';

packages/core/src/tracing/langchain/utils.ts

@@ -537,3 +537,51 @@ export function extractToolDefinitions(extraParams?: Record<string, unknown>): s
   });
   return JSON.stringify(toolDefs);
 }
+
+/** Duck-types a LangChain `CallbackManager` (avoids coupling to a specific `@langchain/core` resolution). */
+function isCallbackManager(value: unknown): value is {
+  addHandler: (handler: unknown, inherit?: boolean) => void;
+  copy: () => unknown;
+  handlers?: unknown[];
+} {
+  if (!value || typeof value !== 'object') {
+    return false;
+  }
+  const candidate = value as { addHandler?: unknown; copy?: unknown };
+  return typeof candidate.addHandler === 'function' && typeof candidate.copy === 'function';
+}
+
+function isSentryHandler(handler: unknown): boolean {
+  return typeof handler === 'object' && (handler as Record<string, unknown>)?.name === 'SentryCallbackHandler';
+}
+
+function containsSentryHandler(handlers: unknown[]): boolean {
+  return handlers.some(isSentryHandler);
+}
+
+/**
+ * Merge `sentryHandler` into a given set of LangChain callbacks or callback manager.
+ * @internal Exported for cross-package instrumentation.
+ */
+export function _INTERNAL_mergeLangChainCallbackHandler(existing: unknown, sentryHandler: unknown): unknown {
+  if (!existing) {
+    return [sentryHandler];
+  }
+
+  if (isCallbackManager(existing)) {
+    if (containsSentryHandler(existing.handlers ?? [])) {
+      return existing;
+    }
+
+    const copied = existing.copy() as { addHandler: (handler: unknown, inherit?: boolean) => void };
+    copied.addHandler(sentryHandler, true);
+    return copied;
+  }
+
+  const handlers = Array.isArray(existing) ? existing : [existing];
+  if (containsSentryHandler(handlers)) {
+    return existing;
+  }
+
+  return [...handlers, sentryHandler];
+}

packages/core/src/tracing/langgraph/index.ts

@@ -30,10 +30,10 @@ import {
   extractAgentNameFromParams,
   extractLLMFromParams,
   extractToolsFromCompiledGraph,
-  mergeSentryCallback,
   setResponseAttributes,
   wrapToolsWithSpans,
 } from './utils';
+import { _INTERNAL_mergeLangChainCallbackHandler } from '../langchain/utils';
 
 let _insideCreateReactAgent = false;
 
@@ -179,7 +179,10 @@ function instrumentCompiledGraphInvoke(
                 ...(typeof graphName === 'string' ? { lc_agent_name: graphName } : {}),
               };
 
-              invokeConfig.callbacks = mergeSentryCallback(invokeConfig.callbacks, sentryCallbackHandler);
+              invokeConfig.callbacks = _INTERNAL_mergeLangChainCallbackHandler(
+                invokeConfig.callbacks,
+                sentryCallbackHandler,
+              );
             }
 
             // Extract available tools from the graph instance

packages/core/src/tracing/langgraph/utils.ts

@@ -334,27 +334,3 @@ export function setResponseAttributes(span: Span, inputMessages: LangChainMessag
     span.setAttribute(GEN_AI_USAGE_TOTAL_TOKENS_ATTRIBUTE, totalTokens);
   }
 }
-
-/** Merge `sentryHandler` into a langchain `callbacks` value (`BaseCallbackHandler[]` or `BaseCallbackManager`). */
-export function mergeSentryCallback(existing: unknown, sentryHandler: unknown): unknown {
-  if (!existing) {
-    return [sentryHandler];
-  }
-
-  if (Array.isArray(existing)) {
-    if (existing.includes(sentryHandler)) {
-      return existing;
-    }
-    return [...existing, sentryHandler];
-  }
-
-  const manager = existing as { addHandler?: (h: unknown) => void; handlers?: unknown[] };
-  if (typeof manager.addHandler === 'function') {
-    const alreadyAdded = Array.isArray(manager.handlers) && manager.handlers.includes(sentryHandler);
-    if (!alreadyAdded) {
-      manager.addHandler(sentryHandler);
-    }
-  }
-
-  return existing;
-}

packages/core/test/lib/tracing/langchain-utils.test.ts

@@ -1,7 +1,11 @@
-import { describe, expect, it } from 'vitest';
+import { describe, expect, it, vi } from 'vitest';
 import { GEN_AI_INPUT_MESSAGES_ATTRIBUTE } from '../../../src/tracing/ai/gen-ai-attributes';
 import type { LangChainMessage } from '../../../src/tracing/langchain/types';
-import { extractChatModelRequestAttributes, normalizeLangChainMessages } from '../../../src/tracing/langchain/utils';
+import {
+  _INTERNAL_mergeLangChainCallbackHandler,
+  extractChatModelRequestAttributes,
+  normalizeLangChainMessages,
+} from '../../../src/tracing/langchain/utils';
 
 describe('normalizeLangChainMessages', () => {
   it('normalizes messages with _getType()', () => {
@@ -246,3 +250,88 @@ describe('extractChatModelRequestAttributes with multimodal content', () => {
     expect(inputMessages).toContain('What is in this image?');
   });
 });
+
+describe('_INTERNAL_mergeLangChainCallbackHandler', () => {
+  const sentryHandler = { name: 'SentryCallbackHandler' };
+
+  function makeFakeCallbackManager(existingHandlers: unknown[] = []) {
+    const manager = {
+      handlers: [...existingHandlers],
+      inheritableHandlers: [...existingHandlers],
+      addHandler: vi.fn(function (this: any, handler: unknown, inherit?: boolean) {
+        this.handlers.push(handler);
+        if (inherit !== false) {
+          this.inheritableHandlers.push(handler);
+        }
+      }),
+      copy: vi.fn(function (this: any) {
+        return makeFakeCallbackManager(this.handlers);
+      }),
+    };
+    return manager;
+  }
+
+  it('returns a fresh array when no existing callbacks are present', () => {
+    expect(_INTERNAL_mergeLangChainCallbackHandler(undefined, sentryHandler)).toStrictEqual([sentryHandler]);
+    expect(_INTERNAL_mergeLangChainCallbackHandler(null, sentryHandler)).toStrictEqual([sentryHandler]);
+  });
+
+  it('appends to an existing callbacks array', () => {
+    const userA = { _user: 'A' };
+    const userB = { _user: 'B' };
+    expect(_INTERNAL_mergeLangChainCallbackHandler([userA, userB], sentryHandler)).toStrictEqual([
+      userA,
+      userB,
+      sentryHandler,
+    ]);
+  });
+
+  it('does not duplicate when the sentry handler is already in the array', () => {
+    const userA = { _user: 'A' };
+    const existing = [userA, sentryHandler];
+    expect(_INTERNAL_mergeLangChainCallbackHandler(existing, sentryHandler)).toBe(existing);
+  });
+
+  it('preserves inheritable handlers when callbacks is a CallbackManager', () => {
+    // Reproduces the LangGraph `streamMode: ['messages']` setup: a
+    // CallbackManager carrying a StreamMessagesHandler is passed via
+    // options.callbacks. Wrapping it as `[manager, sentryHandler]` would
+    // drop the manager's inheritable children — instead we register
+    // Sentry on a copy and keep the existing handler chain intact.
+    const streamMessagesHandler = { name: 'StreamMessagesHandler', lc_prefer_streaming: true };
+    const manager = makeFakeCallbackManager([streamMessagesHandler]);
+    const result = _INTERNAL_mergeLangChainCallbackHandler(manager, sentryHandler) as { handlers: unknown[] };
+    expect(Array.isArray(result)).toBe(false);
+    expect(result.handlers).toEqual([streamMessagesHandler, sentryHandler]);
+  });
+
+  it('copies the manager and registers Sentry as an inheritable handler', () => {
+    const manager = makeFakeCallbackManager([]);
+    const result = _INTERNAL_mergeLangChainCallbackHandler(manager, sentryHandler) as {
+      addHandler: ReturnType<typeof vi.fn>;
+      inheritableHandlers: unknown[];
+    };
+    expect(manager.copy).toHaveBeenCalledTimes(1);
+    expect(manager.handlers).toEqual([]);
+    expect(result.addHandler).toHaveBeenCalledWith(sentryHandler, true);
+    expect(result.inheritableHandlers).toEqual([sentryHandler]);
+  });
+
+  it('returns the manager unchanged without copying when it already contains the handler', () => {
+    const manager = makeFakeCallbackManager([sentryHandler]);
+    const result = _INTERNAL_mergeLangChainCallbackHandler(manager, sentryHandler);
+    expect(result).toBe(manager);
+    expect(manager.copy).not.toHaveBeenCalled();
+    expect(manager.addHandler).not.toHaveBeenCalled();
+  });
+
+  it('wraps a lone callback object into an array with the sentry handler', () => {
+    const opaque = { name: 'NotAManager' };
+    expect(_INTERNAL_mergeLangChainCallbackHandler(opaque, sentryHandler)).toStrictEqual([opaque, sentryHandler]);
+  });
+
+  it('returns unchanged when the lone callback object is already a sentry handler', () => {
+    const existing = { name: 'SentryCallbackHandler' };
+    expect(_INTERNAL_mergeLangChainCallbackHandler(existing, sentryHandler)).toBe(existing);
+  });
+});

packages/core/test/lib/utils/langgraph-utils.test.ts

@@ -1,9 +1,5 @@
-import { describe, expect, it, vi } from 'vitest';
-import {
-  extractAgentNameFromParams,
-  extractLLMFromParams,
-  mergeSentryCallback,
-} from '../../../src/tracing/langgraph/utils';
+import { describe, expect, it } from 'vitest';
+import { extractAgentNameFromParams, extractLLMFromParams } from '../../../src/tracing/langgraph/utils';
 
 describe('extractLLMFromParams', () => {
   it('returns null for empty or invalid args', () => {
@@ -44,40 +40,3 @@ describe('extractAgentNameFromParams', () => {
     expect(extractAgentNameFromParams([{ name: 'my_agent' }])).toBe('my_agent');
   });
 });
-
-describe('mergeSentryCallback', () => {
-  const sentryHandler = { _sentry: true };
-
-  it('returns a fresh array when no existing callbacks are present', () => {
-    expect(mergeSentryCallback(undefined, sentryHandler)).toStrictEqual([sentryHandler]);
-    expect(mergeSentryCallback(null, sentryHandler)).toStrictEqual([sentryHandler]);
-  });
-
-  it('appends to an existing callbacks array', () => {
-    const userA = { _user: 'A' };
-    const userB = { _user: 'B' };
-    expect(mergeSentryCallback([userA, userB], sentryHandler)).toStrictEqual([userA, userB, sentryHandler]);
-  });
-
-  it('does not duplicate when the sentry handler is already in the array', () => {
-    const userA = { _user: 'A' };
-    const existing = [userA, sentryHandler];
-    expect(mergeSentryCallback(existing, sentryHandler)).toBe(existing);
-  });
-
-  it('calls addHandler on a CallbackManager-like object', () => {
-    const addHandler = vi.fn();
-    const manager = { addHandler, handlers: [] as unknown[] };
-    const result = mergeSentryCallback(manager, sentryHandler);
-    expect(result).toBe(manager);
-    expect(addHandler).toHaveBeenCalledWith(sentryHandler);
-    expect(addHandler).toHaveBeenCalledTimes(1);
-  });
-
-  it('does not re-add when the manager already has the sentry handler', () => {
-    const addHandler = vi.fn();
-    const manager = { addHandler, handlers: [sentryHandler] };
-    mergeSentryCallback(manager, sentryHandler);
-    expect(addHandler).not.toHaveBeenCalled();
-  });
-});

packages/node/src/integrations/tracing/langchain/instrumentation.ts

@@ -7,6 +7,7 @@ import {
 } from '@opentelemetry/instrumentation';
 import type { LangChainOptions } from '@sentry/core';
 import {
+  _INTERNAL_mergeLangChainCallbackHandler,
   _INTERNAL_skipAiProviderWrapping,
   ANTHROPIC_AI_INTEGRATION_NAME,
   createLangChainCallbackHandler,
@@ -27,34 +28,6 @@ interface PatchedLangChainExports {
   [key: string]: unknown;
 }
 
-/**
- * Augments a callback handler list with Sentry's handler if not already present
- */
-function augmentCallbackHandlers(handlers: unknown, sentryHandler: unknown): unknown {
-  // Handle null/undefined - return array with just our handler
-  if (!handlers) {
-    return [sentryHandler];
-  }
-
-  // If handlers is already an array
-  if (Array.isArray(handlers)) {
-    // Check if our handler is already in the list
-    if (handlers.includes(sentryHandler)) {
-      return handlers;
-    }
-    // Add our handler to the list
-    return [...handlers, sentryHandler];
-  }
-
-  // If it's a single handler object, convert to array
-  if (typeof handlers === 'object') {
-    return [handlers, sentryHandler];
-  }
-
-  // Unknown type - return original
-  return handlers;
-}
-
 /**
  * Wraps Runnable methods (invoke, stream, batch) to inject Sentry callbacks at request time
  * Uses a Proxy to intercept method calls and augment the options.callbacks
@@ -82,9 +55,7 @@ function wrapRunnableMethod(
       }
 
       // Inject our callback handler into options.callbacks (request time callbacks)
-      const existingCallbacks = options.callbacks;
-      const augmentedCallbacks = augmentCallbackHandlers(existingCallbacks, sentryHandler);
-      options.callbacks = augmentedCallbacks;
+      options.callbacks = _INTERNAL_mergeLangChainCallbackHandler(options.callbacks, sentryHandler);
 
       // Call original method with augmented options
       return Reflect.apply(target, thisArg, args);