fix(langgraph): add resumable stream support to remote graph (#1827)

Co-authored-by: Hunter Lovell <40191806+hntrl@users.noreply.github.com>

.changeset/huge-snakes-sit.md

@@ -0,0 +1,5 @@
+---
+"@langchain/langgraph": patch
+---
+
+Adding support for resumableStreams on remote graphs.

.gitignore

@@ -1,6 +1,10 @@
+# Build outputs
 index.cjs
 index.js
 index.d.ts
+*.d.ts
+*.d.cts
+*.d.mts
 node_modules
 dist
 coverage/

libs/checkpoint-mongodb/.gitignore

@@ -1,7 +1,12 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
+
 node_modules
 dist
 .yarn

libs/checkpoint-postgres/.gitignore

@@ -1,11 +1,14 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
 store.cjs
 store.js
-store.d.ts
-store.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
+
 node_modules
 dist
 .yarn

libs/checkpoint-redis/.gitignore

@@ -1,15 +1,16 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
 shallow.cjs
 shallow.js
-shallow.d.ts
-shallow.d.cts
 store.cjs
 store.js
-store.d.ts
-store.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
+
 node_modules
 dist
 .yarn

libs/checkpoint-sqlite/.gitignore

@@ -1,7 +1,12 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
+
 node_modules
 dist
 .yarn

libs/checkpoint-validation/.gitignore

@@ -1,7 +1,12 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
+
 node_modules
 dist
 .yarn

libs/checkpoint/.gitignore

@@ -1,7 +1,12 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
+
 node_modules
 dist
 .yarn

libs/langgraph-core/.gitignore

@@ -1,31 +1,23 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
 web.cjs
 web.js
-web.d.ts
-web.d.cts
 pregel.cjs
 pregel.js
-pregel.d.ts
-pregel.d.cts
 prebuilt.cjs
 prebuilt.js
-prebuilt.d.ts
-prebuilt.d.cts
 remote.cjs
 remote.js
-remote.d.ts
-remote.d.cts
 zod.cjs
 zod.js
-zod.d.ts
-zod.d.cts
 zod/schema.cjs
 zod/schema.js
-zod/schema.d.ts
-zod/schema.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
 node_modules
 dist
 .yarn

libs/langgraph-core/src/pregel/remote.ts

@@ -53,6 +53,7 @@ export type RemoteGraphParams = Omit<
   url?: string;
   apiKey?: string;
   headers?: Record<string, string>;
+  streamResumable?: boolean;
 };
 
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -187,6 +188,8 @@ export class RemoteGraph<
 
   protected interruptAfter?: Array<keyof Nn> | All;
 
+  protected streamResumable?: boolean;
+
   constructor(params: RemoteGraphParams) {
     super(params);
 
@@ -201,6 +204,7 @@ export class RemoteGraph<
     this.config = params.config;
     this.interruptBefore = params.interruptBefore;
     this.interruptAfter = params.interruptAfter;
+    this.streamResumable = params.streamResumable;
   }
 
   // eslint-disable-next-line @typescript-eslint/ban-ts-comment
@@ -435,6 +439,7 @@ export class RemoteGraph<
         streamSubgraphs,
         ifNotExists: "create",
         signal: mergedConfig.signal,
