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

@@ -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

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: