fix: time travel when going back to interrupt node (#7498)

# Fix: Create fork checkpoint on subgraph time travel

## Problem

When time-traveling to a subgraph checkpoint that has an interrupt, and
then resuming, the resume would load the **wrong state** — it would pick
up the original execution's latest checkpoint instead of the
time-traveled one.

This happened because replaying from a subgraph checkpoint never created
a new parent checkpoint. If the replay hit an interrupt before
`after_tick()` ran, no checkpoint was written at all, so the parent's
"latest" checkpoint was still the old one from the original execution.

## Fix

When the loop detects a time-travel replay (not an `update_state` fork),
it now **eagerly writes a fork checkpoint** at the start of the tick.
This ensures:

1. The parent thread's latest checkpoint points to the replayed state
2. Subsequent `Command(resume=...)` calls find the correct checkpoint
3. Stale `INTERRUPT` pending writes from the old checkpoint are cleared
(they reference old task IDs)

Additionally, the subgraph replay logic now uses the **parent checkpoint
ID** (from `prev_checkpoint_config`) when resolving subgraph checkpoints
during time-travel, matching the existing behavior for `update_state`
forks.

## Checkpoint flow diagrams

### Before fix: time travel leaves no fork

```
Original execution:

  C0 (start) --> C1 (step_a) --> C2 (ask_1 interrupt) --> C3 (resume) --> C4 (ask_2 interrupt) --> C5 (done)

Time travel to C2 (subgraph config):

  Replay runs... hits interrupt... no new checkpoint written.
  Parent "latest" is still C5.

  Command(resume="new_answer"):
    Loads C5 (wrong!) instead of the replayed C2 state.
```

### After fix: time travel creates a fork

```
Original execution:

  C0 --> C1 --> C2 --> C3 --> C4 --> C5 (done)

Time travel to C2 (subgraph config):

  C0 --> C1 --> C2 --> C3 --> C4 --> C5
                  \
                   F1 (fork, source="fork")  <-- new latest

  Command(resume="new_answer"):
    Loads F1 (correct!) --> resumes from the right state.

  After full resume:

  C0 --> C1 --> C2 --> C3 --> C4 --> C5
                  \
                   F1 --> F2 (ask_1 result) --> F3 (ask_2 interrupt) --> F4 (done)
```

### Manual fork via `update_state` (unchanged)

```
  C0 --> C1 --> C2 --> C3
                  \
                   U1 (source="update")  <-- created by update_state()

  This path already worked. The fix skips update/fork sources
  so existing behavior is preserved.
```

## Changes

- **`libs/langgraph/langgraph/pregel/_loop.py`**:
- Extract `is_time_traveling` flag from the existing replay detection
logic for reuse
- Write a fork checkpoint (`source="fork"`) eagerly at the start of a
time-travel tick, before execution begins
- Clear stale `INTERRUPT` pending writes when creating the fork (they
reference old task IDs that won't match the new checkpoint)
- Unify subgraph replay ID resolution: check `source in ("update",
"fork")` instead of a separate `is_time_traveling` condition, since the
new fork checkpoint now has `source="fork"`
- **`libs/langgraph/tests/test_time_travel.py`** and
**`test_time_travel_async.py`**: Added 4 new test cases (sync + async):
- `test_replay_from_before_interrupt_then_resume` — replays from a
checkpoint before an interrupt, resumes with a new answer, and verifies
the full checkpoint history (source, next, values) at each stage
- `test_subgraph_time_travel_resume_from_first_interrupt` — time-travels
to a subgraph's first interrupt, resumes both interrupts with new
answers, and verifies the fork creates a new branch while preserving the
original
- `test_subgraph_time_travel_resume_from_second_interrupt` —
time-travels to a subgraph's second interrupt, resumes with a new
answer, and verifies the first interrupt's original answer is preserved
- `test_subgraph_time_travel_checkpoint_pattern` — verifies the fork
checkpoint branches from the correct replay point and that the full
checkpoint tree is correct after resume
- **`libs/langgraph/tests/test_pregel.py`** /
**`test_pregel_async.py`**: Updated existing
`test_weather_subgraph_state` to account for the new fork checkpoint
appearing in history (history length increases by 1)

Author: sydney-runkle

libs/langgraph/langgraph/pregel/_loop.py

@@ -692,7 +692,7 @@ def _first(
         # writes so that interrupt() calls re-fire instead of returning
         # stale values. But if we're actively resuming, keep them —
         # multi-interrupt scenarios need previously resolved values preserved.
-        if self.is_replaying and (
+        is_time_traveling = self.is_replaying and (
             # Time-travel to a subgraph checkpoint: the parent sets
             # RESUMING=True (it can't distinguish time-travel from resume),
             # so we check if this subgraph's own ns is in checkpoint_map.
@@ -710,7 +710,8 @@ def _first(
                 # (subgraph input is a Send arg, not a Command)
                 or configurable.get(CONFIG_KEY_RESUMING, False)
             )
-        ):
+        )
+        if is_time_traveling:
             self.checkpoint_pending_writes = [
                 w for w in self.checkpoint_pending_writes if w[1] != RESUME
             ]
@@ -765,6 +766,26 @@ def _first(
                 if k in self.checkpoint["channel_versions"]:
                     version = self.checkpoint["channel_versions"][k]
                     self.checkpoint["versions_seen"][INTERRUPT][k] = version
+            # When time-traveling (replaying from a specific checkpoint),
+            # save a fork checkpoint so the replayed execution creates a
+            # new branch. Without this, if the execution hits an interrupt
+            # before after_tick() runs, no new checkpoint is created —
+            # the parent's latest checkpoint remains the old one and
+            # subsequent resumes load the wrong state.
+            # Skip for update_state forks (source=update/fork) since they
+            # already have their own fork checkpoint.
+            if is_time_traveling and self.checkpoint_metadata.get("source") not in (
+                "update",
+                "fork",
+            ):
+                # Clear old INTERRUPT writes from the loaded checkpoint.
+                # The fork will have a new checkpoint_id which changes
+                # task IDs — stale interrupt writes would accumulate and
+                # confuse the multiple-interrupt check in future resumes.
+                self.checkpoint_pending_writes = [
+                    w for w in self.checkpoint_pending_writes if w[1] != INTERRUPT
+                ]
+                self._put_checkpoint({"source": "fork"})
             # produce values output
             self._emit(
                 "values", map_output_values, self.output_keys, True, self.channels
@@ -807,14 +828,18 @@ def _first(
         if not self.is_nested:
             # Pass the resolved before-bound checkpoint ID so subgraphs can
             # find their corresponding checkpoint without re-fetching the
-            # parent. For forks (source=update), use the fork's parent
+            # parent. For forks (source=update/fork), use the fork's parent
             # checkpoint ID since the fork was created after the subgraph's
             # checkpoints from the original execution.
             replay_state: ReplayState | None = None
             if self.is_replaying:
                 replay_checkpoint_id = self.checkpoint["id"]
                 if (
-                    self.checkpoint_metadata.get("source") == "update"
+                    self.checkpoint_metadata.get("source")
+                    in (
+                        "update",
+                        "fork",
+                    )
                     and self.prev_checkpoint_config
                 ):
                     replay_checkpoint_id = self.prev_checkpoint_config[CONF].get(

libs/langgraph/tests/test_pregel.py

@@ -615,16 +615,19 @@ def _edge(st: MyState) -> Literal["__end__", "node_one", "node_two"]:
         )
     ]
 
-    assert len(new_history) == len(history) + 1
-    for original, new in zip(history, new_history[1:]):
+    # +2: one fork checkpoint from time travel, one from the new execution
+    assert len(new_history) == len(history) + 2
+    # new_history[0] is the new execution result, new_history[1] is the fork
+    assert new_history[1].metadata["source"] == "fork"
+    for original, new in zip(history, new_history[2:]):
         assert original.values == new.values
         assert original.next == new.next
         assert original.metadata["step"] == new.metadata["step"]
 
     def _get_tasks(hist: list, start: int):
         return [h.tasks for h in hist[start:]]
 
-    assert _get_tasks(new_history, 1) == _get_tasks(history, 0)
+    assert _get_tasks(new_history, 2) == _get_tasks(history, 0)
 
 
 def test_batch_two_processes_in_out() -> None:

libs/langgraph/tests/test_pregel_async.py

@@ -2086,16 +2086,19 @@ def _edge(st: MyState) -> Literal["__end__", "node_one", "node_two"]:
         )
     ]
 
-    assert len(new_history) == len(history) + 1
-    for original, new in zip(history, new_history[1:]):
+    # +2: one fork checkpoint from time travel, one from the new execution
+    assert len(new_history) == len(history) + 2
+    # new_history[0] is the new execution result, new_history[1] is the fork
+    assert new_history[1].metadata["source"] == "fork"
+    for original, new in zip(history, new_history[2:]):
         assert original.values == new.values
         assert original.next == new.next
         assert original.metadata["step"] == new.metadata["step"]
 
     def _get_tasks(hist: list, start: int):
         return [h.tasks for h in hist[start:]]
 
-    assert _get_tasks(new_history, 1) == _get_tasks(history, 0)
+    assert _get_tasks(new_history, 2) == _get_tasks(history, 0)
 
 
 async def test_cond_edge_after_send() -> None: