add overwrite command

- Bump langgraph version to 0.6.10

libs/langgraph/langgraph/pregel/__init__.py

@@ -1,3 +1,4 @@
+from langgraph.pregel._write import Overwrite, overwrite
 from langgraph.pregel.main import NodeBuilder, Pregel
 
-__all__ = ("Pregel", "NodeBuilder")
+__all__ = ("Pregel", "NodeBuilder", "overwrite", "Overwrite")

libs/langgraph/langgraph/pregel/_algo.py

@@ -71,6 +71,7 @@ from langgraph.pregel._call import get_runnable_for_task, identifier
 from langgraph.pregel._io import read_channels
 from langgraph.pregel._log import logger
 from langgraph.pregel._read import INPUT_CACHE_KEY_TYPE, PregelNode
+from langgraph.pregel._write import _Overwrite
 from langgraph.runtime import DEFAULT_RUNTIME, Runtime
 from langgraph.types import (
     All,
@@ -198,8 +199,16 @@ def local_read(
         # apply writes
         local_channels: dict[str, BaseChannel] = {}
         for k in channels:
-            cc = channels[k].copy()
-            cc.update(updated[k])
+            if updated[k]:
+                # If any overwrite is present for this channel, reflect it directly
+                ow = next((v for v in updated[k] if isinstance(v, _Overwrite)), None)
+                if ow is not None:
+                    cc = channels[k].from_checkpoint(ow.value)
+                else:
+                    cc = channels[k].copy()
+                    cc.update(updated[k])
+            else:
+                cc = channels[k].copy()
             local_channels[k] = cc
         # read fresh values
         values = read_channels(local_channels, select)
@@ -277,12 +286,16 @@ def apply_writes(
 
     # Group writes by channel
     pending_writes_by_channel: dict[str, list[Any]] = defaultdict(list)
+    overwrite_by_channel: dict[str, Any] = {}
     for task in tasks:
         for chan, val in task.writes:
             if chan in (NO_WRITES, PUSH, RESUME, INTERRUPT, RETURN, ERROR):
-                pass
-            elif chan in channels:
-                pending_writes_by_channel[chan].append(val)
+                continue
+            if chan in channels:
+                if isinstance(val, _Overwrite):
+                    overwrite_by_channel[chan] = val.value
+                else:
+                    pending_writes_by_channel[chan].append(val)
             else:
                 logger.warning(
                     f"Task {task.name} with path {task.path} wrote to unknown channel {chan}, ignoring it."
@@ -290,13 +303,26 @@ def apply_writes(
 
     # Apply writes to channels
     updated_channels: set[str] = set()
-    for chan, vals in pending_writes_by_channel.items():
+    for chan in set(pending_writes_by_channel.keys()) | set(
+        overwrite_by_channel.keys()
+    ):
         if chan in channels:
-            if channels[chan].update(vals) and next_version is not None:
-                checkpoint["channel_versions"][chan] = next_version
-                # unavailable channels can't trigger tasks, so don't add them
+            if chan in overwrite_by_channel:
+                # Overwrite the entire channel value, bypassing reducers.
+                channels[chan] = channels[chan].from_checkpoint(
+                    overwrite_by_channel[chan]
+                )
+                if next_version is not None:
+                    checkpoint["channel_versions"][chan] = next_version
                 if channels[chan].is_available():
                     updated_channels.add(chan)
+            else:
+                vals = pending_writes_by_channel.get(chan, [])
+                if channels[chan].update(vals) and next_version is not None:
+                    checkpoint["channel_versions"][chan] = next_version
+                    # unavailable channels can't trigger tasks, so don't add them
+                    if channels[chan].is_available():
+                        updated_channels.add(chan)
 
     # Channels that weren't updated in this step are notified of a new step
     if bump_step:

libs/langgraph/langgraph/pregel/_loop.py

@@ -114,7 +114,6 @@ from langgraph.types import (
     CachePolicy,
     Command,
     Durability,
-    Interrupt,
     PregelExecutableTask,
     RetryPolicy,
     StreamMode,
@@ -249,7 +248,6 @@ class PregelLoop:
         self.retry_policy = retry_policy
         self.cache_policy = cache_policy
         self.durability = durability
-        self.skipped_task_ids: set[str] = set()
         if self.stream is not None and CONFIG_KEY_STREAM in config[CONF]:
             self.stream = DuplexStream(self.stream, config[CONF][CONFIG_KEY_STREAM])
         scratchpad: PregelScratchpad | None = config[CONF].get(CONFIG_KEY_SCRATCHPAD)
@@ -318,19 +316,14 @@ class PregelLoop:
             writes_to_save: WritesT = [
                 w[1:] for w in self.checkpoint_pending_writes if w[0] == task_id
             ] + list(writes)
-            self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
         else:
-            writes_to_save = [
-                # aggregate existing interrupts for this task
-                (ch, self._merge_interrupts(task_id, v) if ch == INTERRUPT else v)
-                for ch, v in writes
-            ]
-
-            # replace all writes for this task_id in one shot
+            # remove existing writes for this task
             self.checkpoint_pending_writes = [
                 w for w in self.checkpoint_pending_writes if w[0] != task_id
-            ] + [(task_id, c, v) for c, v in writes_to_save]
-
+            ]
+            writes_to_save = writes
+        # save writes
+        self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes)
         if self.durability != "exit" and self.checkpointer_put_writes is not None:
             config = patch_configurable(
                 self.checkpoint_config,
@@ -478,20 +471,6 @@ class PregelLoop:
             cache_policy=self.cache_policy,
         )
 
-        resume_map = self.config.get(CONF, {}).get(CONFIG_KEY_RESUME_MAP, {})
-        if resume_map:
-            skipped_interrupt_ids = self._pending_interrupts() - set(resume_map)
-            self.skipped_task_ids = {
-                task_id
-                for task_id, channel, value in self.checkpoint_pending_writes
-                if channel == INTERRUPT
-                # interrupts within a task are uncovered sequentially as resumes are provided,
-                # so we only need to check the last interrupt id
-                and value[-1].id in skipped_interrupt_ids
-            }
-        else:
-            self.skipped_task_ids = set()
-
         # produce debug output
         if self._checkpointer_put_after_previous is not None:
             self._emit(
@@ -537,45 +516,9 @@ class PregelLoop:
             if task.writes:
                 self.output_writes(task.id, task.writes, cached=True)
 
-        if self.skipped_task_ids:
-            # remove tasks with writes that may have been matched from previous loop
-            self.skipped_task_ids = {
-                task_id
-                for task_id in self.skipped_task_ids
-                if not self.tasks[task_id].writes
-            }
-            # output interrupt writes for blocked tasks so they are still visible in the stream
-            for task_id, channel, value in self.checkpoint_pending_writes:
-                if task_id in self.skipped_task_ids and channel == INTERRUPT:
-                    # find resume count for this task
-                    resumes = next(
-                        (
-                            v
-                            for tid, ch, v in self.checkpoint_pending_writes
-                            if tid == task_id and ch == RESUME
-                        ),
-                        None,
-                    )
-                    resume_count = len(resumes) if resumes is not None else 0
-                    # only output unresumed interrupts
-                    if resume_count < len(value):
-                        self.output_writes(task_id, [(INTERRUPT, value[resume_count:])])
-
         return True
 
     def after_tick(self) -> None:
-        if self.skipped_task_ids:
-            # raise early GraphInterrupt for skipped tasks.
-            # since we know len(resumes) != len(interrupts) for these tasks, we
-            # can prevent unnecessary node re-execution by raising preemptively
-            interrupts = []
-            for task_id, channel, value in self.checkpoint_pending_writes:
-                if channel == INTERRUPT and task_id in self.skipped_task_ids:
-                    interrupts.extend(value)
-            if interrupts:
-                raise GraphInterrupt(interrupts)
-
-        self.skipped_task_ids.clear()
         # finish superstep
         writes = [w for t in self.tasks.values() for w in t.writes]
         # all tasks have finished
@@ -627,53 +570,34 @@ class PregelLoop:
 
     def _pending_interrupts(self) -> set[str]:
         """Return the set of interrupt ids that are pending without corresponding resume values."""
-        # mapping of task ids to (interrupt_id, interrupt_count)
-        pending_interrupts: dict[str, tuple[str, int]] = {}
-        # mapping of task ids to resume count
-        pending_resumes: dict[str, int] = {}
+        # mapping of task ids to interrupt ids
+        pending_interrupts: dict[str, str] = {}
 
-        for task_id, channel, value in self.checkpoint_pending_writes:
-            if channel == INTERRUPT:
-                pending_interrupts[task_id] = (
-                    value[0].id,
-                    len(value),
-                )
-            elif channel == RESUME:
-                resume_list = value if isinstance(value, list) else [value]
-                pending_resumes[task_id] = len(resume_list)
+        # set of resume task ids
+        pending_resumes: set[str] = set()
 
-        # keep only interrupt ids where resume_count < interrupt_count
+        for task_id, write_type, value in self.checkpoint_pending_writes:
+            if write_type == INTERRUPT:
+                # interrupts is always a list, but there should only be one element
+                pending_interrupts[task_id] = value[0].id
+            elif write_type == RESUME:
+                pending_resumes.add(task_id)
+
+        resumed_interrupt_ids = {
+            pending_interrupts[task_id]
+            for task_id in pending_resumes
+            if task_id in pending_interrupts
+        }
+
+        # Keep only interrupts whose interrupt_id is not resumed
         hanging_interrupts: set[str] = {
             interrupt_id
-            for task_id, (interrupt_id, interrupt_count) in pending_interrupts.items()
-            if pending_resumes.get(task_id, 0) < interrupt_count
+            for interrupt_id in pending_interrupts.values()
+            if interrupt_id not in resumed_interrupt_ids
         }
 
         return hanging_interrupts
 
-    def _merge_interrupts(
-        self, task_id: str, value: Sequence[Interrupt]
-    ) -> Sequence[Interrupt]:
-        """Normalize interrupt value to list and merge with existing interrupts.
-
-        If the interrupt ID matches existing, append; otherwise replace.
-
-        Returns list of Interrupt objects for this task.
-        """
-        new = value if isinstance(value, list) else list(value)
-        existing = next(
-            (
-                v
-                for tid, ch, v in self.checkpoint_pending_writes
-                if tid == task_id and ch == INTERRUPT
-            ),
-            None,
-        )
-        if existing is None:
-            return new
-        old = existing if isinstance(existing, list) else list(existing)
-        return old + new if old and new and old[0].id == new[0].id else new
-
     def _first(
         self, *, input_keys: str | Sequence[str], updated_channels: set[str] | None
     ) -> set[str] | None:
@@ -1102,7 +1026,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
 
     def put_writes(self, task_id: str, writes: WritesT) -> None:
         """Put writes for a task, to be read by the next tick."""
-
         super().put_writes(task_id, writes)
         if not writes or self.cache is None or not hasattr(self, "tasks"):
             return

libs/langgraph/langgraph/pregel/_write.py

@@ -26,6 +26,32 @@ SKIP_WRITE = object()
 PASSTHROUGH = object()
 
 
+class _Overwrite:
+    """Marker wrapper indicating a direct channel overwrite.
+
+    Use via `overwrite(channel, value)` or `Overwrite(value)`.
+    """
+
+    __slots__ = ("value",)
+
+    def __init__(self, value: Any):
+        self.value = value
+
+
+def Overwrite(value: Any) -> _Overwrite:
+    """Wrap a value to force overwrite a channel, bypassing reducers."""
+    return _Overwrite(value)
+
+
+def overwrite(channel: str, value: Any) -> ChannelWriteEntry:
+    """Convenience factory for a write that overwrites the target channel.
+
+    Example:
+        NodeBuilder().write_to(overwrite("foo", 123))
+    """
+    return ChannelWriteEntry(channel, Overwrite(value))
+
+
 class ChannelWriteEntry(NamedTuple):
     channel: str
     """Channel name to write to."""

libs/langgraph/langgraph/pregel/main.py

@@ -2672,11 +2672,7 @@ class Pregel(
                     for task in loop.match_cached_writes():
                         loop.output_writes(task.id, task.writes, cached=True)
                     for _ in runner.tick(
-                        [
-                            t
-                            for t in loop.tasks.values()
-                            if not t.writes and t.id not in loop.skipped_task_ids
-                        ],
+                        [t for t in loop.tasks.values() if not t.writes],
                         timeout=self.step_timeout,
                         get_waiter=get_waiter,
                         schedule_task=loop.accept_push,
@@ -2995,11 +2991,7 @@ class Pregel(
                         for task in await loop.amatch_cached_writes():
                             loop.output_writes(task.id, task.writes, cached=True)
                         async for _ in runner.atick(
-                            [
-                                t
-                                for t in loop.tasks.values()
-                                if not t.writes and t.id not in loop.skipped_task_ids
-                            ],
+                            [t for t in loop.tasks.values() if not t.writes],
                             timeout=self.step_timeout,
                             get_waiter=get_waiter,
                             schedule_task=loop.aaccept_push,

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "0.6.9"
+version = "0.6.10"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.9"

libs/langgraph/tests/test_interruption.py

@@ -1,21 +1,12 @@
-import operator
-import sys
-from typing import Annotated
-
 import pytest
 from langgraph.checkpoint.base import BaseCheckpointSaver
 from typing_extensions import TypedDict
 
 from langgraph.graph import END, START, StateGraph
-from langgraph.types import Command, Durability, Send, interrupt
+from langgraph.types import Durability
 
 pytestmark = pytest.mark.anyio
 
-NEEDS_CONTEXTVARS = pytest.mark.skipif(
-    sys.version_info < (3, 11),
-    reason="Python 3.11+ is required for async contextvars support",
-)
-
 
 def test_interruption_without_state_updates(
     sync_checkpointer: BaseCheckpointSaver, durability: Durability
@@ -99,486 +90,3 @@ async def test_interruption_without_state_updates_async(
     assert (await graph.aget_state(thread)).next == ()
     n_checkpoints = len([c async for c in graph.aget_state_history(thread)])
     assert n_checkpoints == (5 if durability != "exit" else 3)
-
-
-def test_interrupt_with_send_payloads(sync_checkpointer: BaseCheckpointSaver) -> None:
-    """Test interruption in map node with Send payloads and human-in-the-loop resume."""
-
-    # Global counter to track node executions
-    node_counter = {"entry": 0, "map_node": 0}
-
-    class State(TypedDict):
-        items: list[str]
-        processed: Annotated[list[str], operator.add]
-
-    def entry_node(state: State):
-        node_counter["entry"] += 1
-        return {}  # No state updates in entry node
-
-    def send_to_map(state: State):
-        return [Send("map_node", {"item": item}) for item in state["items"]]
-
-    def map_node(state: State):
-        node_counter["map_node"] += 1
-        if "dangerous" in state["item"]:
-            value = interrupt({"processing": state["item"]})
-            return {"processed": [f"processed_{value}"]}
-        else:
-            return {"processed": [f"processed_{state['item']}_auto"]}
-
-    builder = StateGraph(State)
-    builder.add_node("entry", entry_node)
-    builder.add_node("map_node", map_node)
-    builder.add_edge(START, "entry")
-    builder.add_conditional_edges("entry", send_to_map, ["map_node"])
-    builder.add_edge("map_node", END)
-
-    graph = builder.compile(checkpointer=sync_checkpointer)
-
-    config = {"configurable": {"thread_id": "test_interrupt_send"}}
-
-    # Run until interrupts
-    result = graph.invoke(
-        {"items": ["item1", "dangerous_item1", "dangerous_item2"]}, config=config
-    )
-
-    # Verify we have interrupts (only one for dangerous_item)
-    interrupts = result.get("__interrupt__", [])
-    assert len(interrupts) == 2
-    assert "dangerous_item" in interrupts[0].value["processing"]
-
-    # Resume with mapping of interrupt IDs to values
-    resume_map = {i.id: f"human_input_{i.value['processing']}" for i in interrupts}
-
-    final_result = graph.invoke(Command(resume=resume_map), config=config)
-
-    # Verify final result contains processed items
-    assert "processed" in final_result
-    processed_items = final_result["processed"]
-    assert len(processed_items) == 3
-    assert "processed_item1_auto" in processed_items  # item1 processed automatically
-    assert any(
-        "processed_human_input_dangerous_item1" in item for item in processed_items
-    )  # dangerous_item1 processed after interrupt
-    assert any(
-        "processed_human_input_dangerous_item2" in item for item in processed_items
-    )  # dangerous_item2 processed after interrupt
-
-    # Verify node execution counts
-    assert node_counter["entry"] == 1  # Entry node runs once
-    # Map node runs 3 times initially (item1 completes, 2 dangerous_items interrupt),
-    # then 2 times on resume
-    assert node_counter["map_node"] == 5
-
-
-@NEEDS_CONTEXTVARS
-async def test_interrupt_with_send_payloads_async(
-    async_checkpointer: BaseCheckpointSaver, durability: Durability
-) -> None:
-    """Test interruption in map node with Send payloads and human-in-the-loop resume."""
-
-    # Global counter to track node executions
-    node_counter = {"entry": 0, "map_node": 0}
-
-    class State(TypedDict):
-        items: list[str]
-        processed: Annotated[list[str], operator.add]
-
-    def entry_node(state: State):
-        node_counter["entry"] += 1
-        return {}  # No state updates in entry node
-
-    def send_to_map(state: State):
-        return [Send("map_node", {"item": item}) for item in state["items"]]
-
-    def map_node(state: State):
-        node_counter["map_node"] += 1
-        if "dangerous" in state["item"]:
-            value = interrupt({"processing": state["item"]})
-            return {"processed": [f"processed_{value}"]}
-        else:
-            return {"processed": [f"processed_{state['item']}_auto"]}
-
-    builder = StateGraph(State)
-    builder.add_node("entry", entry_node)
-    builder.add_node("map_node", map_node)
-    builder.add_edge(START, "entry")
-    builder.add_conditional_edges("entry", send_to_map, ["map_node"])
-    builder.add_edge("map_node", END)
-
-    graph = builder.compile(checkpointer=async_checkpointer)
-
-    config = {"configurable": {"thread_id": "test_interrupt_send"}}
-
-    # Run until interrupts
-    result = await graph.ainvoke(
-        {"items": ["item1", "dangerous_item1", "dangerous_item2"]}, config=config
-    )
-
-    # Verify we have interrupts (only one for dangerous_item)
-    interrupts = result.get("__interrupt__", [])
-    assert len(interrupts) == 2
-    assert "dangerous_item" in interrupts[0].value["processing"]
-
-    # Resume with mapping of interrupt IDs to values
-    resume_map = {i.id: f"human_input_{i.value['processing']}" for i in interrupts}
-
-    final_result = await graph.ainvoke(Command(resume=resume_map), config=config)
-
-    # Verify final result contains processed items
-    assert "processed" in final_result
-    processed_items = final_result["processed"]
-    assert len(processed_items) == 3
-    assert "processed_item1_auto" in processed_items  # item1 processed automatically
-    assert any(
-        "processed_human_input_dangerous_item1" in item for item in processed_items
-    )  # dangerous_item1 processed after interrupt
-    assert any(
-        "processed_human_input_dangerous_item2" in item for item in processed_items
-    )  # dangerous_item2 processed after interrupt
-
-    # Verify node execution counts
-    assert node_counter["entry"] == 1  # Entry node runs once
-    # Map node runs 3 times initially (item1 completes, 2 dangerous_items interrupt),
-    # then 2 times on resume
-    assert node_counter["map_node"] == 5
-
-
-def test_interrupt_with_send_payloads_sequential_resume(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Test interruption in map node with Send payloads and sequential resume."""
-
-    # Global counter to track node executions
-    node_counter = {"entry": 0, "map_node": 0}
-
-    class State(TypedDict):
-        items: list[str]
-        processed: Annotated[list[str], operator.add]
-
-    def entry_node(state: State):
-        node_counter["entry"] += 1
-        return {}  # No state updates in entry node
-
-    def send_to_map(state: State):
-        return [Send("map_node", {"item": item}) for item in state["items"]]
-
-    def map_node(state: State):
-        node_counter["map_node"] += 1
-        if "dangerous" in state["item"]:
-            value = interrupt({"processing": state["item"]})
-            return {"processed": [f"processed_{value}"]}
-        else:
-            return {"processed": [f"processed_{state['item']}_auto"]}
-
-    builder = StateGraph(State)
-    builder.add_node("entry", entry_node)
-    builder.add_node("map_node", map_node)
-    builder.add_edge(START, "entry")
-    builder.add_conditional_edges("entry", send_to_map, ["map_node"])
-    builder.add_edge("map_node", END)
-
-    graph = builder.compile(checkpointer=sync_checkpointer)
-
-    config = {"configurable": {"thread_id": "test_interrupt_send_sequential"}}
-
-    # Run until interrupts
-    result = graph.invoke(
-        {"items": ["item1", "dangerous_item1", "dangerous_item2"]}, config=config
-    )
-
-    # Verify we have interrupts
-    interrupts = result.get("__interrupt__", [])
-    assert len(interrupts) == 2
-    assert "dangerous_item" in interrupts[0].value["processing"]
-
-    # Resume first interrupt only
-    first_interrupt = interrupts[0]
-    first_resume_map = {
-        first_interrupt.id: f"human_input_{first_interrupt.value['processing']}"
-    }
-
-    partial_result = graph.invoke(Command(resume=first_resume_map), config=config)
-
-    # Verify we still have one pending interrupt
-    remaining_interrupts = partial_result.get("__interrupt__", [])
-    assert len(remaining_interrupts) == 1
-
-    # Resume second interrupt
-    second_interrupt = remaining_interrupts[0]
-    second_resume_map = {
-        second_interrupt.id: f"human_input_{second_interrupt.value['processing']}"
-    }
-
-    final_result = graph.invoke(Command(resume=second_resume_map), config=config)
-
-    # Verify final result contains processed items
-    assert "processed" in final_result
-    processed_items = final_result["processed"]
-    assert len(processed_items) == 3
-    assert "processed_item1_auto" in processed_items  # item1 processed automatically
-    assert any(
-        "processed_human_input_dangerous_item1" in item for item in processed_items
-    )  # dangerous_item1 processed after interrupt
-    assert any(
-        "processed_human_input_dangerous_item2" in item for item in processed_items
-    )  # dangerous_item2 processed after interrupt
-
-    # Verify node execution counts
-    assert node_counter["entry"] == 1  # Entry node runs once
-    # Map node runs 3 times initially (item1 completes, 2 dangerous_items interrupt),
-    # then 1 time on first resume, then 1 time on second resume
-    assert node_counter["map_node"] == 5
-
-
-@NEEDS_CONTEXTVARS
-async def test_interrupt_with_send_payloads_sequential_resume_async(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Test interruption in map node with Send payloads and sequential resume."""
-
-    # Global counter to track node executions
-    node_counter = {"entry": 0, "map_node": 0}
-
-    class State(TypedDict):
-        items: list[str]
-        processed: Annotated[list[str], operator.add]
-
-    def entry_node(state: State):
-        node_counter["entry"] += 1
-        return {}  # No state updates in entry node
-
-    def send_to_map(state: State):
-        return [Send("map_node", {"item": item}) for item in state["items"]]
-
-    def map_node(state: State):
-        node_counter["map_node"] += 1
-        if "dangerous" in state["item"]:
-            value = interrupt({"processing": state["item"]})
-            return {"processed": [f"processed_{value}"]}
-        else:
-            return {"processed": [f"processed_{state['item']}_auto"]}
-
-    builder = StateGraph(State)
-    builder.add_node("entry", entry_node)
-    builder.add_node("map_node", map_node)
-    builder.add_edge(START, "entry")
-    builder.add_conditional_edges("entry", send_to_map, ["map_node"])
-    builder.add_edge("map_node", END)
-
-    graph = builder.compile(checkpointer=async_checkpointer)
-
-    config = {"configurable": {"thread_id": "test_interrupt_send_sequential"}}
-
-    # Run until interrupts
-    result = await graph.ainvoke(
-        {"items": ["item1", "dangerous_item1", "dangerous_item2"]}, config=config
-    )
-
-    # Verify we have interrupts
-    interrupts = result.get("__interrupt__", [])
-    assert len(interrupts) == 2
-    assert "dangerous_item" in interrupts[0].value["processing"]
-
-    # Resume first interrupt only
-    first_interrupt = interrupts[0]
-    first_resume_map = {
-        first_interrupt.id: f"human_input_{first_interrupt.value['processing']}"
-    }
-
-    partial_result = await graph.ainvoke(
-        Command(resume=first_resume_map), config=config
-    )
-
-    # Verify we still have one pending interrupt
-    remaining_interrupts = partial_result.get("__interrupt__", [])
-    assert len(remaining_interrupts) == 1
-
-    # Resume second interrupt
-    second_interrupt = remaining_interrupts[0]
-    second_resume_map = {
-        second_interrupt.id: f"human_input_{second_interrupt.value['processing']}"
-    }
-
-    final_result = await graph.ainvoke(Command(resume=second_resume_map), config=config)
-
-    # Verify final result contains processed items
-    assert "processed" in final_result
-    processed_items = final_result["processed"]
-    assert len(processed_items) == 3
-    assert "processed_item1_auto" in processed_items  # item1 processed automatically
-    assert any(
-        "processed_human_input_dangerous_item1" in item for item in processed_items
-    )  # dangerous_item1 processed after interrupt
-    assert any(
-        "processed_human_input_dangerous_item2" in item for item in processed_items
-    )  # dangerous_item2 processed after interrupt
-
-    # Verify node execution counts
-    assert node_counter["entry"] == 1  # Entry node runs once
-    # Map node runs 3 times initially (item1 completes, 2 dangerous_items interrupt),
-    # then 1 time on first resume, then 1 time on second resume
-    assert node_counter["map_node"] == 5
-
-
-def test_node_with_multiple_interrupts_requires_full_resume(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Test a number of different resume patterns for a node with multiple interrupts,
-
-    Ensures that a node is not re-executed until valid resume values have been provided to all
-    discovered interrupts"""
-
-    node_counter = 0
-
-    class State(TypedDict):
-        input: str
-
-    def double_interrupt_node(state: State):
-        nonlocal node_counter
-        node_counter += 1
-        first = interrupt("first")
-        second = interrupt("second")
-        third = interrupt("third")
-        return {"input": f"{first}-{second}-{third}"}
-
-    builder = StateGraph(State)
-    builder.add_node("double_interrupt", double_interrupt_node)
-    builder.add_edge(START, "double_interrupt")
-    builder.add_edge("double_interrupt", END)
-
-    graph = builder.compile(checkpointer=sync_checkpointer)
-
-    config = {"configurable": {"thread_id": "test_double_interrupt"}}
-
-    result = graph.invoke({"input": "start"}, config=config)
-
-    interrupts = result.get("__interrupt__", [])
-    assert len(interrupts) == 1
-    first_interrupt = interrupts[0]
-    assert node_counter == 1
-
-    # invoke with an interrupt map that matches double_interrupt_node.
-    # this should execute the node
-    partial = graph.invoke(
-        Command(resume={first_interrupt.id: "human_first"}), config=config
-    )
-    remaining_interrupts = partial.get("__interrupt__", [])
-    assert len(remaining_interrupts) == 1
-    assert remaining_interrupts[0].value == "second"
-    assert node_counter == 2
-
-    # invoke with an interrupt map that DOES NOT match double_interrupt_node.
-    # this should not execute the node because the optimization kicks in
-    partial = graph.invoke(
-        Command(resume={"00000000000000000000000000000000": "nothing_burger"}),
-        config=config,
-    )
-    remaining_interrupts = partial.get("__interrupt__", [])
-    assert len(remaining_interrupts) == 1
-    assert remaining_interrupts[0].value == "second"
-    assert node_counter == 2
-
-    # invoke with None resume. this should execute the node
-    partial = graph.invoke(None, config=config)
-    remaining_interrupts = partial.get("__interrupt__", [])
-    assert len(remaining_interrupts) == 1
-    assert remaining_interrupts[0].value == "second"
-    assert node_counter == 3
-
-    # invoke with nonspecific resume. this should execute the node
-    partial = graph.invoke(Command(resume="human_second"), config=config)
-    remaining_interrupts = partial.get("__interrupt__", [])
-    assert len(remaining_interrupts) == 1
-    print("REMAINING INTERRUPTS: ", remaining_interrupts)
-    assert remaining_interrupts[0].value == "third"
-    assert node_counter == 4
-
-    # finally, invoke with an interrupt map that matches double_interrupt_node.
-    # this should execute the node and all interrupts should be resolved
-    final_result = graph.invoke(Command(resume="human_third"), config=config)
-    assert "input" in final_result
-    assert final_result["input"] == "human_first-human_second-human_third"
-    assert node_counter == 5
-
-
-@NEEDS_CONTEXTVARS
-async def test_node_with_multiple_interrupts_requires_full_resume_async(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Test a number of different resume patterns for a node with multiple interrupts,
-
-    Ensures that a node is not re-executed until valid resume values have been provided to all
-    discovered interrupts"""
-
-    node_counter = 0
-
-    class State(TypedDict):
-        input: str
-
-    def double_interrupt_node(state: State):
-        nonlocal node_counter
-        node_counter += 1
-        first = interrupt("first")
-        second = interrupt("second")
-        third = interrupt("third")
-        return {"input": f"{first}-{second}-{third}"}
-
-    builder = StateGraph(State)
-    builder.add_node("double_interrupt", double_interrupt_node)
-    builder.add_edge(START, "double_interrupt")
-    builder.add_edge("double_interrupt", END)
-
-    graph = builder.compile(checkpointer=async_checkpointer)
-
-    config = {"configurable": {"thread_id": "test_double_interrupt"}}
-
-    result = await graph.ainvoke({"input": "start"}, config=config)
-
-    interrupts = result.get("__interrupt__", [])
-    assert len(interrupts) == 1
-    first_interrupt = interrupts[0]
-    assert node_counter == 1
-
-    # invoke with an interrupt map that matches double_interrupt_node.
-    # this should execute the node
-    partial = await graph.ainvoke(
-        Command(resume={first_interrupt.id: "human_first"}), config=config
-    )
-    remaining_interrupts = partial.get("__interrupt__", [])
-    assert len(remaining_interrupts) == 1
-    assert remaining_interrupts[0].value == "second"
-    assert node_counter == 2
-
-    # invoke with an interrupt map that DOES NOT match double_interrupt_node.
-    # this should not execute the node because the optimization kicks in
-    partial = await graph.ainvoke(
-        Command(resume={"00000000000000000000000000000000": "nothing_burger"}),
-        config=config,
-    )
-    remaining_interrupts = partial.get("__interrupt__", [])
-    assert len(remaining_interrupts) == 1
-    assert remaining_interrupts[0].value == "second"
-    assert node_counter == 2
-
-    # invoke with None resume. this should execute the node
-    partial = await graph.ainvoke(None, config=config)
-    remaining_interrupts = partial.get("__interrupt__", [])
-    assert len(remaining_interrupts) == 1
-    assert remaining_interrupts[0].value == "second"
-    assert node_counter == 3
-
-    # invoke with nonspecific resume. this should execute the node
-    partial = await graph.ainvoke(Command(resume="human_second"), config=config)
-    remaining_interrupts = partial.get("__interrupt__", [])
-    assert len(remaining_interrupts) == 1
-    print("REMAINING INTERRUPTS: ", remaining_interrupts)
-    assert remaining_interrupts[0].value == "third"
-    assert node_counter == 4
-
-    # finally, invoke with an interrupt map that matches double_interrupt_node.
-    # this should execute the node and all interrupts should be resolved
-    final_result = await graph.ainvoke(Command(resume="human_third"), config=config)
-    assert "input" in final_result
-    assert final_result["input"] == "human_first-human_second-human_third"
-    assert node_counter == 5

libs/langgraph/uv.lock

@@ -1428,7 +1428,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "0.6.9"
+version = "0.6.10"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/uv.lock

@@ -257,7 +257,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "0.6.9"
+version = "0.6.10"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },