continue

libs/langgraph/langgraph/pregel/_loop.py

@@ -562,19 +562,8 @@ class PregelLoop:
         self.checkpoint_pending_writes.clear()
         # "not skip_done_tasks" only applies to first tick after resuming
         self.skip_done_tasks = True
-        # collect child checkpoint mappings from subgraph tasks
-        children: dict[str, str] = {}
-        parent_ns = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "")
-        for task in self.tasks.values():
-            task_map = task.config.get(CONF, {}).get(CONFIG_KEY_CHECKPOINT_MAP, {})
-            for ns, ckpt_id in task_map.items():
-                if ns != parent_ns:
-                    children[ns] = ckpt_id
         # save checkpoint
-        metadata: CheckpointMetadata = {"source": "loop"}
-        if children:
-            metadata["children"] = children
-        self._put_checkpoint(metadata)
+        self._put_checkpoint({"source": "loop"})
         # after execution, check if we should interrupt
         if self.interrupt_after and should_interrupt(
             self.checkpoint, self.interrupt_after, self.tasks.values()
@@ -747,24 +736,12 @@ class PregelLoop:
         # Propagate resuming and replaying flags to subgraphs.
         if not self.is_nested:
             is_replaying = not self.skip_done_tasks
-            patch: dict[str, Any] = {
-                CONFIG_KEY_RESUMING: is_resuming,
-                CONFIG_KEY_REPLAYING: is_replaying,
-            }
-            # Load child checkpoint mappings from metadata so that
-            # checkpointer=True subgraphs can load the correct checkpoint
-            # during replay (instead of always loading the latest).
-            if is_replaying:
-                children = self.checkpoint_metadata.get("children", {})
-                if children:
-                    current_map = self.config[CONF].get(CONFIG_KEY_CHECKPOINT_MAP, {})
-                    patch[CONFIG_KEY_CHECKPOINT_MAP] = {
-                        **current_map,
-                        **children,
-                    }
             self.config = patch_configurable(
                 self.config,
-                patch,
+                {
+                    CONFIG_KEY_RESUMING: is_resuming,
+                    CONFIG_KEY_REPLAYING: is_replaying,
+                },
             )
         # set flag
         self.status = "pending"
@@ -910,19 +887,6 @@ class PregelLoop:
         elif exc_type is None:
             # save final output
             self.output = read_channels(self.channels, self.output_keys)
-        # Write back checkpoint_id to CONFIG_KEY_CHECKPOINT_MAP so the parent
-        # can record which subgraph checkpoint corresponds to its own checkpoint.
-        # Only for checkpointer=True subgraphs (recast ns has no NS_END).
-        if (
-            self.is_nested
-            and self.checkpoint_ns
-            and all(NS_END not in part for part in self.checkpoint_ns)
-            and CONFIG_KEY_CHECKPOINT_MAP in self.config[CONF]
-        ):
-            recast_ns = NS_SEP.join(self.checkpoint_ns)
-            self.config[CONF][CONFIG_KEY_CHECKPOINT_MAP][recast_ns] = self.checkpoint[
-                "id"
-            ]
 
     def _emit(
         self,

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(