feat(sdk-js): add docs for reconnectOnMount, expose run metadata

docs/docs/cloud/how-tos/use_stream_react.md

@@ -1,8 +1,8 @@
-# How to integrate LangGraph into your React application
+How to integrate LangGraph into your React application# How to integrate LangGraph into your React application
 
-!!! info "Prerequisites" 
+!!! info "Prerequisites"
 
-    - [LangGraph Platform](../../concepts/langgraph_platform.md) 
+    - [LangGraph Platform](../../concepts/langgraph_platform.md)
     - [LangGraph Server](../../concepts/langgraph_server.md)
 
 The `useStream()` React hook provides a seamless way to integrate LangGraph into your React applications. It handles all the complexities of streaming, state management, and branching logic, letting you focus on building great chat experiences.
@@ -113,6 +113,115 @@ export default function App() {
 }
 ```
 
+### Resume a stream after page refresh
+
+The `useStream()` hook can automatically resume an already ongoing run on mount by passing `reconnectOnMount: true`. This is useful for resuming an ongoing stream after a full page refresh without losing any messages that were generated in the meantime.
+
+```tsx
+const thread = useStream<{ messages: Message[] }>({
+  apiUrl: "http://localhost:2024",
+  assistantId: "agent",
+  reconnectOnMount: true,
+});
+```
+
+By default the ID of the created run is stored in `window.sessionStorage` under `lg:stream:${threadId}` key. This can be swapped by passing a custom storage in `reconnectOnMount`.
+
+```tsx
+const thread = useStream<{ messages: Message[] }>({
+  apiUrl: "http://localhost:2024",
+  assistantId: "agent",
+  reconnectOnMount: () => window.localStorage,
+});
+```
+
+Finally, you can manually handle the lifecycle of stream resuming by using the run callbacks and `joinStream` method. Make sure to pass `streamResumable: true` and `onDisconnect: "continue"` when creating the run.
+
+````tsx
+import type { Message } from "@langchain/langgraph-sdk";
+import { useStream } from "@langchain/langgraph-sdk/react";
+import { useCallback, useState, useEffect, useRef } from "react";
+
+export default function App() {
+  const [threadId, onThreadId] = useSearchParam("threadId");
+
+  const thread = useStream<{ messages: Message[] }>({
+    apiUrl: "http://localhost:2024",
+    assistantId: "agent",
+
+    threadId,
+    onThreadId,
+
+    onCreated: (run) => {
+      window.sessionStorage.setItem(`resume:${run.thread_id}`, run.run_id);
+    },
+    onFinish: (_, run) => {
+      window.sessionStorage.removeItem(`resume:${run?.thread_id}`);
+    },
+  });
+
+  // Ensure that we only join the stream once per thread.
+  const joinedThreadId = useRef<string | null>(null);
+  useEffect(() => {
+    if (!threadId) return;
+
+    const resume = window.sessionStorage.getItem(`resume:${threadId}`);
+    if (resume && joinedThreadId.current !== threadId) {
+      thread.joinStream(resume);
+      joinedThreadId.current = threadId;
+    }
+  }, [threadId]);
+
+  return (
+    <form
+      onSubmit={(e) => {
+        e.preventDefault();
+        const form = e.target as HTMLFormElement;
+        const message = new FormData(form).get("message") as string;
+        thread.submit(
+          { messages: [{ type: "human", content: message }] },
+          { streamResumable: true }
+        );
+      }}
+    >
+      <div>
+        {thread.messages.map((message) => (
+          <div key={message.id}>{JSON.stringify(message.content)}</div>
+        ))}
+      </div>
+      <input type="text" name="message" />
+      <button type="submit">Send</button>
+    </form>
+  );
+}
+
+// Utility method to retrieve and persist data in URL as search param
+function useSearchParam(key: string) {
+  const [value, setValue] = useState<string | null>(() => {
+    const params = new URLSearchParams(window.location.search);
+    return params.get(key) ?? null;
+  });
+
+  const update = useCallback(
+    (value: string | null) => {
+      setValue(value);
+
+      const url = new URL(window.location.href);
+      if (value == null) {
+        url.searchParams.delete(key);
+      } else {
+        url.searchParams.set(key, value);
+      }
+
+      window.history.pushState({}, "", url.toString());
+    },
+    [key]
+  );
+
+  return [value, update] as const;
+}
+```
+
 ### Thread Management
 
 Keep track of conversations with built-in thread management. You can access the current thread ID and get notified when new threads are created:
@@ -127,7 +236,7 @@ const thread = useStream<{ messages: Message[] }>({
   threadId: threadId,
   onThreadId: setThreadId,
 });
-```
+````
 
 We recommend storing the `threadId` in your URL's query parameters to let users resume conversations after page refreshes.
 

libs/sdk-js/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@langchain/langgraph-sdk",
-  "version": "0.0.78",
+  "version": "0.0.79",
   "description": "Client library for interacting with the LangGraph API",
   "type": "module",
   "packageManager": "yarn@1.22.19",

libs/sdk-js/src/react/stream.tsx

@@ -405,6 +405,11 @@ type GetCustomEventType<Bag extends BagTemplate> = Bag extends {
   ? Bag["CustomEventType"]
   : unknown;
 
+interface RunCallbackMeta {
+  run_id: string;
+  thread_id: string;
+}
+
 export interface UseStreamOptions<
   StateType extends Record<string, unknown> = Record<string, unknown>,
   Bag extends BagTemplate = BagTemplate,
@@ -450,17 +455,20 @@ export interface UseStreamOptions<
   /**
    * Callback that is called when an error occurs.
    */
-  onError?: (error: unknown) => void;
+  onError?: (error: unknown, run: RunCallbackMeta | undefined) => void;
 
   /**
    * Callback that is called when the stream is finished.
    */
-  onFinish?: (state: ThreadState<StateType>) => void;
+  onFinish?: (
+    state: ThreadState<StateType>,
+    run: RunCallbackMeta | undefined,
+  ) => void;
 
   /**
    * Callback that is called when a new stream is created.
    */
-  onCreated?: (run: { run_id: string; thread_id: string }) => void;
+  onCreated?: (run: RunCallbackMeta) => void;
 
   /**
    * Callback that is called when an update event is received.
@@ -846,8 +854,12 @@ export function useStream<
     action: (signal: AbortSignal) => Promise<{
       onSuccess: () => Promise<ThreadState<StateType>[]>;
       stream: AsyncGenerator<EventStreamEvent>;
+      getCallbackMeta: () => { thread_id: string; run_id: string } | undefined;
     }>,
   ) {
+    let getCallbackMeta:
+      | (() => { thread_id: string; run_id: string } | undefined)
+      | undefined;
     try {
       setIsLoading(true);
       setStreamError(undefined);
@@ -856,6 +868,7 @@ export function useStream<
       abortRef.current = new AbortController();
 
       const run = await action(abortRef.current.signal);
+      getCallbackMeta = run.getCallbackMeta;
 
       let streamError: StreamError | undefined;
       for await (const { event, data } of run.stream) {
@@ -916,7 +929,7 @@ export function useStream<
       if (streamError != null) throw streamError;
 
       const lastHead = result.at(0);
-      if (lastHead) onFinish?.(lastHead);
+      if (lastHead) onFinish?.(lastHead, getCallbackMeta?.());
     } catch (error) {
       if (
         !(
@@ -926,7 +939,7 @@ export function useStream<
       ) {
         console.error(error);
         setStreamError(error);
-        onError?.(error);
+        onError?.(error, getCallbackMeta?.());
       }
     } finally {
       setIsLoading(false);
@@ -953,6 +966,7 @@ export function useStream<
           return history.mutate(threadId);
         },
         stream,
+        getCallbackMeta: () => ({ thread_id: threadId, run_id: runId }),
       };
     });
   };
@@ -1004,6 +1018,9 @@ export function useStream<
       // @ts-expect-error
       if (checkpoint != null) delete checkpoint.thread_id;
       let rejoinKey: `lg:stream:${string}` | undefined;
+      let callbackMeta: RunCallbackMeta | undefined;
+      const streamResumable =
+        submitOptions?.streamResumable ?? !!runMetadataStorage;
 
       const stream = client.runs.stream(usableThreadId, assistantId, {
         input: values as Record<string, unknown>,
@@ -1017,29 +1034,31 @@ export function useStream<
         onCompletion: submitOptions?.onCompletion,
         onDisconnect:
           submitOptions?.onDisconnect ??
-          (runMetadataStorage ? "continue" : "cancel"),
+          (streamResumable ? "continue" : "cancel"),
 
         signal,
 
         checkpoint,
         streamMode,
         streamSubgraphs: submitOptions?.streamSubgraphs,
-        streamResumable: submitOptions?.streamResumable ?? !!runMetadataStorage,
+        streamResumable,
         onRunCreated(params) {
-          const runParams = {
+          callbackMeta = {
             run_id: params.run_id,
             thread_id: params.thread_id ?? usableThreadId,
           };
+
           if (runMetadataStorage) {
-            rejoinKey = `lg:stream:${runParams.thread_id}`;
-            runMetadataStorage.setItem(rejoinKey, runParams.run_id);
+            rejoinKey = `lg:stream:${callbackMeta.thread_id}`;
+            runMetadataStorage.setItem(rejoinKey, callbackMeta.run_id);
           }
-          onCreated?.(runParams);
+          onCreated?.(callbackMeta);
         },
       }) as AsyncGenerator<EventStreamEvent>;
 
       return {
         stream,
+        getCallbackMeta: () => callbackMeta,
         onSuccess: () => {
           if (rejoinKey) runMetadataStorage?.removeItem(rejoinKey);
           return history.mutate(usableThreadId);