From 8dc4465d059c9e7929c116c0423ca490cc887e49 Mon Sep 17 00:00:00 2001
From: jito <pigberger70@gmail.com>
Date: Thu, 18 Sep 2025 04:35:07 +0900
Subject: [PATCH] fix(langgraph): reuse cached writes on nested resume to
 prevent task re-execution (#6161)

**Description**: fix #6050.

Root cause: In nested graphs, the first tick after resume often included
a checkpoint_id, which set skip_done_tasks=False. This skipped matching
pending writes and re-executed already-completed helper @task on
subsequent resumes.

Change: Initialize skip_done_tasks=True when resuming inside a nested
graph. Use original config[CONF] for checkpoint_id presence, and
self.config[CONF] for resuming (current loop state). Added a concise
comment clarifying the different config sources.

**Issue**: #6050

**Tests**:
Add regression test `test_nested_graph_resume_reuses_cached_task_writes`

---------

Signed-off-by: jitokim <pigberger70@gmail.com>
Co-authored-by: Caspar Broekhuizen <casparbroekhuizen@gmail.com>
---
 libs/langgraph/langgraph/pregel/_loop.py |  4 +-
 libs/langgraph/tests/test_pregel.py      | 67 ++++++++++++++++++++++++
 2 files changed, 70 insertions(+), 1 deletion(-)

diff --git a/libs/langgraph/langgraph/pregel/_loop.py b/libs/langgraph/langgraph/pregel/_loop.py
index 3583b6ff1..023ac5545 100644
--- a/libs/langgraph/langgraph/pregel/_loop.py
+++ b/libs/langgraph/langgraph/pregel/_loop.py
@@ -242,7 +242,9 @@ 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]
+        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._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 b9d2b641c..09274782c 100644
--- a/libs/langgraph/tests/test_pregel.py
+++ b/libs/langgraph/tests/test_pregel.py
@@ -3445,6 +3445,73 @@ 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: