feat: Add @langchain/langgraph-sdk instrumentation

.changeset/langgraph-sdk-instrumentation.md

@@ -0,0 +1,5 @@
+---
+"braintrust": minor
+---
+
+feat: Add `@langchain/langgraph-sdk` instrumentation

e2e/config/pr-comment-scenarios.json

@@ -778,6 +778,21 @@
       }
     ]
   },
+  {
+    "scenarioDirName": "langgraph-sdk-instrumentation",
+    "label": "LangGraph Platform SDK Instrumentation",
+    "metadataScenario": "langgraph-sdk-instrumentation",
+    "variants": [
+      {
+        "variantKey": "langgraph-sdk-v1",
+        "label": "v1 pinned"
+      },
+      {
+        "variantKey": "langgraph-sdk-v1-latest",
+        "label": "v1 latest"
+      }
+    ]
+  },
   {
     "scenarioDirName": "elevenlabs-instrumentation",
     "label": "ElevenLabs Instrumentation",

e2e/helpers/mock-braintrust-server.test.ts

@@ -0,0 +1,134 @@
+import { createServer } from "node:http";
+import type { AddressInfo } from "node:net";
+import { setTimeout } from "node:timers/promises";
+import { describe, expect, it } from "vitest";
+import { startMockBraintrustServer } from "./mock-braintrust-server";
+
+describe("production forwarding", () => {
+  it.each(["/logs3", "/otel/v1/traces"])(
+    "preserves write order for %s even when the initial write is slow",
+    async (path) => {
+      let releaseInitial!: () => void;
+      const initialGate = new Promise<void>((resolve) => {
+        releaseInitial = resolve;
+      });
+      let initialReceived!: () => void;
+      const initialStarted = new Promise<void>((resolve) => {
+        initialReceived = resolve;
+      });
+      const applied: number[] = [];
+      let stored: Record<string, unknown> = {};
+      const upstream = createServer(async (req, res) => {
+        let body = "";
+        for await (const chunk of req) body += chunk;
+        const {
+          sequence,
+          rows: [row],
+        } = JSON.parse(body);
+        if (sequence === 1) {
+          initialReceived();
+          await initialGate;
+        }
+        stored = row._is_merge ? { ...stored, ...row } : row;
+        applied.push(sequence);
+        res.end("{}");
+      });
+      await new Promise<void>((resolve) =>
+        upstream.listen(0, "127.0.0.1", resolve),
+      );
+      const url = `http://127.0.0.1:${(upstream.address() as AddressInfo).port}`;
+      const server = await startMockBraintrustServer({
+        prodForwarding: {
+          apiKey: "test-only-key",
+          apiUrl: url,
+          appUrl: url,
+          orgId: "org",
+          orgName: "org",
+          projectId: "project",
+          projectName: "tmp-luca-forwarding-test",
+        },
+      });
+      try {
+        for (const [index, row] of [
+          { id: "span", input: "hello", metrics: { start: 1 } },
+          {
+            id: "span",
+            _is_merge: true,
+            error: "expected failure",
+            metrics: { start: 1, end: 2 },
+          },
+        ].entries()) {
+          const response = await fetch(`${server.url}${path}`, {
+            method: "POST",
+            headers: { "Content-Type": "application/json" },
+            body: JSON.stringify({
+              api_version: 2,
+              sequence: index + 1,
+              rows: [row],
+            }),
+          });
+          expect(response.ok).toBe(true);
+          await response.text();
+          if (index === 0) await initialStarted;
+        }
+        // Give an incorrectly concurrent second request time to overtake the
+        // gated initial upsert. The mock should still acknowledge both promptly.
+        await setTimeout(50);
+        releaseInitial();
+        await server.close();
+        expect(applied).toEqual([1, 2]);
+        expect(stored).toMatchObject({
+          input: "hello",
+          error: "expected failure",
+          metrics: { end: 2 },
+        });
+      } finally {
+        releaseInitial();
+        upstream.closeAllConnections();
+        await new Promise<void>((resolve) => upstream.close(() => resolve()));
+      }
+    },
+  );
+
+  it("reports a failed write without preventing later queued writes", async () => {
+    const received: number[] = [];
+    const upstream = createServer(async (req, res) => {
+      let body = "";
+      for await (const chunk of req) body += chunk;
+      const { sequence } = JSON.parse(body);
+      received.push(sequence);
+      res.statusCode = sequence === 1 ? 500 : 200;
+      res.end(sequence === 1 ? "initial write failed" : "{}");
+    });
+    await new Promise<void>((resolve) =>
+      upstream.listen(0, "127.0.0.1", resolve),
+    );
+    const url = `http://127.0.0.1:${(upstream.address() as AddressInfo).port}`;
+    const server = await startMockBraintrustServer({
+      prodForwarding: {
+        apiKey: "test-only-key",
+        apiUrl: url,
+        appUrl: url,
+        orgId: "org",
+        orgName: "org",
+        projectId: "project",
+        projectName: "tmp-luca-forwarding-test",
+      },
+    });
+    try {
+      for (const sequence of [1, 2]) {
+        const response = await fetch(`${server.url}/logs3`, {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({ sequence, api_version: 2, rows: [] }),
+        });
+        await response.text();
+      }
+      await expect(server.close()).rejects.toThrow("initial write failed");
+      expect(received).toEqual([1, 2]);
+    } finally {
+      upstream.closeAllConnections();
+      await new Promise<void>((resolve) => upstream.close(() => resolve()));
+    }
+  });
+});

e2e/helpers/mock-braintrust-server.ts

@@ -290,7 +290,7 @@ export async function startMockBraintrustServer(
   >();
   let serverUrl = "";
   let xactCursor = 0;
-  const pendingProdForwarding = new Set<Promise<void>>();
+  let prodForwardingTail = Promise.resolve();
 
   if (prodForwarding) {
     projectsByName.set(prodForwarding.projectName, {
@@ -402,17 +402,15 @@ export async function startMockBraintrustServer(
     );
   }
 
-  function trackProdForwarding(context: string, promise: Promise<void>): void {
-    pendingProdForwarding.add(promise);
-    void promise.then(
-      () => {
-        pendingProdForwarding.delete(promise);
-      },
-      (error) => {
-        recordProdForwardingError(context, error);
-        pendingProdForwarding.delete(promise);
-      },
-    );
+  function trackProdForwarding(
+    context: string,
+    send: () => Promise<void>,
+  ): void {
+    // A later upsert can overwrite an earlier merge. Acknowledge the local
+    // request promptly, but preserve ingestion order when forwarding upstream.
+    prodForwardingTail = prodForwardingTail.then(send).catch((error) => {
+      recordProdForwardingError(context, error);
+    });
   }
 
   function requestForProdForwarding(
@@ -734,8 +732,7 @@ export async function startMockBraintrustServer(
             persistPayload(payload);
           }
           if (prodForwarding) {
-            trackProdForwarding(
-              "POST /logs3",
+            trackProdForwarding("POST /logs3", () =>
               forwardProdRequest(capturedRequest, {
                 drainResponseBody: true,
               }).then(() => undefined),
@@ -750,8 +747,7 @@ export async function startMockBraintrustServer(
           capturedRequest.path === "/otel/v1/traces"
         ) {
           if (prodForwarding) {
-            trackProdForwarding(
-              "POST /otel/v1/traces",
+            trackProdForwarding("POST /otel/v1/traces", () =>
               forwardProdRequest(capturedRequest, {
                 drainResponseBody: true,
               }).then(() => undefined),
@@ -785,9 +781,7 @@ export async function startMockBraintrustServer(
       await new Promise<void>((resolve, reject) => {
         server.close((error) => (error ? reject(error) : resolve()));
       });
-      while (pendingProdForwarding.size > 0) {
-        await Promise.allSettled([...pendingProdForwarding]);
-      }
+      await prodForwardingTail;
       if (prodForwardingErrors.length > 0) {
         throw new Error(
           [

e2e/helpers/pr-e2e-links.test.ts

@@ -0,0 +1,59 @@
+import { execFile } from "node:child_process";
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { promisify } from "node:util";
+import { fileURLToPath } from "node:url";
+import { expect, it } from "vitest";
+
+it("links published runs while preserving legacy records and excluding local-only runs", async () => {
+  const dir = await mkdtemp(path.join(tmpdir(), "braintrust-e2e-links-"));
+  const configPath = path.join(dir, "config.json");
+  try {
+    await writeFile(
+      configPath,
+      JSON.stringify([
+        {
+          scenarioDirName: "scenario",
+          label: "Scenario",
+          metadataScenario: "scenario",
+        },
+      ]),
+    );
+    await writeFile(
+      path.join(dir, "runs.ndjson"),
+      [
+        { testRunId: "e2e-published", forwardToProduction: true },
+        { testRunId: "e2e-local-only", forwardToProduction: false },
+        { testRunId: "e2e-legacy" },
+      ]
+        .map((record) =>
+          JSON.stringify({ scenarioDirName: "scenario", ...record }),
+        )
+        .join("\n"),
+    );
+    const { stdout } = await promisify(execFile)(
+      process.execPath,
+      [
+        fileURLToPath(
+          new URL("../scripts/build-pr-e2e-links-comment.mjs", import.meta.url),
+        ),
+        "--config",
+        configPath,
+      ],
+      {
+        env: {
+          ...process.env,
+          BRAINTRUST_API_KEY: "",
+          BRAINTRUST_ORG_NAME: "Test",
+          BRAINTRUST_E2E_RUN_CONTEXT_DIR: dir,
+        },
+      },
+    );
+    expect(stdout).toContain("e2e-published");
+    expect(stdout).toContain("e2e-legacy");
+    expect(stdout).not.toContain("e2e-local-only");
+  } finally {
+    await rm(dir, { recursive: true, force: true });
+  }
+});

e2e/helpers/scenario-harness.ts

@@ -80,6 +80,7 @@ export interface ScenarioRunContext {
 }
 
 interface ScenarioRunContextRecord {
+  forwardToProduction?: boolean;
   entry: string;
   runner: ScenarioRunner;
   scenarioDirName: string;
@@ -700,9 +701,13 @@ interface ScenarioHarness {
 
 export async function withScenarioHarness(
   body: (harness: ScenarioHarness) => Promise<void>,
+  optionsForHarness: { forwardToProduction?: boolean } = {},
 ): Promise<void> {
   const { getProdForwarding } = await import("./prod-forwarding");
-  const prodForwarding = getProdForwarding();
+  const prodForwarding =
+    optionsForHarness.forwardToProduction === false
+      ? null
+      : getProdForwarding();
   const testRunId = createTestRunId();
   const server = await startMockBraintrustServer({
     prodForwarding,
@@ -832,6 +837,7 @@ export async function withScenarioHarness(
   ): Promise<ScenarioResult> => {
     const result = await run();
     await recordScenarioRunContext({
+      forwardToProduction: optionsForHarness.forwardToProduction,
       entry: options.entry ?? defaultEntry,
       runner,
       scenarioDirName: path.basename(