rename

libs/langgraph/langgraph/_internal/_constants.py

@@ -41,6 +41,8 @@ CONFIG_KEY_CACHE = sys.intern("__pregel_cache")
 # holds a `BaseCache` made available to subgraphs
 CONFIG_KEY_RESUMING = sys.intern("__pregel_resuming")
 # holds a boolean indicating if subgraphs should resume from a previous checkpoint
+CONFIG_KEY_REPLAYING = sys.intern("__pregel_replaying")
+# holds a boolean indicating if subgraphs should replay (re-run tasks, drop cached RESUME writes)
 CONFIG_KEY_TASK_ID = sys.intern("__pregel_task_id")
 # holds the task ID for the current task
 CONFIG_KEY_THREAD_ID = sys.intern("thread_id")
@@ -98,6 +100,7 @@ RESERVED = {
     CONFIG_KEY_STREAM,
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_RESUMING,
+    CONFIG_KEY_REPLAYING,
     CONFIG_KEY_TASK_ID,
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_ID,

libs/langgraph/langgraph/pregel/_loop.py

@@ -42,6 +42,7 @@ from langgraph._internal._constants import (
     CONFIG_KEY_CHECKPOINT_ID,
     CONFIG_KEY_CHECKPOINT_MAP,
     CONFIG_KEY_CHECKPOINT_NS,
+    CONFIG_KEY_REPLAYING,
     CONFIG_KEY_RESUME_MAP,
     CONFIG_KEY_RESUMING,
     CONFIG_KEY_SCRATCHPAD,
@@ -152,7 +153,7 @@ class PregelLoop:
     input_keys: str | Sequence[str]
     output_keys: str | Sequence[str]
     stream_keys: str | Sequence[str]
-    skip_done_tasks: bool
+    is_replaying: bool
     is_nested: bool
     manager: None | AsyncParentRunManager | ParentRunManager
     interrupt_after: All | Sequence[str]
@@ -244,7 +245,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.is_replaying = CONFIG_KEY_CHECKPOINT_ID in config[
+            CONF
+        ] or config[CONF].get(CONFIG_KEY_REPLAYING, False)
         self._migrate_checkpoint = migrate_checkpoint
         self.trigger_to_nodes = trigger_to_nodes
         self.retry_policy = retry_policy
@@ -451,7 +454,7 @@ class PregelLoop:
             # save the new task
             self.tasks[pushed.id] = pushed
             # match any pending writes to the new task
-            if self.skip_done_tasks:
+            if not self.is_replaying:
                 self._match_writes({pushed.id: pushed})
             # return the new task, to be started if not run before
             return pushed
@@ -515,7 +518,7 @@ class PregelLoop:
             return False
 
         # if there are pending writes from a previous loop, apply them
-        if self.skip_done_tasks and self.checkpoint_pending_writes:
+        if not self.is_replaying and self.checkpoint_pending_writes:
             self._match_writes(self.tasks)
 
         # before execution, check if we should interrupt
@@ -557,8 +560,8 @@ class PregelLoop:
             )
         # clear pending writes
         self.checkpoint_pending_writes.clear()
-        # "not skip_done_tasks" only applies to first tick after resuming
-        self.skip_done_tasks = True
+        # only replay (re-execute) done tasks on the first tick
+        self.is_replaying = False
         # save checkpoint
         self._put_checkpoint({"source": "loop"})
         # after execution, check if we should interrupt
@@ -567,8 +570,9 @@ class PregelLoop:
         ):
             self.status = "interrupt_after"
             raise GraphInterrupt()
-        # unset resuming flag
+        # unset resuming/replaying flags
         self.config[CONF].pop(CONFIG_KEY_RESUMING, None)
+        self.config[CONF].pop(CONFIG_KEY_REPLAYING, None)
 
     def match_cached_writes(self) -> Sequence[PregelExecutableTask]:
         raise NotImplementedError
@@ -729,18 +733,14 @@ class PregelLoop:
             self._put_checkpoint({"source": "input"})
         elif CONFIG_KEY_RESUMING not in configurable:
             raise EmptyInputError(f"Received no input for {input_keys}")
-        # Propagate resuming flag to subgraphs (only the outer graph does this).
+        # Propagate resuming and replaying flags to subgraphs.
         if not self.is_nested:
-            has_resume_value = (
-                isinstance(self.input, Command) and self.input.resume is not None
-            )
-            # When forking (skip_done_tasks=False, i.e. specific checkpoint_id),
-            # subgraphs should NOT resume — they start fresh.
-            # When genuinely resuming from latest, subgraphs should also resume.
-            is_fork = not self.skip_done_tasks
-            subgraph_should_resume = has_resume_value or (is_resuming and not is_fork)
             self.config = patch_configurable(
-                self.config, {CONFIG_KEY_RESUMING: subgraph_should_resume}
+                self.config,
+                {
+                    CONFIG_KEY_RESUMING: is_resuming,
+                    CONFIG_KEY_REPLAYING: self.is_replaying,
+                },
             )
         # set flag
         self.status = "pending"
@@ -1099,6 +1099,14 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             saved = self.checkpointer.get_tuple(self.checkpoint_config)
         else:
             saved = None
