fix(api): apply namespace to messages-tuple stream mode (#1386)

.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/langgraph-api/src/stream.mts

@@ -271,7 +271,11 @@ export async function* streamState(
 
       if (mode === "messages") {
         if (userStreamMode.includes("messages-tuple")) {
-          yield { event: "messages", data };
+          if (kwargs.subgraphs && ns?.length) {
+            yield { event: `messages|${ns.join("|")}`, data };
+          } else {
+            yield { event: "messages", data };
+          }
         }
       } else if (userStreamMode.includes(mode)) {
         if (kwargs.subgraphs && ns?.length) {

libs/langgraph-api/tests/api.test.mts

@@ -3,7 +3,10 @@ import type {
   BaseMessageLike,
   MessageType,
 } from "@langchain/core/messages";
-import { Client, type FeedbackStreamEvent } from "@langchain/langgraph-sdk";
+import {
+  Client,
+  type MessagesTupleStreamEvent,
+} from "@langchain/langgraph-sdk";
 import { RemoteGraph } from "@langchain/langgraph/remote";
 import postgres from "postgres";
 import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
@@ -936,18 +939,24 @@ describe("runs", () => {
     const stream = await client.runs.stream(
       thread.thread_id,
       assistant.assistant_id,
-      { input, streamMode: "messages-tuple", config: globalConfig }
+      {
+        input,
+        streamMode: "messages-tuple",
+        config: globalConfig,
+        streamSubgraphs: true,
+      }
     );
 
     const chunks = await gatherIterator(stream);
-    const runId = findLast(
-      chunks,
-      (i): i is FeedbackStreamEvent => i.event === "metadata"
-    )?.data.run_id;
+    const runId = findLast(chunks, (i) => i.event === "metadata")?.data.run_id;
+    expect(runId).not.toBeUndefined();
     expect(runId).not.toBeNull();
 
     const messages = chunks
-      .filter((i) => i.event === "messages")
+      .filter(
+        (i): i is MessagesTupleStreamEvent =>
+          i.event.startsWith("messages|") || i.event === "messages"
+      )
       .map((i) => i.data[0]);
 
     expect(messages).toHaveLength("begin".length + "end".length + 1);
@@ -957,7 +966,7 @@ describe("runs", () => {
       ..."end".split("").map((c) => ({ content: c })),
     ]);
 
-    const seenEventTypes = new Set(chunks.map((i) => i.event));
+    const seenEventTypes = new Set(chunks.map((i) => i.event.split("|")[0]));
     expect(seenEventTypes).toEqual(new Set(["metadata", "messages"]));
 
     const run = await client.runs.get(thread.thread_id, runId as string);