feat(server-utils): Rewrite SentryLangChainInstrumentation to orchestrion

packages/core/src/shared-exports.ts

@@ -216,6 +216,7 @@ export { instrumentStream as instrumentGoogleGenAIStream } from './tracing/googl
 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_getLangChainEmbeddingsSpanOptions } from './tracing/langchain/embeddings';
 export { _INTERNAL_mergeLangChainCallbackHandler } from './tracing/langchain/utils';
 export { LANGCHAIN_INTEGRATION_NAME } from './tracing/langchain/constants';
 export type { LangChainOptions, LangChainIntegration } from './tracing/langchain/types';

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

@@ -55,6 +55,32 @@ function extractEmbeddingAttributes(instance: unknown): Record<string, unknown>
   return attributes;
 }
 
+/**
+ * Builds the span options for a LangChain embedding call from the embeddings instance and input.
+ *
+ * @internal Exported so the diagnostics-channel (orchestrion) instrumentation can build the same
+ * span as the prototype-patching path below.
+ */
+export function _INTERNAL_getLangChainEmbeddingsSpanOptions(
+  instance: unknown,
+  input: unknown,
+  options: LangChainOptions = {},
+): { name: string; op: string; attributes: Record<string, SpanAttributeValue> } {
+  const { recordInputs } = resolveAIRecordingOptions(options);
+  const attributes = extractEmbeddingAttributes(instance);
+  const modelName = attributes[GEN_AI_REQUEST_MODEL_ATTRIBUTE] || 'unknown';
+
+  if (recordInputs && input != null) {
+    attributes[GEN_AI_EMBEDDINGS_INPUT_ATTRIBUTE] = typeof input === 'string' ? input : JSON.stringify(input);
+  }
+
+  return {
+    name: `embeddings ${modelName}`,
+    op: GEN_AI_EMBEDDINGS_OPERATION_ATTRIBUTE,
+    attributes: attributes as Record<string, SpanAttributeValue>,
+  };
+}
+
 /**
  * Wraps a LangChain embedding method (embedQuery or embedDocuments) to create Sentry spans.
  *
@@ -64,35 +90,16 @@ export function instrumentEmbeddingMethod(
   originalMethod: (...args: unknown[]) => Promise<unknown>,
   options: LangChainOptions = {},
 ): (...args: unknown[]) => Promise<unknown> {
-  const { recordInputs } = resolveAIRecordingOptions(options);
-
   return new Proxy(originalMethod, {
     apply(target, thisArg, args: unknown[]): Promise<unknown> {
-      const attributes = extractEmbeddingAttributes(thisArg);
-      const modelName = attributes[GEN_AI_REQUEST_MODEL_ATTRIBUTE] || 'unknown';
-
-      if (recordInputs) {
-        const input = args[0];
-        if (input != null) {
-          attributes[GEN_AI_EMBEDDINGS_INPUT_ATTRIBUTE] = typeof input === 'string' ? input : JSON.stringify(input);
-        }
-      }
-
-      return startSpan(
-        {
-          name: `embeddings ${modelName}`,
-          op: GEN_AI_EMBEDDINGS_OPERATION_ATTRIBUTE,
-          attributes: attributes as Record<string, SpanAttributeValue>,
-        },
-        () => {
-          return Reflect.apply(target, thisArg, args).then(undefined, error => {
-            captureException(error, {
-              mechanism: { handled: false, type: 'auto.ai.langchain' },
-            });
-            throw error;
+      return startSpan(_INTERNAL_getLangChainEmbeddingsSpanOptions(thisArg, args[0], options), () => {
+        return Reflect.apply(target, thisArg, args).then(undefined, error => {
+          captureException(error, {
+            mechanism: { handled: false, type: 'auto.ai.langchain' },
           });
-        },
-      );
+          throw error;
+        });
+      });
     },
   }) as (...args: unknown[]) => Promise<unknown>;
 }

packages/server-utils/src/integrations/tracing-channel/langchain.ts

@@ -0,0 +1,121 @@
+import * as diagnosticsChannel from 'node:diagnostics_channel';
+import type { IntegrationFn, LangChainOptions, Span } from '@sentry/core';
+import {
+  _INTERNAL_getLangChainEmbeddingsSpanOptions,
+  _INTERNAL_mergeLangChainCallbackHandler,
+  _INTERNAL_skipAiProviderWrapping,
+  ANTHROPIC_AI_INTEGRATION_NAME,
+  createLangChainCallbackHandler,
+  debug,
+  defineIntegration,
+  GOOGLE_GENAI_INTEGRATION_NAME,
+  LANGCHAIN_INTEGRATION_NAME,
+  OPENAI_INTEGRATION_NAME,
+  startInactiveSpan,
+  waitForTracingChannelBinding,
+} from '@sentry/core';
+import { DEBUG_BUILD } from '../../debug-build';
+import { CHANNELS } from '../../orchestrion/channels';
+import { langchainEmbeddingsChannels } from '../../orchestrion/config/langchain';
+import { bindTracingChannelToSpan } from '../../tracing-channel';
+
+// Same name as the OTel integration by design: when enabled, the OTel 'LangChain' integration is
+// dropped from the default set (see the Node opt-in loader).
+const INTEGRATION_NAME = LANGCHAIN_INTEGRATION_NAME;
+
+// LangChain drives the underlying AI provider SDKs itself, so while it's active those providers must
+// not also instrument, or every call would produce two spans (mirrors the OTel path's skip list).
+const SKIPPED_PROVIDERS = [OPENAI_INTEGRATION_NAME, ANTHROPIC_AI_INTEGRATION_NAME, GOOGLE_GENAI_INTEGRATION_NAME];
+
+// The chat-model channels carry the live args array of `invoke(input, options)` / `_streamIterator(input, options)`.
+interface RunnableChannelContext {
+  arguments: unknown[];
+}
+
+// The embeddings channels carry the instance (`self`) and the `embedQuery(text)` / `embedDocuments(texts)` args.
+interface EmbeddingsChannelContext {
+  self?: unknown;
+  arguments: unknown[];
+}
+
+let subscribed = false;
+
+// Registered lazily on the first LangChain call (not at `setupOnce`) so a direct provider call made
+// before any LangChain call still gets its own span — matches the OTel patch-on-import timing. It
+// also stops the underlying SDK from double-instrumenting embeddings, whose `embedQuery`/
+// `embedDocuments` call the provider SDK (e.g. `openai`) internally.
+function markProvidersSkipped(): void {
+  _INTERNAL_skipAiProviderWrapping(SKIPPED_PROVIDERS);
+}
+
+const _langChainChannelIntegration = ((options: LangChainOptions = {}) => {
+  return {
+    name: INTEGRATION_NAME,
+    setupOnce() {
+      // `tracingChannel` is unavailable before Node 18.19, and a second `init()` would double-subscribe.
+      if (!diagnosticsChannel.tracingChannel || subscribed) {
+        return;
+      }
+      subscribed = true;
+
+      // One stateful handler tracks spans across the whole run tree, just like the OTel path.
+      const sentryHandler = createLangChainCallbackHandler(options);
+
+      // Chat models: inject the Sentry callback handler into the call options (arg 1). LangChain's own
+      // callback dispatch then creates the spans, exactly as in the OTel path, so no span is opened
+      // here — a `start` subscriber (which also makes orchestrion wrap the function) is enough.
+      const injectHandler = (message: unknown): void => {
+        markProvidersSkipped();
+
+        const args = (message as RunnableChannelContext).arguments;
+        if (!Array.isArray(args)) {
+          return;
+        }
+
+        let callOptions = args[1] as Record<string, unknown> | undefined;
+        if (!callOptions || typeof callOptions !== 'object' || Array.isArray(callOptions)) {
+          callOptions = {};
+          args[1] = callOptions;
+        }
+
+        callOptions.callbacks = _INTERNAL_mergeLangChainCallbackHandler(callOptions.callbacks, sentryHandler);
+      };
+
+      for (const channelName of [CHANNELS.LANGCHAIN_CHAT_MODEL_INVOKE, CHANNELS.LANGCHAIN_CHAT_MODEL_STREAM]) {
+        DEBUG_BUILD && debug.log(`[orchestrion:langchain] subscribing to channel "${channelName}"`);
+        diagnosticsChannel.tracingChannel<RunnableChannelContext>(channelName).start.subscribe(injectHandler);
+      }
+
+      // Embeddings don't use the callback system — the OTel path wraps the method in its own span, so
+      // do the same here. `bindTracingChannelToSpan` needs the async-context binding that
+      // `initOpenTelemetry()` registers after `setupOnce`, so wait for it before subscribing.
+      waitForTracingChannelBinding(() => {
+        for (const channelName of langchainEmbeddingsChannels) {
+          DEBUG_BUILD && debug.log(`[orchestrion:langchain] subscribing to channel "${channelName}"`);
+          bindTracingChannelToSpan(
+            diagnosticsChannel.tracingChannel<EmbeddingsChannelContext>(channelName),
+            data => createEmbeddingsSpan(data, options),
+            { captureError: () => ({ mechanism: { handled: false, type: 'auto.ai.langchain' } }) },
+          );
+        }
+      });
+    },
+  };
+}) satisfies IntegrationFn;
+
+function createEmbeddingsSpan(data: EmbeddingsChannelContext, options: LangChainOptions): Span {
+  // `embedQuery`/`embedDocuments` call the provider SDK internally, so skip that SDK's own
+  // instrumentation before its channel fires (the producer runs at the embeddings channel's `start`).
+  markProvidersSkipped();
+
+  const input = (data.arguments ?? [])[0];
+
+  return startInactiveSpan(_INTERNAL_getLangChainEmbeddingsSpanOptions(data.self, input, options));
+}
+
+/**
+ * EXPERIMENTAL — orchestrion-driven LangChain integration. Subscribes to the diagnostics_channels
+ * injected into `@langchain/core`'s `BaseChatModel` (to inject the Sentry callback handler) and into
+ * `@langchain/openai`'s embedding methods, so it requires the orchestrion runtime hook or bundler plugin.
+ */
+export const langChainChannelIntegration = defineIntegration(_langChainChannelIntegration);

packages/server-utils/src/orchestrion/config/langchain.ts

@@ -1,6 +1,73 @@
 import type { InstrumentationConfig } from '..';
 
-// TODO: Stub for the `langchain` orchestrion integration (ports `SentryLangChainInstrumentation`).
-export const langchainConfig: InstrumentationConfig[] = [];
+// `@langchain/*` packages ship dual CJS/ESM builds (`.cjs` for `require`, `.js` for `import`) and the
+// matcher compares `filePath` exactly, so each hook is declared once per built file.
 
-export const langchainChannels = {} as const;
+// LangChain's chat model methods live on `BaseChatModel` in `@langchain/core` and are inherited by
+// every provider class (`ChatAnthropic`, `ChatOpenAI`, …), so a single hook there covers all
+// providers. `invoke` also backs `.batch()` (which calls `invoke` per item); `_streamIterator`
+// backs `.stream()`. The vendored OTel instrumentation instead patched each provider package to
+// dodge `@langchain/core` being bundled, but orchestrion transforms its source directly regardless
+// of bundling.
+const chatModelConfig = ['dist/language_models/chat_models.cjs', 'dist/language_models/chat_models.js'].flatMap(
+  filePath => {
+    const module = { name: '@langchain/core', versionRange: '>=0.1.0 <2.0.0', filePath };
+
+    return [
+      {
+        channelName: 'chatModelInvoke',
+        module,
+        functionQuery: { className: 'BaseChatModel', methodName: 'invoke', kind: 'Async' as const },
+      },
+      {
+        channelName: 'chatModelStream',
+        module,
+        functionQuery: { className: 'BaseChatModel', methodName: '_streamIterator', kind: 'Async' as const },
+      },
+    ];
+  },
+);
+
+// Embeddings have no shared concrete method on the base class (each provider implements
+// `embedQuery`/`embedDocuments`), so they're hooked per package. These are the provider packages the
+// vendored OTel instrumentation covered that actually ship an `Embeddings` subclass: `@langchain/openai`,
+// `@langchain/google-genai` and `@langchain/mistralai` define the two methods on their own class, while
+// `@langchain/google-vertexai` inherits them from the shared `@langchain/google-common` base, so that base
+// is the module hooked for it. (anthropic and groq are chat-only — they ship no embeddings class.) The
+// `embedQuery`/`embedDocuments` channel names are per-method; orchestrion prefixes them with the module
+// name, so the full channel strings stay distinct across packages.
+const EMBED_QUERY = 'embedQuery';
+const EMBED_DOCUMENTS = 'embedDocuments';
+
+const EMBEDDINGS_PROVIDERS = [
+  { name: '@langchain/openai', versionRange: '>=0.1.0 <2.0.0', methods: [EMBED_QUERY, EMBED_DOCUMENTS] },
+  { name: '@langchain/google-genai', versionRange: '>=0.1.0 <3.0.0', methods: [EMBED_QUERY, EMBED_DOCUMENTS] },
+  { name: '@langchain/mistralai', versionRange: '>=0.1.0 <2.0.0', methods: [EMBED_QUERY, EMBED_DOCUMENTS] },
+  // `@langchain/google-vertexai` inherits its embed methods from this shared base. The base's
+  // `embedQuery` delegates to `embedDocuments`, so hooking only `embedDocuments` still traces both
+  // entry points as a single span each, instead of emitting a nested duplicate for `embedQuery`.
+  { name: '@langchain/google-common', versionRange: '>=0.1.0 <3.0.0', methods: [EMBED_DOCUMENTS] },
+];
+
+const embeddingsConfig = EMBEDDINGS_PROVIDERS.flatMap(({ name, versionRange, methods }) =>
+  ['dist/embeddings.cjs', 'dist/embeddings.js'].flatMap(filePath =>
+    methods.map(method => ({
+      channelName: method,
+      module: { name, versionRange, filePath },
+      functionQuery: { methodName: method, kind: 'Async' as const },
+    })),
+  ),
+);
+
+export const langchainConfig = [...chatModelConfig, ...embeddingsConfig] satisfies InstrumentationConfig[];
+
+// The embeddings channel strings the subscriber binds to, derived from the provider list above so that
+// adding a provider is a single edit that both instruments it and subscribes the listener to it.
+export const langchainEmbeddingsChannels = EMBEDDINGS_PROVIDERS.flatMap(({ name, methods }) =>
+  methods.map(method => `orchestrion:${name}:${method}`),
+);
+
+export const langchainChannels = {
+  LANGCHAIN_CHAT_MODEL_INVOKE: 'orchestrion:@langchain/core:chatModelInvoke',
+  LANGCHAIN_CHAT_MODEL_STREAM: 'orchestrion:@langchain/core:chatModelStream',
+} as const;

