fix(sdk): useStream should receive and merge messages from subgraphs (#1387)

.changeset/bright-boats-teach.md

@@ -0,0 +1,5 @@
+---
+"@langchain/langgraph-sdk": patch
+---
+
+useStream should receive and merge messages from subgraphs

.changeset/silly-webs-lie.md

@@ -0,0 +1,7 @@
+---
+"@langchain/langgraph-api": patch
+"@langchain/langgraph-cli": patch
+"@langchain/langgraph-ui": patch
+---
+
+Fix apply namespace to messages-tuple stream mode

libs/sdk-validation/src/tests/stream.test.tsx

@@ -47,7 +47,16 @@ const agent = new StateGraph(MessagesAnnotation)
   .addEdge(START, "agent")
   .compile();
 
-const app = createEmbedServer({ graph: { agent }, checkpointer, threads });
+const parentAgent = new StateGraph(MessagesAnnotation)
+  .addNode("agent", agent, { subgraphs: [agent] })
+  .addEdge(START, "agent")
+  .compile();
+
+const app = createEmbedServer({
+  graph: { agent, parentAgent },
+  checkpointer,
+  threads,
+});
 const server = setupServer(http.all("*", (ctx) => app.fetch(ctx.request)));
 
 function TestChatComponent() {
@@ -753,4 +762,65 @@ describe("useStream", () => {
       expect(screen.getByTestId("message-1")).toHaveTextContent("Hey");
     });
   });
+
+  it("streamSubgraphs: true and messages-tuple", async () => {
+    const user = userEvent.setup();
+
+    function TestComponent() {
+      const { submit, messages } = useStream({
+        assistantId: "parentAgent",
+        apiKey: "test-api-key",
+      });
+
+      return (
+        <div>
+          <div data-testid="messages">
+            {messages.map((msg, i) => (
+              <div key={msg.id ?? i} data-testid={`message-${i}`}>
+                {typeof msg.content === "string"
+                  ? msg.content
+                  : JSON.stringify(msg.content)}
+              </div>
+            ))}
+          </div>
+          <button
+            data-testid="submit"
+            onClick={() =>
+              submit(
+                { messages: [{ content: "Hello", type: "human" }] },
+                { streamSubgraphs: true }
+              )
+            }
+          >
+            Send
+          </button>
+        </div>
+      );
+    }
+
+    render(<TestComponent />);
+
+    await user.click(screen.getByTestId("submit"));
+
+    // Make sure that we're properly streaming the tokens from subgraphs
+    await waitFor(() => {
+      expect(screen.getByTestId("message-0")).toHaveTextContent("Hello");
+      expect(screen.getByTestId("message-1").textContent).toBe("H");
+    });
+
+    await waitFor(() => {
+      expect(screen.getByTestId("message-0")).toHaveTextContent("Hello");
+      expect(screen.getByTestId("message-1").textContent).toBe("He");
+    });
+
+    await waitFor(() => {
+      expect(screen.getByTestId("message-0")).toHaveTextContent("Hello");
+      expect(screen.getByTestId("message-1").textContent).toBe("Hey");
+    });
+
+    await waitFor(() => {
+      expect(screen.getByTestId("message-0")).toHaveTextContent("Hello");
+      expect(screen.getByTestId("message-1").textContent).toBe("Hey");
+    });
+  });
 });

libs/sdk/src/react/stream.tsx

@@ -1041,8 +1041,13 @@ export function useStream<
           }
           setStreamValues(data);
         }
-        if (event === "messages") {
-          const [serialized] = data;
+        if (
+          event === "messages" ||
+          // if `streamSubgraphs: true`, then we also want
+          // to also receive messages from subgraphs
+          event.startsWith("messages|")
+        ) {
+          const [serialized] = data as MessagesTupleStreamEvent["data"];
 
           const messageId = messageManagerRef.current.add(serialized);
           if (!messageId) {