fix: interrupt stream mode values (#6475)

This PR improves the consistency of interrupt streaming. 

- when streaming with stream_mode values, the stream chunk now contains
the entire state alongside the interrupt:

```python
class State(TypedDict):
    robot_input: str
# at this point in time robot_input is already set to  "beep boop i am a robot"
app.stream(..., stream_mode="values")
# before
{"__interrupt__": (Interrupt(value="interrupt",))}}
# after
{"robot_input": "beep boop i am a robot", "__interrupt__": (Interrupt(value="interrupt"))}
```

- when streaming with stream_mode=["values", "updates"], interrupts are
surfaced in both an update stream chunk and the value stream chunk, when
previously we keep interrupt in values only if we request values mode
only

```python
class State(TypedDict):
    robot_input: str
# at this point in time robot_input is already set to  "beep boop i am a robot"
app.stream(..., stream_mode=["values", "updates"])
# before (interrupt would only emit on update chunk, there would be no values chunk)
("updates", {"__interrupt__": (Interrupt(value="interrupt",))}})
# after
("updates", {"__interrupt__": (Interrupt(value="interrupt",))}})
("values", {"robot_input": "beep boop i am a robot", "__interrupt__": (Interrupt(value="interrupt"))})
```

For housekeeping: this PR improves on this revert:
https://github.com/langchain-ai/langgraph/pull/6141

libs/langgraph/langgraph/pregel/_loop.py

@@ -766,6 +766,7 @@ class PregelLoop:
             ]
             self.checkpoint["channel_values"][TASKS] = sanitized_tasks
         # bail if no checkpointer
+
         if do_checkpoint and self._checkpointer_put_after_previous is not None:
             self.prev_checkpoint_config = (
                 self.checkpoint_config
@@ -935,8 +936,15 @@ class PregelLoop:
                 stream_modes = self.stream.modes if self.stream else []
                 if "updates" in stream_modes:
                     self._emit("updates", lambda: iter(interrupts))
-                elif "values" in stream_modes:
-                    self._emit("values", lambda: iter(interrupts))
+                if "values" in stream_modes:
+                    current_values = read_channels(self.channels, self.output_keys)
+                    # self.output_keys is a sequence, stream chunk contains entire state and interrupts
+                    if isinstance(current_values, dict):
+                        current_values[INTERRUPT] = interrupts[0][INTERRUPT]
+                        self._emit("values", lambda: iter([current_values]))
+                    # self.output_keys is a string, stream chunk contains only interrupts
+                    else:
+                        self._emit("values", lambda: iter(interrupts))
             elif writes[0][0] != ERROR:
                 self._emit(
                     "updates",

libs/langgraph/tests/test_pregel.py

@@ -8422,23 +8422,55 @@ def test_null_resume_disallowed_with_multiple_interrupts(
     }
 
 
-def test_interrupt_stream_mode_values():
-    """Test that interrupts are surfaced when steam_mode='values'"""
+def test_interrupt_stream_mode_values(sync_checkpointer: BaseCheckpointSaver):
+    """Test that interrupts are surfaced on 'values' stream mode"""
 
     class State(TypedDict):
+        robot_input: str
         human_input: str
 
+    def robot_input_node(state: State) -> State:
+        return {"robot_input": "beep boop i am a robot"}
+
     def human_input_node(state: State) -> Command:
         human_input = interrupt("interrupt")
         return Command(update={"human_input": human_input})
 
     builder = StateGraph(State)
+    builder.add_node(robot_input_node)
     builder.add_node(human_input_node)
-    builder.add_edge(START, "human_input_node")
-    app = builder.compile()
+    builder.add_edge(START, "robot_input_node")
+    builder.add_edge("robot_input_node", "human_input_node")
+    app = builder.compile(checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": str(uuid.uuid4())}}
 
-    result = [*app.stream(State(), stream_mode="values")]
-    assert "__interrupt__" in result[-1]
+    result = [*app.stream(State(), config, stream_mode=["updates", "values"])]
+    assert len(result) == 4
+    assert result == [
+        ("updates", {"robot_input_node": {"robot_input": "beep boop i am a robot"}}),
+        ("values", {"robot_input": "beep boop i am a robot"}),
+        ("updates", {"__interrupt__": (Interrupt(value="interrupt", id=AnyStr()),)}),
+        (
+            "values",
+            {
+                "robot_input": "beep boop i am a robot",
+                "__interrupt__": (Interrupt(value="interrupt", id=AnyStr()),),
+            },
+        ),
+    ]
+    resume_result = [
+        *app.stream(
+            Command(resume="i am a human"), config, stream_mode=["updates", "values"]
+        )
+    ]
+    assert resume_result == [
+        ("values", {"robot_input": "beep boop i am a robot"}),
+        ("updates", {"human_input_node": {"human_input": "i am a human"}}),
+        (
+            "values",
+            {"robot_input": "beep boop i am a robot", "human_input": "i am a human"},
+        ),
+    ]
 
 
 def test_supersteps_populate_task_results(

libs/langgraph/tests/test_pregel_async.py

@@ -9196,6 +9196,64 @@ async def test_astream_waiter_cleanup_on_cancel(
     assert all(t.done() for t in recorded_tasks)
 
 
+@NEEDS_CONTEXTVARS
+async def test_interrupt_stream_mode_values(async_checkpointer: BaseCheckpointSaver):
+    """Test that interrupts are surfaced on 'values' stream mode"""
+
+    class State(TypedDict):
+        robot_input: str
+        human_input: str
+
+    def robot_input_node(state: State) -> State:
+        return {"robot_input": "beep boop i am a robot"}
+
+    def human_input_node(state: State) -> Command:
+        human_input = interrupt("interrupt")
+        return Command(update={"human_input": human_input})
+
+    builder = StateGraph(State)
+    builder.add_node(robot_input_node)
+    builder.add_node(human_input_node)
+    builder.add_edge(START, "robot_input_node")
+    builder.add_edge("robot_input_node", "human_input_node")
+    app = builder.compile(checkpointer=async_checkpointer)
+    config = {"configurable": {"thread_id": str(uuid.uuid4())}}
+
+    result = [
+        (mode, e)
+        async for mode, e in app.astream(
+            State(), config, stream_mode=["updates", "values"]
+        )
+    ]
+    assert len(result) == 4
+    assert result == [
+        ("updates", {"robot_input_node": {"robot_input": "beep boop i am a robot"}}),
+        ("values", {"robot_input": "beep boop i am a robot"}),
+        ("updates", {"__interrupt__": (Interrupt(value="interrupt", id=AnyStr()),)}),
+        (
+            "values",
+            {
+                "robot_input": "beep boop i am a robot",
+                "__interrupt__": (Interrupt(value="interrupt", id=AnyStr()),),
+            },
+        ),
+    ]
+    resume_result = [
+        (mode, e)
+        async for mode, e in app.astream(
+            Command(resume="i am a human"), config, stream_mode=["updates", "values"]
+        )
+    ]
+    assert resume_result == [
+        ("values", {"robot_input": "beep boop i am a robot"}),
+        ("updates", {"human_input_node": {"human_input": "i am a human"}}),
+        (
+            "values",
+            {"robot_input": "beep boop i am a robot", "human_input": "i am a human"},
+        ),
+    ]
+
+
 async def test_supersteps_populate_task_results(
     async_checkpointer: BaseCheckpointSaver,
 ) -> None:

libs/prebuilt/tests/test_react_agent.py

@@ -638,8 +638,9 @@ def test_react_agent_parallel_tool_calls(
     for event in agent.stream(
         {"messages": [("user", query)]}, config, stream_mode="values"
     ):
-        if messages := event.get("messages"):
-            message_types.append([m.type for m in messages])
+        if "__interrupt__" not in event: 
+            if messages := event.get("messages"):
+                message_types.append([m.type for m in messages])
 
     if version == "v1":
         assert message_types == [