From 0c73af562490685665fe324cae17ebcd35ae410f Mon Sep 17 00:00:00 2001 From: Caspar Broekhuizen Date: Tue, 30 Sep 2025 11:28:41 -0700 Subject: [PATCH] fix(langgraph): revert -- reuse cached writes on nested resume to prevent task re-execution (#6227) Reverts langchain-ai/langgraph#6161 --- libs/langgraph/langgraph/pregel/_loop.py | 4 +- libs/langgraph/tests/test_pregel.py | 67 ------------------------ 2 files changed, 1 insertion(+), 70 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/_loop.py b/libs/langgraph/langgraph/pregel/_loop.py index 023ac5545..3583b6ff1 100644 --- a/libs/langgraph/langgraph/pregel/_loop.py +++ b/libs/langgraph/langgraph/pregel/_loop.py @@ -242,9 +242,7 @@ class PregelLoop: self.interrupt_before = interrupt_before self.manager = manager self.is_nested = CONFIG_KEY_TASK_ID in self.config.get(CONF, {}) - self.skip_done_tasks = CONFIG_KEY_CHECKPOINT_ID not in config[CONF] or ( - CONFIG_KEY_RESUMING in self.config[CONF] and self.is_nested - ) + self.skip_done_tasks = CONFIG_KEY_CHECKPOINT_ID not in config[CONF] self._migrate_checkpoint = migrate_checkpoint self.trigger_to_nodes = trigger_to_nodes self.retry_policy = retry_policy diff --git a/libs/langgraph/tests/test_pregel.py b/libs/langgraph/tests/test_pregel.py index 09274782c..b9d2b641c 100644 --- a/libs/langgraph/tests/test_pregel.py +++ b/libs/langgraph/tests/test_pregel.py @@ -3445,73 +3445,6 @@ def test_stream_buffering_single_node(sync_checkpointer: BaseCheckpointSaver) -> ] -def test_nested_graph_resume_reuses_cached_task_writes( - sync_checkpointer: BaseCheckpointSaver, -) -> None: - # Reproduces issue where a helper @task inside a nested graph re-executes - # on resume instead of reusing cached writes. Ensures it runs only once. - counter_parent = 0 - counter_sub = 0 - - @task - def get_time_parent() -> float: - nonlocal counter_parent - counter_parent += 1 - return time.time() - - @task - def get_time_subgraph() -> float: - nonlocal counter_sub - counter_sub += 1 - return time.time() - - class State(TypedDict): - state_counter: int - - # Subgraph that calls a helper task and then interrupts - sub = StateGraph(State) - - def human_node(_: State): - _ = get_time_subgraph().result() - interrupt("what is your name?") - - sub.add_node("human_node", human_node) - sub.set_entry_point("human_node") - sub.set_finish_point("human_node") - subgraph = sub.compile(checkpointer=sync_checkpointer) - - # Parent graph that calls a helper task and interrupts, then enters subgraph - parent = StateGraph(State) - - def parent_node(_: State): - _ = get_time_parent().result() - interrupt("what is your parent name?") - - parent.add_node("parent_node", parent_node) - parent.add_node("subgraph", subgraph) - parent.add_edge(START, "parent_node") - parent.add_edge("parent_node", "subgraph") - parent.add_edge("subgraph", END) - graph = parent.compile(checkpointer=sync_checkpointer) - - cfg_parent = {"configurable": {"thread_id": str(uuid.uuid4())}} - - # First run – interrupts in parent node - for _ in graph.stream({"state_counter": 1}, cfg_parent): - pass - - # Resume 1 – proceeds into subgraph, interrupts there - for _ in graph.stream(Command(resume="resume-1"), cfg_parent): - pass - - # Resume 2 – completes without re-running subgraph helper task - for _ in graph.stream(Command(resume="resume-2"), cfg_parent): - pass - - assert counter_parent == 1 - assert counter_sub == 1 - - def test_nested_graph_interrupts_parallel( sync_checkpointer: BaseCheckpointSaver, durability: Durability ) -> None: