Match MessagesTransformer scope filter to JS root-feed semantics

Commit 133082af preserved the full checkpoint ns on v2 message
events so subgraph chat-model tokens emit at their full path (matching
JS's handleChatModelStart). That promoted root-level chat tokens from
the empty tuple to depth 1 (e.g. ("call_model:<task>",)), but the root
MessagesTransformer was scope_exact=True at scope=(), so every
streamed message was dropped — run.messages yielded zero streams for
any graph calling model.invoke() / stream_v2().

Mirror JS's namespaces=[[]], depth=1 filter: accept events at the
transformer's scope or exactly one segment deeper. SubgraphTransformer
still forwards deeper events to the matching child mini-mux, whose
own MessagesTransformer applies the same scope+1 rule.

Also update stale test assertions: ChatModelStream.output.content is
now always a list of v1 content blocks by construction, so tests
compare via str(stream.text) (sync) or an explicit block-shape
assertion on the assembled AIMessage (async, since AsyncProjection
isn't str-able).

libs/langgraph/langgraph/stream/transformers.py

@@ -119,9 +119,19 @@ class MessagesTransformer(StreamTransformer):
 
     Native transformer — the `messages` projection is exposed as a
     direct attribute on the run stream.
+
+    `scope_exact = False`: matches events at the transformer's own
+    namespace **or** exactly one segment deeper (the chat-model /
+    node's own task ns). Mirrors JS's root-feed filter
+    (`namespaces=[[]], depth=1`) — root accepts depth-0 events plus
+    its own nodes' depth-1 tokens; subgraph mini-muxes accept their
+    own scope plus their internal nodes' tokens. Events deeper than
+    scope + 1 are dropped (the enclosing `SubgraphTransformer` has
+    already forwarded them to the matching child mini-mux).
     """
 
     _native = True
+    scope_exact = False
     required_stream_modes = ("messages",)
 
     def __init__(self, scope: tuple[str, ...] = ()) -> None:
@@ -186,10 +196,16 @@ class MessagesTransformer(StreamTransformer):
         )
 
     def process(self, event: ProtocolEvent) -> bool:
-        # Namespace filtering is handled by the mux via `scope_exact`.
         if event["method"] != "messages":
             return True
         params = event["params"]
+        # Accept events at our scope or exactly one segment deeper
+        # (the chat-model / node's own task ns). Deeper events belong
+        # to a subgraph and are routed by `SubgraphTransformer`.
+        ns = tuple(params["namespace"])
+        depth = len(self.scope)
+        if len(ns) > depth + 1 or ns[:depth] != self.scope:
+            return True
 
         payload, metadata = params["data"]
         node: str | None = metadata.get("langgraph_node")

libs/langgraph/tests/test_pregel_stream_v2.py

@@ -973,20 +973,24 @@ class TestMessagesTransformer:
         assert hasattr(items[0], "dispatch")
         assert items[0].message_id == "run-1"
 
-    def test_ignores_non_root_namespace(self) -> None:
-        """Namespace filtering is enforced by the mux via `scope_exact`."""
+    def test_ignores_deeper_than_scope_plus_one(self) -> None:
+        """Root MessagesTransformer accepts scope + 1 (matching JS's
+        `depth=1` root feed), so a depth-1 chat-model ns is kept, but
+        events nested two or more segments deep belong to a subgraph
+        and are dropped.
+        """
         mux = StreamMux([MessagesTransformer()], is_async=False)
         t = mux.transformer_by_key("messages")
         assert isinstance(t, MessagesTransformer)
         t._bind_pump(lambda: False)
         it = iter(t._log)
 
-        meta = {"langgraph_node": "llm", "run_id": "run-1"}
+        meta = {"langgraph_node": "llm", "run_id": "run-deep"}
         mux.push(
             _event(
                 "messages",
-                ({"event": "message-start", "message_id": "run-1"}, meta),
-                namespace=["sub"],
+                ({"event": "message-start", "message_id": "run-deep"}, meta),
+                namespace=["outer:task", "inner:task"],
             )
         )
         t._log.close()

libs/langgraph/tests/test_stream_messages_transformer.py

@@ -183,7 +183,7 @@ class TestProtocolEventRouting:
         log.close()
         (stream,) = list(log._items)
         assert stream.done
-        assert stream.output.content == "hello world"
+        assert str(stream.text) == "hello world"
 
     def test_message_finish_cleans_up_routing(self) -> None:
         t, log = _make_sync_transformer()
@@ -220,8 +220,8 @@ class TestProtocolEventRouting:
         streams = list(log._items)
         assert len(streams) == 2
         by_id = {s.message_id: s for s in streams}
-        assert by_id["run-a"].output.content == "aaaa"
-        assert by_id["run-b"].output.content == "bbbb"
+        assert str(by_id["run-a"].text) == "aaaa"
+        assert str(by_id["run-b"].text) == "bbbb"
 
     def test_text_deltas_accumulated_on_stream(self) -> None:
         t, log = _make_sync_transformer()
@@ -270,7 +270,7 @@ class TestWholeMessageFallback:
         log.close()
         (stream,) = list(log._items)
         assert stream.done
-        assert stream.output.content == "the full answer"
+        assert str(stream.text) == "the full answer"
 
     def test_whole_message_has_full_lifecycle(self) -> None:
         t, log = _make_sync_transformer()
@@ -317,7 +317,12 @@ class TestFiltering:
         assert t.process(values_event) is True
 
     def test_subgraph_namespace_dropped(self) -> None:
-        """Root MessagesTransformer (via the mux) ignores non-root events."""
+        """Root MessagesTransformer accepts its scope + one segment (JS
+        `depth=1`), so a root-node chat-model ns (depth 1) is picked
+        up, but deeper subgraph chatter (depth 2+) is dropped —
+        `SubgraphTransformer` has already forwarded those into the
+        matching child mini-mux.
+        """
         from langgraph.stream._mux import StreamMux
 
         mux = StreamMux([MessagesTransformer()], is_async=False)
@@ -326,22 +331,34 @@ class TestFiltering:
         t._log._subscribed = True
         t._bind_pump(lambda: False)
 
-        mux.push(
-            {
-                "type": "event",
-                "method": "messages",
-                "params": {
-                    "namespace": ["subgraph"],
-                    "timestamp": TS,
-                    "data": (
-                        {"event": "message-start", "message_id": "run-x"},
-                        {"run_id": "run-x"},
-                    ),
-                },
-            }
-        )
+        def _push(ns: list[str], run_id: str) -> None:
+            mux.push(
+                {
+                    "type": "event",
+                    "method": "messages",
+                    "params": {
+                        "namespace": ns,
+                        "timestamp": TS,
+                        "data": (
+                            {
+                                "event": "message-start",
+                                "message_id": run_id,
+                            },
+                            {"run_id": run_id},
+                        ),
+                    },
+                }
+            )
+
+        # Depth 1 (a root-node chat model): accepted.
+        _push(["call_model:task-1"], run_id="run-root")
+        # Depth 2 (subgraph internal): dropped by the root transformer.
+        _push(["outer:task-1", "inner:task-2"], run_id="run-sub")
+
         t._log.close()
-        assert list(t._log._items) == []
+        streams = list(t._log._items)
+        assert len(streams) == 1
+        assert streams[0].message_id == "run-root"
 
 
 # ---------------------------------------------------------------------------
@@ -410,7 +427,7 @@ class TestAsyncMode:
             t.process(_proto_event(evt))
         (stream,) = list(log._items)
         msg = await stream.output
-        assert msg.content == "async"
+        assert msg.content == [{"type": "text", "text": "async", "index": 0}]
 
 
 # ---------------------------------------------------------------------------
@@ -471,7 +488,7 @@ class TestViaMux:
         mux.close()
 
         (stream,) = list(log._items)
-        assert stream.output.content == "mux stream"
+        assert str(stream.text) == "mux stream"
 
     def test_whole_message_via_mux(self) -> None:
         t = MessagesTransformer()
@@ -485,7 +502,7 @@ class TestViaMux:
         mux.close()
 
         (stream,) = list(log._items)
-        assert stream.output.content == "result"
+        assert str(stream.text) == "result"
 
     @pytest.mark.anyio
     async def test_async_streaming_via_mux(self) -> None:
@@ -501,7 +518,7 @@ class TestViaMux:
         streams = list(log._items)
         assert len(streams) == 1
         msg = await streams[0].output
-        assert msg.content == "async mux"
+        assert msg.content == [{"type": "text", "text": "async mux", "index": 0}]
         await mux.aclose()
 
 
@@ -545,7 +562,7 @@ class TestEndToEnd:
 
         assert len(streams) == 1
         assert isinstance(streams[0], ChatModelStream)
-        assert streams[0].output.content == "hello world"
+        assert str(streams[0].text) == "hello world"
 
     def test_node_stream_v2_text_deltas_iterate(self) -> None:
         """Consumer can iterate `.text` on the streamed message in real time."""
@@ -588,7 +605,7 @@ class TestEndToEnd:
         streams = list(run.messages)
 
         assert len(streams) == 1
-        assert streams[0].output.content == "hardcoded"
+        assert str(streams[0].text) == "hardcoded"
 
     @pytest.mark.anyio
     async def test_async_node_calling_astream_v2(self) -> None:
@@ -616,7 +633,7 @@ class TestEndToEnd:
         assert len(streams) == 1
         assert isinstance(streams[0], AsyncChatModelStream)
         msg = await streams[0].output
-        assert msg.content == "async answer"
+        assert msg.content == [{"type": "text", "text": "async answer", "index": 0}]
 
     @pytest.mark.anyio
     async def test_nested_async_iteration_yields_text_deltas(self) -> None:
@@ -693,7 +710,7 @@ class TestEndToEndV2Invoke:
         )
         stream = streams[0]
         assert isinstance(stream, ChatModelStream)
-        assert stream.output.content == "hello world"
+        assert str(stream.text) == "hello world"
 
     def test_invoke_v2_emits_protocol_events(self) -> None:
         """Iterating the stream yields the full v2 lifecycle (not v1 chunks)."""
@@ -726,7 +743,7 @@ class TestEndToEndV2Invoke:
             assert isinstance(event, dict)
             assert "event" in event
         # Typed projection still assembles the final text.
-        assert stream.output.content == "streamed answer"
+        assert str(stream.text) == "streamed answer"
 
     def test_invoke_text_deltas_iterate_live(self) -> None:
         """`.text` projection yields deltas in order."""
@@ -774,7 +791,7 @@ class TestEndToEndV2Invoke:
         streams = list(run.messages)
 
         assert len(streams) == 2
-        contents = {s.output.content for s in streams}
+        contents = {str(s.text) for s in streams}
         assert contents == {"alpha", "beta"}
 
     def test_invoke_plus_constructed_message_two_streams(self) -> None:
@@ -805,9 +822,9 @@ class TestEndToEndV2Invoke:
 
         assert len(streams) == 2
         assert streams[0].node == "streaming_node"
-        assert streams[0].output.content == "live stream"
+        assert str(streams[0].text) == "live stream"
         assert streams[1].node == "constructed_node"
-        assert streams[1].output.content == "hardcoded"
+        assert str(streams[1].text) == "hardcoded"
         assert streams[1].message_id == "constructed-1"
 
     @pytest.mark.anyio
@@ -835,7 +852,7 @@ class TestEndToEndV2Invoke:
         assert len(streams) == 1
         assert isinstance(streams[0], AsyncChatModelStream)
         msg = await streams[0].output
-        assert msg.content == "async invoke"
+        assert msg.content == [{"type": "text", "text": "async invoke", "index": 0}]
 
 
 class TestDirectMessagesModeStaysV1:

libs/langgraph/tests/test_stream_subgraph_transformer.py

@@ -237,7 +237,14 @@ class TestSubgraphTransformerUnit:
             {
                 "type": "event",
                 "method": "messages",
-                "params": {"namespace": ["t:c"], "timestamp": TS, "data": "x"},
+                "params": {
+                    "namespace": ["t:c"],
+                    "timestamp": TS,
+                    "data": (
+                        {"event": "message-start", "message_id": "m1"},
+                        {"run_id": "m1"},
+                    ),
+                },
             }
         )
         assert list(transformer._root_log._items) == []