push

The loop already merges __interrupt__ into the values dict in
output_writes(), so invoke/ainvoke don't need to subscribe to
both "updates" and "values" stream modes. Simplify to just use
stream_mode="values" and read interrupts directly from the
values payload.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

libs/langgraph/langgraph/pregel/main.py

@@ -3091,13 +3091,13 @@ class Pregel(
         chunks: list[dict[str, Any] | Any] = []
         interrupts: list[Interrupt] = []
 
+        subgraphs = kwargs.get("subgraphs", False)
+
         for chunk in self.stream(
             input,
             config,
             context=context,
-            stream_mode=["updates", "values"]
+            stream_mode="values" if stream_mode == "values" else stream_mode,
-            if stream_mode == "values"
-            else stream_mode,
             print_mode=print_mode,
             output_keys=output_keys,
             interrupt_before=interrupt_before,
@@ -3106,20 +3106,16 @@ class Pregel(
             **kwargs,
         ):
             if stream_mode == "values":
-                if len(chunk) == 2:
+                # when subgraphs=True with a single stream_mode, stream()
-                    mode, payload = cast(tuple[StreamMode, Any], chunk)
+                # yields (namespace, payload) tuples — unwrap them
-                else:
+                if subgraphs and isinstance(chunk, tuple):
-                    _, mode, payload = cast(
+                    chunk = chunk[-1]
-                        tuple[tuple[str, ...], StreamMode, Any], chunk
+                latest = chunk
-                    )
                 if (
-                    mode == "updates"
+                    isinstance(chunk, dict)
-                    and isinstance(payload, dict)
+                    and (ints := chunk.get(INTERRUPT)) is not None
-                    and (ints := payload.get(INTERRUPT)) is not None
                 ):
                     interrupts.extend(ints)
-                elif mode == "values":
-                    latest = payload
             else:
                 chunks.append(chunk)
 
@@ -3181,13 +3177,13 @@ class Pregel(
         chunks: list[dict[str, Any] | Any] = []
         interrupts: list[Interrupt] = []
 
+        subgraphs = kwargs.get("subgraphs", False)
+
         async for chunk in self.astream(
             input,
             config,
             context=context,
-            stream_mode=["updates", "values"]
+            stream_mode="values" if stream_mode == "values" else stream_mode,
-            if stream_mode == "values"
-            else stream_mode,
             print_mode=print_mode,
             output_keys=output_keys,
             interrupt_before=interrupt_before,
@@ -3196,20 +3192,16 @@ class Pregel(
             **kwargs,
         ):
             if stream_mode == "values":
-                if len(chunk) == 2:
+                # when subgraphs=True with a single stream_mode, astream()
-                    mode, payload = cast(tuple[StreamMode, Any], chunk)
+                # yields (namespace, payload) tuples — unwrap them
-                else:
+                if subgraphs and isinstance(chunk, tuple):
-                    _, mode, payload = cast(
+                    chunk = chunk[-1]
-                        tuple[tuple[str, ...], StreamMode, Any], chunk
+                latest = chunk
-                    )
                 if (
-                    mode == "updates"
+                    isinstance(chunk, dict)
-                    and isinstance(payload, dict)
+                    and (ints := chunk.get(INTERRUPT)) is not None
-                    and (ints := payload.get(INTERRUPT)) is not None
                 ):
                     interrupts.extend(ints)
-                elif mode == "values":
-                    latest = payload
             else:
                 chunks.append(chunk)