this is diabolical

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
@@ -618,22 +622,42 @@ class PregelLoop:
     def _first(
         self, *, input_keys: str | Sequence[str], updated_channels: set[str] | None
     ) -> set[str] | None:
-        # resuming from previous checkpoint requires
-        # - finding a previous checkpoint
-        # - receiving None input (outer graph) or RESUMING flag (subgraph)
+        # Resuming from a previous checkpoint requires two things:
+        # 1. A prior checkpoint exists (channel_versions is non-empty)
+        # 2. The input signals continuation (not a fresh run with new input)
         configurable = self.config.get(CONF, {})
-        is_resuming = bool(self.checkpoint["channel_versions"]) and bool(
-            configurable.get(
-                CONFIG_KEY_RESUMING,
-                self.input is None
-                or isinstance(self.input, Command)
-                or (
-                    not self.is_nested
-                    and self.config.get("metadata", {}).get("run_id")
-                    == self.checkpoint_metadata.get("run_id", MISSING)
-                ),
+        has_prior_checkpoint = bool(self.checkpoint["channel_versions"])
+        # For subgraphs, the parent explicitly sets CONFIG_KEY_RESUMING.
+        # For the outer graph, we infer from the input:
+        #   - None input: resume after interrupt (invoke(None, config))
+        #   - Command input: any Command operates on existing state
+        #   - Same run_id: re-entry into an ongoing run (e.g. stream reconnect)
+        input_signals_resume = (
+            self.input is None
+            or isinstance(self.input, Command)
+            or (
+                not self.is_nested
+                and self.config.get("metadata", {}).get("run_id")
+                == self.checkpoint_metadata.get("run_id", MISSING)
             )
         )
+        is_resuming = has_prior_checkpoint and bool(
+            configurable.get(CONFIG_KEY_RESUMING, input_signals_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=...) or CONFIG_KEY_RESUMING), keep them —
+        # multi-interrupt scenarios need previously resolved values preserved.
+        if self.is_replaying:
+            is_resume_with_value = (
+                isinstance(self.input, Command) and self.input.resume is not None
+            ) or configurable.get(CONFIG_KEY_RESUMING, False)
+            if not is_resume_with_value:
+                self.checkpoint_pending_writes = [
+                    w for w in self.checkpoint_pending_writes if w[1] != RESUME
+                ]
 
         # map command to writes
         if isinstance(self.input, Command):
@@ -723,10 +747,14 @@ class PregelLoop:
             self._put_checkpoint({"source": "input"})
         elif CONFIG_KEY_RESUMING not in configurable:
             raise EmptyInputError(f"Received no input for {input_keys}")
-        # update config
+        # Propagate resuming and replaying flags to subgraphs.
         if not self.is_nested:
             self.config = patch_configurable(
-                self.config, {CONFIG_KEY_RESUMING: is_resuming}
+                self.config,
+                {
+                    CONFIG_KEY_RESUMING: is_resuming,
+                    CONFIG_KEY_REPLAYING: self.is_replaying,
+                },
             )
         # set flag
         self.status = "pending"
@@ -1078,6 +1106,59 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             },
         )
 