+        # When replaying a subgraph that wasn't in the checkpoint map
+        # (parent checkpoint predates this subgraph), start fresh.
+        if (
+            saved is not None
+            and self.config[CONF].get(CONFIG_KEY_REPLAYING)
+            and not self.checkpoint_config.get(CONF, {}).get(CONFIG_KEY_CHECKPOINT_ID)
+        ):
+            saved = None
         if saved is None:
             saved = CheckpointTuple(
                 self.checkpoint_config, empty_checkpoint(), {"step": -2}, None, []
@@ -1123,19 +1131,19 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             if saved.pending_writes is not None
             else []
         )
-        # When replaying from a specific checkpoint (fork), drop cached
-        # RESUME writes so that interrupt() calls re-fire instead of
-        # returning stale values. But if the input directly carries a
-        # resume value, keep them — multi-interrupt scenarios need
-        # previously resolved RESUME values preserved.
-        is_replaying = not self.skip_done_tasks
-        has_resume_value = (
-            self.config.get(CONF, {}).get(CONFIG_KEY_RESUMING) is True
-        ) or (isinstance(self.input, Command) and self.input.resume is not None)
-        if is_replaying and not has_resume_value:
-            self.checkpoint_pending_writes = [
-                w for w in self.checkpoint_pending_writes if w[1] != RESUME
-            ]
+        # When replaying from a specific checkpoint, drop cached RESUME
+        # writes so that interrupt() calls re-fire instead of returning
+        # stale values. But if a resume value is being provided (e.g.
+        # Command(resume=...) on a specific checkpoint), keep them —
+        # multi-interrupt scenarios need previously resolved values preserved.
+        if self.is_replaying:
+            has_resume_value = (
+                isinstance(self.input, Command) and self.input.resume is not None
+            ) or self.config.get(CONF, {}).get(CONFIG_KEY_RESUMING, False)
+            if not has_resume_value:
+                self.checkpoint_pending_writes = [
+                    w for w in self.checkpoint_pending_writes if w[1] != RESUME
+                ]
 
         self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
         self.channels, self.managed = channels_from_checkpoint(
@@ -1291,6 +1299,14 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
         else:
             saved = None
+        # When replaying a subgraph that wasn't in the checkpoint map
+        # (parent checkpoint predates this subgraph), start fresh.
+        if (
+            saved is not None
+            and self.config[CONF].get(CONFIG_KEY_REPLAYING)
+            and not self.checkpoint_config.get(CONF, {}).get(CONFIG_KEY_CHECKPOINT_ID)
+        ):
+            saved = None
         if saved is None:
             saved = CheckpointTuple(
                 self.checkpoint_config, empty_checkpoint(), {"step": -2}, None, []
@@ -1315,19 +1331,19 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             if saved.pending_writes is not None
             else []
         )
-        # When replaying from a specific checkpoint (fork), drop cached
-        # RESUME writes so that interrupt() calls re-fire instead of
-        # returning stale values. But if the input directly carries a
-        # resume value, keep them — multi-interrupt scenarios need
-        # previously resolved RESUME values preserved.
-        is_replaying = not self.skip_done_tasks
-        has_resume_value = (
-            self.config.get(CONF, {}).get(CONFIG_KEY_RESUMING) is True
-        ) or (isinstance(self.input, Command) and self.input.resume is not None)
-        if is_replaying and not has_resume_value:
-            self.checkpoint_pending_writes = [
-                w for w in self.checkpoint_pending_writes if w[1] != RESUME
-            ]
+        # When replaying from a specific checkpoint, drop cached RESUME
+        # writes so that interrupt() calls re-fire instead of returning
+        # stale values. But if a resume value is being provided (e.g.
+        # Command(resume=...) on a specific checkpoint), keep them —
+        # multi-interrupt scenarios need previously resolved values preserved.
+        if self.is_replaying:
+            has_resume_value = (
+                isinstance(self.input, Command) and self.input.resume is not None
+            ) or self.config.get(CONF, {}).get(CONFIG_KEY_RESUMING, False)
+            if not has_resume_value:
+                self.checkpoint_pending_writes = [
+                    w for w in self.checkpoint_pending_writes if w[1] != RESUME
+                ]
 
         self.submit = await self.stack.enter_async_context(
             AsyncBackgroundExecutor(self.config)

libs/langgraph/tests/test_time_travel.py

@@ -781,10 +781,15 @@ def test_subgraph_interrupt_replay_from_parent(
     history = list(graph.get_state_history(config))
     before_sub = [s for s in history if s.next == ("subgraph_node",)][-1]
 
-    # Replay — interrupt re-fires
+    # Replay from before subgraph — subgraph starts fresh, interrupt re-fires
     called.clear()
     replay_result = graph.invoke(None, before_sub.config)
     assert "__interrupt__" in replay_result
+    # Subgraph ran from scratch (step_a and ask_human called)
+    assert "step_a" in called
+    assert "ask_human" in called
+    # step_b should NOT be called (interrupt stops execution)
+    assert "step_b" not in called
 
 
 def test_subgraph_interrupt_fork_from_parent(
@@ -930,6 +935,11 @@ def test_subgraph_interrupt_replay_from_interrupt_checkpoint(
     called.clear()
     replay_result = graph.invoke(None, interrupt_checkpoint.config)
     assert "__interrupt__" in replay_result
+    # Subgraph starts fresh during replay — all nodes re-run from scratch.
+    # step_a re-runs, ask_human re-fires interrupt, step_b not reached.
+    assert "step_a" in called
+    assert "ask_human" in called
+    assert "step_b" not in called
 
 
 def test_subgraph_interrupt_fork_no_sub_checkpointer(