diff --git a/libs/langgraph/langgraph/pregel/_loop.py b/libs/langgraph/langgraph/pregel/_loop.py index 3583b6ff1..0dd5999fa 100644 --- a/libs/langgraph/langgraph/pregel/_loop.py +++ b/libs/langgraph/langgraph/pregel/_loop.py @@ -114,6 +114,7 @@ from langgraph.types import ( CachePolicy, Command, Durability, + Interrupt, PregelExecutableTask, RetryPolicy, StreamMode, @@ -248,6 +249,7 @@ 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) @@ -316,14 +318,19 @@ 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: - # remove existing writes for this task + 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 self.checkpoint_pending_writes = [ w for w in self.checkpoint_pending_writes if w[0] != task_id - ] - writes_to_save = writes - # save writes - self.checkpoint_pending_writes.extend((task_id, c, v) for c, v in writes) + ] + [(task_id, c, v) for c, v in writes_to_save] + if self.durability != "exit" and self.checkpointer_put_writes is not None: config = patch_configurable( self.checkpoint_config, @@ -471,6 +478,20 @@ 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( @@ -516,9 +537,45 @@ 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 @@ -570,34 +627,53 @@ 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 ids - pending_interrupts: dict[str, str] = {} + # 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] = {} - # set of resume task ids - pending_resumes: set[str] = set() + 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) - 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 + # keep only interrupt ids where resume_count < interrupt_count hanging_interrupts: set[str] = { interrupt_id - for interrupt_id in pending_interrupts.values() - if interrupt_id not in resumed_interrupt_ids + for task_id, (interrupt_id, interrupt_count) in pending_interrupts.items() + if pending_resumes.get(task_id, 0) < interrupt_count } 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: @@ -1026,6 +1102,7 @@ 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 diff --git a/libs/langgraph/langgraph/pregel/main.py b/libs/langgraph/langgraph/pregel/main.py index 3454a8169..70a3ba68c 100644 --- a/libs/langgraph/langgraph/pregel/main.py +++ b/libs/langgraph/langgraph/pregel/main.py @@ -2672,7 +2672,11 @@ 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], + [ + t + for t in loop.tasks.values() + if not t.writes and t.id not in loop.skipped_task_ids + ], timeout=self.step_timeout, get_waiter=get_waiter, schedule_task=loop.accept_push, @@ -2991,7 +2995,11 @@ 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], + [ + t + for t in loop.tasks.values() + if not t.writes and t.id not in loop.skipped_task_ids + ], timeout=self.step_timeout, get_waiter=get_waiter, schedule_task=loop.aaccept_push, diff --git a/libs/langgraph/tests/test_interruption.py b/libs/langgraph/tests/test_interruption.py index a484d74e5..4bd30124d 100644 --- a/libs/langgraph/tests/test_interruption.py +++ b/libs/langgraph/tests/test_interruption.py @@ -1,12 +1,21 @@ +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 Durability +from langgraph.types import Command, Durability, Send, interrupt 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 @@ -90,3 +99,486 @@ 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