From 1bf0f1b7f8e839464af043c77633e4ef8bcb8437 Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Thu, 23 Apr 2026 09:49:25 -0400 Subject: [PATCH] Match MessagesTransformer scope filter to JS root-feed semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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:",)), 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). --- .../langgraph/stream/transformers.py | 18 +++- libs/langgraph/tests/test_pregel_stream_v2.py | 14 ++-- .../tests/test_stream_messages_transformer.py | 83 +++++++++++-------- .../tests/test_stream_subgraph_transformer.py | 9 +- 4 files changed, 84 insertions(+), 40 deletions(-) diff --git a/libs/langgraph/langgraph/stream/transformers.py b/libs/langgraph/langgraph/stream/transformers.py index 16e994cf4..f5ac39bf5 100644 --- a/libs/langgraph/langgraph/stream/transformers.py +++ b/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") diff --git a/libs/langgraph/tests/test_pregel_stream_v2.py b/libs/langgraph/tests/test_pregel_stream_v2.py index 137413873..233c40498 100644 --- a/libs/langgraph/tests/test_pregel_stream_v2.py +++ b/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() diff --git a/libs/langgraph/tests/test_stream_messages_transformer.py b/libs/langgraph/tests/test_stream_messages_transformer.py index 3164e7147..f7e47da96 100644 --- a/libs/langgraph/tests/test_stream_messages_transformer.py +++ b/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: diff --git a/libs/langgraph/tests/test_stream_subgraph_transformer.py b/libs/langgraph/tests/test_stream_subgraph_transformer.py index 790133a54..6a137a6ef 100644 --- a/libs/langgraph/tests/test_stream_subgraph_transformer.py +++ b/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) == []