packages/server-utils/src/orchestrion/index.ts

@@ -13,6 +13,7 @@ import { koaChannelIntegration } from '../integrations/tracing-channel/koa';
 import { ioredisChannelIntegration } from '../integrations/tracing-channel/ioredis';
 import { kafkajsChannelIntegration } from '../integrations/tracing-channel/kafkajs';
 import { knexChannelIntegration } from '../integrations/tracing-channel/knex';
+import { langChainChannelIntegration } from '../integrations/tracing-channel/langchain';
 import { langGraphChannelIntegration } from '../integrations/tracing-channel/langgraph';
 import { lruMemoizerChannelIntegration } from '../integrations/tracing-channel/lru-memoizer';
 import { mongooseChannelIntegration } from '../integrations/tracing-channel/mongoose';
@@ -42,6 +43,7 @@ export {
   ioredisChannelIntegration,
   kafkajsChannelIntegration,
   knexChannelIntegration,
+  langChainChannelIntegration,
   langGraphChannelIntegration,
   lruMemoizerChannelIntegration,
   mongooseChannelIntegration,
@@ -98,6 +100,7 @@ export const channelIntegrations = {
   openaiIntegration: openaiChannelIntegration,
   anthropicIntegration: anthropicChannelIntegration,
   googleGenAIIntegration: googleGenAIChannelIntegration,
+  langChainIntegration: langChainChannelIntegration,
   langGraphIntegration: langGraphChannelIntegration,
   vercelAiIntegration: vercelAiChannelIntegration,
   amqplibIntegration: amqplibChannelIntegration,