+    def _get_parent_checkpoint_id(self) -> str | None:
+        """Get the parent checkpoint_id to use as an upper bound for finding
+        the subgraph's checkpoint. For forks, we need the original parent
+        checkpoint (not the fork), so we look up the parent checkpoint's
+        parent_config."""
+        checkpoint_map = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})
+        parent_ns = NS_SEP.join(self.checkpoint_ns[:-1]) if self.checkpoint_ns else ""
+        parent_checkpoint_id = checkpoint_map.get(parent_ns)
+        if not parent_checkpoint_id or not self.checkpointer:
+            return None
+        # Check if this is a fork (source=update) — if so, use the fork's
+        # parent checkpoint_id instead, since the fork was created after
+        # the subgraph's checkpoints from the original execution.
+        parent_config: RunnableConfig = {
+            **self.checkpoint_config,
+            CONF: {
+                **self.checkpoint_config.get(CONF, {}),
+                CONFIG_KEY_CHECKPOINT_NS: parent_ns,
+                CONFIG_KEY_CHECKPOINT_ID: parent_checkpoint_id,
+            },
+        }
+        parent_saved = self.checkpointer.get_tuple(parent_config)
+        if parent_saved and parent_saved.metadata.get("source") == "update":
+            if parent_saved.parent_config:
+                return parent_saved.parent_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
+        return parent_checkpoint_id
+
+    def _get_checkpoint_before_parent(self) -> CheckpointTuple | None:
+        """Find the subgraph checkpoint that was current at the parent's
+        checkpoint time, using the parent checkpoint_id as an upper bound.
+
+        Returns a CheckpointTuple with the historical channel_values but fresh
+        execution state (empty channel_versions/versions_seen) so the subgraph
+        re-runs its nodes while retaining accumulated data.
+
+        Returns None to start fresh if no such checkpoint exists."""
+        parent_checkpoint_id = self._get_parent_checkpoint_id()
+        if parent_checkpoint_id and self.checkpointer:
+            before_config: RunnableConfig = {
+                CONF: {"checkpoint_id": parent_checkpoint_id}
+            }
+            for saved in self.checkpointer.list(
+                self.checkpoint_config, before=before_config, limit=1
+            ):
+                checkpoint = empty_checkpoint()
+                checkpoint["channel_values"] = saved.checkpoint.get(
+                    "channel_values", {}
+                )
+                return CheckpointTuple(
+                    self.checkpoint_config, checkpoint, {"step": -2}, None, []
+                )
+        return None
+
     # context manager
 
     def __enter__(self) -> Self:
@@ -1085,6 +1166,16 @@ 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.
+        # For stateful subgraphs (checkpointer=True), find the checkpoint
+        # that was current at the parent's checkpoint time.
+        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 = self._get_checkpoint_before_parent()
         if saved is None:
             saved = CheckpointTuple(
                 self.checkpoint_config, empty_checkpoint(), {"step": -2}, None, []
@@ -1109,7 +1200,6 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             if saved.pending_writes is not None
             else []
         )
-
         self.submit = self.stack.enter_context(BackgroundExecutor(self.config))
         self.channels, self.managed = channels_from_checkpoint(
             self.specs, self.checkpoint
@@ -1257,6 +1347,48 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             },
         )
 
+    async def _aget_parent_checkpoint_id(self) -> str | None:
+        checkpoint_map = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})
+        parent_ns = NS_SEP.join(self.checkpoint_ns[:-1]) if self.checkpoint_ns else ""
+        parent_checkpoint_id = checkpoint_map.get(parent_ns)
+        if not parent_checkpoint_id or not self.checkpointer:
+            return None
+        parent_config: RunnableConfig = {
+            **self.checkpoint_config,
+            CONF: {
+                **self.checkpoint_config.get(CONF, {}),
+                CONFIG_KEY_CHECKPOINT_NS: parent_ns,
+                CONFIG_KEY_CHECKPOINT_ID: parent_checkpoint_id,
+            },
+        }
+        parent_saved = await self.checkpointer.aget_tuple(parent_config)
+        if parent_saved and parent_saved.metadata.get("source") == "update":
+            if parent_saved.parent_config:
+                return parent_saved.parent_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
+        return parent_checkpoint_id
+
+    async def _aget_checkpoint_before_parent(self) -> CheckpointTuple | None:
+        parent_checkpoint_id = await self._aget_parent_checkpoint_id()
+        if parent_checkpoint_id and self.checkpointer:
+            before_config: RunnableConfig = {
+                CONF: {"checkpoint_id": parent_checkpoint_id}
+            }
+            async for saved in self.checkpointer.alist(
+                self.checkpoint_config, before=before_config, limit=1
+            ):
+                checkpoint = empty_checkpoint()
+                checkpoint["channel_values"] = saved.checkpoint.get(
+                    "channel_values", {}
+                )
+                return CheckpointTuple(
+                    self.checkpoint_config,
+                    checkpoint,
+                    {"step": -2},
+                    None,
+                    [],
+                )
+        return None
+
     # context manager
 
     async def __aenter__(self) -> Self:
@@ -1264,6 +1396,16 @@ 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.
+        # For stateful subgraphs (checkpointer=True), find the checkpoint
+        # that was current at the parent's checkpoint time.
+        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 = await self._aget_checkpoint_before_parent()
         if saved is None:
             saved = CheckpointTuple(
                 self.checkpoint_config, empty_checkpoint(), {"step": -2}, None, []
@@ -1288,7 +1430,6 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             if saved.pending_writes is not None
             else []
         )
-
         self.submit = await self.stack.enter_async_context(
             AsyncBackgroundExecutor(self.config)
         )