+        streamResumable: this.streamResumable,
       }
     )) {
       let mode;

libs/langgraph-core/src/tests/remote-graph-resumable.test.ts

@@ -0,0 +1,265 @@
+/* eslint-disable @typescript-eslint/no-explicit-any */
+import { describe, test, expect, vi, beforeEach, afterEach } from "vitest";
+import { Client } from "@langchain/langgraph-sdk";
+import { RemoteGraph } from "../pregel/remote.js";
+import { StreamMode } from "../pregel/types.js";
+import { gatherIterator } from "../utils.js";
+
+const textEncoder = new TextEncoder();
+
+/**
+ * Helper to create an SSE-formatted response body
+ */
+const createSSEResponseBody = (
+  chunks: Array<{ id?: string; event: string; data: unknown }>
+): ReadableStream<Uint8Array> => {
+  const sseLines = chunks.flatMap((chunk) => {
+    const lines: string[] = [];
+    if (chunk.id) lines.push(`id: ${chunk.id}\n`);
+    if (chunk.event) lines.push(`event: ${chunk.event}\n`);
+    lines.push(`data: ${JSON.stringify(chunk.data)}\n`);
+    lines.push("\n");
+    return lines;
+  });
+
+  const uint8Arrays = sseLines.map((line) => textEncoder.encode(line));
+
+  let index = 0;
+  return new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (index < uint8Arrays.length) {
+        controller.enqueue(uint8Arrays[index]);
+        index += 1;
+      } else {
+        controller.close();
+      }
+    },
+  });
+};
+
+/**
+ * Helper to create a stream that yields some chunks then errors
+ * This avoids race conditions with Readable.from() and controller.error()
+ */
+const createErroringSSEResponseBody = (
+  chunks: Array<{ id?: string; event: string; data: unknown }>,
+  error: Error
+): ReadableStream<Uint8Array> => {
+  const sseLines = chunks.flatMap((chunk) => {
+    const lines: string[] = [];
+    if (chunk.id) lines.push(`id: ${chunk.id}\n`);
+    if (chunk.event) lines.push(`event: ${chunk.event}\n`);
+    lines.push(`data: ${JSON.stringify(chunk.data)}\n`);
+    lines.push("\n");
+    return lines;
+  });
+
+  const uint8Arrays = sseLines.map((line) => textEncoder.encode(line));
+
+  let index = 0;
+  return new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (index < uint8Arrays.length) {
+        controller.enqueue(uint8Arrays[index]);
+        index += 1;
+      } else {
+        controller.error(error);
+      }
+    },
+  });
+};
+
+describe("RemoteGraph with streamResumable", () => {
+  let mockFetch: ReturnType<typeof vi.fn>;
+  let client: Client;
+
+  beforeEach(() => {
+    vi.useFakeTimers();
+    mockFetch = vi.fn();
+
+    client = new Client({
+      apiUrl: "http://localhost:8000",
+      apiKey: "test-key",
+      callerOptions: {
+        fetch: mockFetch,
+      },
+    });
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+    vi.useRealTimers();
+  });
+
+  test("works with normal stream", async () => {
+    const chunks = [
+      {
+        id: "1",
+        event: "metadata",
+        data: { run_id: "run-1", thread_id: "test_thread" },
+      },
+      { id: "2", event: "values", data: { messages: ["hello"] } },
+      { id: "3", event: "values", data: { messages: ["hello", "world"] } },
+    ];
+
+    mockFetch.mockResolvedValueOnce(
+      new Response(createSSEResponseBody(chunks), {
+        status: 200,
+        headers: {
+          "content-type": "text/event-stream",
+          location: "/threads/test_thread/runs/stream",
+        },
+      })
+    );
+
+    const remoteGraph = new RemoteGraph({
+      graphId: "test_graph",
+      client,
+      streamResumable: true,
+    });
+
+    const config = {
+      configurable: { thread_id: "test_thread" },
+      streamMode: ["values", "metadata"] as StreamMode[],
+    };
+
+    // Consume the stream
+    const results = [];
+    const stream = await remoteGraph.stream({ input: "test" }, config);
+    for await (const chunk of stream) {
+      results.push(chunk);
+    }
+
+    // Verify we got all the value
+    expect(results.length).toBe(3);
+
+    // Verify streamResumable was passed in the request body
+    const [, init] = mockFetch.mock.calls[0];
+    const body = JSON.parse(init.body as string);
+    expect(body.stream_resumable).toBe(true);
+  });
+
+  test("handles network failures during stream and retries with Location header", async () => {
+    let callCount = 0;
+    const locationPath = "/threads/test_thread/runs/run-1/stream";
+
+    mockFetch.mockImplementation(() => {
+      callCount += 1;
+
+      if (callCount === 1) {
+        console.log("first call");
+        // First call: return partial stream that errors mid-stream
+        const partialChunks = [
+          {
+            id: "1",
+            event: "metadata",
+            data: { run_id: "run-1", thread_id: "test_thread" },
+          },
+          { id: "2", event: "values", data: { messages: ["hello"] } },
+        ];
+
+        return Promise.resolve(
+          new Response(
+            createErroringSSEResponseBody(
+              partialChunks,
+              new TypeError("Network connection lost")
+            ),
+            {
+              status: 200,
+              headers: {
+                "content-type": "text/event-stream",
+                location: locationPath,
+              },
+            }
+          )
+        );
+      } else {
+        console.log("retry");
+        // Retry: return remaining events after reconnection
+        const remainingChunks = [
+          { id: "3", event: "values", data: { messages: ["hello", "world"] } },
+        ];
+
+        return Promise.resolve(
+          new Response(createSSEResponseBody(remainingChunks), {
+            status: 200,
+            headers: {
+              "content-type": "text/event-stream",
+              location: locationPath,
+            },
+          })
+        );
+      }
+    });
+
+    const remoteGraph = new RemoteGraph({
+      graphId: "test_graph",
+      client,
+      streamResumable: true,
+    });
+
+    const config = {
+      configurable: { thread_id: "test_thread" },
+      streamMode: ["values", "metadata"] as StreamMode[],
+    };
+
+    // This should handle the network failure and retry
+    const stream = await remoteGraph.stream({ input: "test" }, config);
+    const generator = gatherIterator(stream);
+
+    await vi.runAllTimersAsync();
+
+    const results = await generator;
+    expect(mockFetch).toHaveBeenCalledTimes(2);
+    expect(results.length).toBe(3);
+
+    const [, init] = mockFetch.mock.calls[0];
+    const body = JSON.parse(init.body as string);
+    expect(body.stream_resumable).toBe(true);
+  });
+
+  test("fails normally when streamResumable is false", async () => {
+    const chunks = [
+      {
+        event: "metadata",
+        data: { run_id: "run-1", thread_id: "test_thread" },
+      },
+    ];
+
+    mockFetch.mockResolvedValueOnce(
+      new Response(
+        createErroringSSEResponseBody(chunks, new TypeError("network error")),
+        {
+          status: 200,
+          headers: { "content-type": "text/event-stream" },
+        }
+      )
+    );
+
+    const remoteGraph = new RemoteGraph({
+      graphId: "test_graph",
+      client,
+      streamResumable: false, // Explicitly disabled
+    });
+
+    const config = {
+      configurable: { thread_id: "test_thread" },
+      streamMode: ["values", "metadata"] as StreamMode[],
+    };
+
+    const stream = await remoteGraph.stream({ input: "test" }, config);
+
+    // Should get a chunk first
+    const value = await stream.next();
+    expect(value.done).toBe(false);
+    expect(value.value[0]).toEqual("metadata");
+    expect(value.value[1]).toEqual({
+      run_id: "run-1",
+      thread_id: "test_thread",
+    });
+
+    // Then we should fail
+    await expect(stream.next()).rejects.toThrow("network error");
+    expect(mockFetch).toHaveBeenCalledTimes(1);
+  });
+});

libs/langgraph-cua/.gitignore

@@ -1,11 +1,14 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
 utils.cjs
 utils.js
-utils.d.ts
-utils.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
+
 node_modules
 dist
 .yarn

libs/langgraph-supervisor/.gitignore

@@ -1,7 +1,12 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
+
 node_modules
 dist
 .yarn

libs/langgraph-swarm/.gitignore

@@ -1,7 +1,12 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
+
 node_modules
 dist
 .yarn

libs/langgraph/.gitignore

@@ -1,31 +1,23 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
 web.cjs
 web.js
-web.d.ts
-web.d.cts
 pregel.cjs
 pregel.js
-pregel.d.ts
-pregel.d.cts
 prebuilt.cjs
 prebuilt.js
-prebuilt.d.ts
-prebuilt.d.cts
 remote.cjs
 remote.js
-remote.d.ts
-remote.d.cts
 zod.cjs
 zod.js
-zod.d.ts
-zod.d.cts
 zod/schema.cjs
 zod/schema.js
-zod/schema.d.ts
-zod/schema.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
 node_modules
 dist
 .yarn

libs/sdk/.gitignore

@@ -1,31 +1,23 @@
+# Build outputs - JS/CJS files
 index.cjs
 index.js
-index.d.ts
-index.d.cts
 client.cjs
 client.js
-client.d.ts
-client.d.cts
 auth.cjs
 auth.js
-auth.d.ts
-auth.d.cts
 react.cjs
 react.js
-react.d.ts
-react.d.cts
 logging.cjs
 logging.js
-logging.d.ts
-logging.d.cts
 react-ui.cjs
 react-ui.js
-react-ui.d.ts
-react-ui.d.cts
 react-ui/server.cjs
 react-ui/server.js
-react-ui/server.d.ts
-react-ui/server.d.cts
+
+# Generated TypeScript declarations
+*.d.ts
+*.d.cts
+*.d.mts
 node_modules
 dist
 .yarn