fix apply Command(resume=…) when paired with explicit non-head checkpoint

## Summary

Add a user-facing design doc and `get_delta_channel_keepset` helper for
third-party `BaseCheckpointSaver` authors who need to support graphs
using `DeltaChannel`.

**Deliverables:**

1. ~~**`docs/delta-channel-checkpointer-guide.md`** — comprehensive
guide covering~~:
moved to docs repo

2. **`BaseCheckpointSaver.get_delta_channel_keepset` /
`aget_delta_channel_keepset`** — returns the minimum set of ancestor
`checkpoint_id`s that must survive deletion for a given head's
`DeltaChannel` reconstruction to remain intact. Enables safe `prune`
implementations without silently corrupting delta history.

3. **Docstring warnings** on `prune`, `aprune`, `delete_for_runs`,
`adelete_for_runs`, `copy_thread`, `acopy_thread` explaining the
DeltaChannel pitfall (silent data loss if ancestor writes/snapshots are
deleted).

4. **Three new conformance capabilities** in
`libs/checkpoint-conformance`:
- `delta_channel_history` — validates the `aget_delta_channel_history`
walk contract
   - `delta_channel_keepset` — validates the keep-set contract
- `delta_channel_reconstruction` — end-to-end round-trip (aput +
aput_writes + history + reconstruct)

## Test plan

- [x] `make format lint` passes in `libs/checkpoint`,
`libs/checkpoint-conformance`
- [x] All three new conformance capabilities pass against
`InMemorySaver`
- [x] Run conformance against SQLite saver 
- [x] Run conformance against Postgres saver

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

libs/checkpoint-conformance/pyproject.toml

@@ -50,6 +50,7 @@ unresolved-attribute = "ignore"
 unresolved-import = "ignore"
 invalid-argument-type = "ignore"
 invalid-return-type = "ignore"
+missing-typed-dict-key = "ignore"
 
 [tool.ruff]
 lint.select = [

libs/langgraph/langgraph/pregel/_loop.py

@@ -776,6 +776,17 @@ class PregelLoop:
                 and configurable.get(CONFIG_KEY_CHECKPOINT_NS, "")
                 in configurable.get(CONFIG_KEY_CHECKPOINT_MAP, {})
             )
+            # Outer graph: time-travel-resume — explicit non-head
+            # checkpoint paired with a Command(resume=...). Without this,
+            # `Command(resume=...) + checkpoint=<non-head>` would match the
+            # plain-resume exclusion below and skip the cleanup needed to
+            # treat this as a fork.
+            or (
+                not self.is_nested
+                and getattr(self, "_loaded_explicit_non_head", False)
+                and input_is_command
+                and cast(Command, self.input).resume is not None
+            )
             or not (
                 # Outer graph: resume arrives as Command(resume=...)
                 (input_is_command and cast(Command, self.input).resume is not None)
@@ -1500,6 +1511,13 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
 
     def __enter__(self) -> Self:
         self._graph_lifecycle_events = deque()
+        # Set by the explicit-id branch when the loaded checkpoint is not
+        # the latest for the (thread, ns). Lets _first distinguish a
+        # plain resume (`Command(resume=...)` against head, which carries
+        # an explicit checkpoint_id only because clients like LangGraph
+        # Studio echo it back) from time-travel-resume (`Command(resume=
+        # ...)` against an explicit non-head checkpoint).
+        self._loaded_explicit_non_head = False
         if not self.checkpointer:
             saved = None
         elif self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
@@ -1507,6 +1525,14 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
             # This covers both normal replay and subgraphs resolved via
             # checkpoint_map during time-travel.
             saved = self.checkpointer.get_tuple(self.checkpoint_config)
+            if saved is not None and not self.is_nested:
+                head = self.checkpointer.get_tuple(
+                    patch_configurable(
+                        self.checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
+                    )
+                )
+                if head is not None and head.checkpoint["id"] != saved.checkpoint["id"]:
+                    self._loaded_explicit_non_head = True
         elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
             # Subgraph replay: the parent is replaying and passed us a
             # replay_state with its checkpoint_id. Look up our checkpoint
@@ -1758,6 +1784,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
 
     async def __aenter__(self) -> Self:
         self._graph_lifecycle_events = deque()
+        self._loaded_explicit_non_head = False
         if not self.checkpointer:
             saved = None
         elif self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID):
@@ -1765,6 +1792,14 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
             # This covers both normal replay and subgraphs resolved via
             # checkpoint_map during time-travel.
             saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
+            if saved is not None and not self.is_nested:
+                head = await self.checkpointer.aget_tuple(
+                    patch_configurable(
+                        self.checkpoint_config, {CONFIG_KEY_CHECKPOINT_ID: None}
+                    )
+                )
+                if head is not None and head.checkpoint["id"] != saved.checkpoint["id"]:
+                    self._loaded_explicit_non_head = True
         elif replay_state := self.config[CONF].get(CONFIG_KEY_REPLAY_STATE):
             # Subgraph replay: the parent is replaying and passed us a
             # replay_state with its checkpoint_id. Look up our checkpoint

libs/langgraph/tests/test_retry.py

@@ -1674,15 +1674,28 @@ async def test_arun_with_retry_timeout_observer_tracks_attempts():
 async def test_arun_with_retry_timeout_observer_emits_progress_on_heartbeat():
     events: list = []
 
+    # `_TimedAttemptScope.__init__` sets `_last_progress` to `time.monotonic()`,
+    # but the watchdog itself doesn't start running until after `wrap_config`
+    # and task scheduling — under CI load that gap can be large enough to eat
+    # the entire idle window before the task body's first await even runs. We
+    # defend against that by:
+    #   1. Using a generous idle_timeout so scheduling slack stays well within it.
+    #   2. Calling `runtime.heartbeat()` BEFORE the first sleep, which resets
+    #      `_last_progress` to "now" the moment the task body actually starts.
+    idle_timeout_s = 1.0
+
     class HeartbeatProc:
         async def ainvoke(self, input, config):
             runtime = config[CONF][CONFIG_KEY_RUNTIME]
+            runtime.heartbeat()  # reset the idle clock at task-body entry
             for _ in range(8):
                 await asyncio.sleep(0.05)
                 runtime.heartbeat()
             return "ok"
 
-    task = _make_task(HeartbeatProc(), timeout=_idle_timeout(0.2), name="heartbeat")
+    task = _make_task(
+        HeartbeatProc(), timeout=_idle_timeout(idle_timeout_s), name="heartbeat"
+    )
     task.config[CONF][CONFIG_KEY_TIMED_ATTEMPT_OBSERVER] = events.append
     assert await arun_with_retry(task, retry_policy=None) == "ok"
 
@@ -1691,13 +1704,13 @@ async def test_arun_with_retry_timeout_observer_emits_progress_on_heartbeat():
     assert by_event[-1] == "finish"
     progress = [ev for ev in events if ev.event == "progress"]
     assert progress, "expected at least one progress event from heartbeat"
-    # Rate limit is `idle_timeout / 4` = 0.05s; with 8 heartbeats spaced ~0.05s
-    # we should see at most ~one progress event per heartbeat (well below 8).
+    # Rate limit is `idle_timeout / 4` = 0.25s; with the task running for
+    # ~400ms we expect 1–2 progress events (well below the 9 heartbeats).
     assert len(progress) <= len(by_event)
     for ev in progress:
         assert ev.context.task_name == "heartbeat"
         assert ev.context.attempt == 1
-        assert ev.context.idle_timeout_secs == 0.2
+        assert ev.context.idle_timeout_secs == idle_timeout_s
         assert isinstance(ev.progress_at, datetime)
 
 

libs/langgraph/tests/test_time_travel.py

@@ -3964,3 +3964,96 @@ def test_subgraph_called_in_loop_loads_state_on_replay(
 
     assert len(observed) == 1
     assert observed[0] == ("sub_step", {"sub_trail": ["s", "s", "s"]})
+
+
+def test_subgraph_time_travel_resume_with_command_in_one_call(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """Repro for time-travel-resume issued as a SINGLE call:
+    `invoke(Command(resume=...), config={..., checkpoint_id: <non-head>})`.
+
+    Equivalent to the existing two-call pattern
+    (`test_subgraph_time_travel_resume_from_first_interrupt`) which does
+    `invoke(None, sub_config)` then `invoke(Command(resume=...))`. Clients
+    like LangGraph API server / Studio can submit this as one request.
+    """
+    called: list[str] = []
+
+    def step_a(state: State) -> State:
+        called.append("step_a")
+        return {"value": ["step_a_done"]}
+
+    def ask_1(state: State) -> State:
+        called.append("ask_1")
+        answer = interrupt("Question 1?")
+        return {"value": [f"ask_1:{answer}"]}
+
+    def ask_2(state: State) -> State:
+        called.append("ask_2")
+        answer = interrupt("Question 2?")
+        return {"value": [f"ask_2:{answer}"]}
+
+    executor = (
+        StateGraph(State)
+        .add_node("step_a", step_a)
+        .add_node("ask_1", ask_1)
+        .add_node("ask_2", ask_2)
+        .add_edge(START, "step_a")
+        .add_edge("step_a", "ask_1")
+        .add_edge("ask_1", "ask_2")
+        .add_edge("ask_2", "__end__")
+        .compile(checkpointer=True)
+    )
+
+    graph = (
+        StateGraph(State)
+        .add_node("executor", executor)
+        .add_edge(START, "executor")
+        .compile(checkpointer=sync_checkpointer)
+    )
+
+    config = {"configurable": {"thread_id": "1"}}
+
+    # Run to completion through both interrupts.
+    graph.invoke({"value": []}, config)
+    parent_state_at_first = graph.get_state(config, subgraphs=True)
+    sub_config_at_first = parent_state_at_first.tasks[0].state.config
+    graph.invoke(Command(resume="answer_1"), config)
+    graph.invoke(Command(resume="answer_2"), config)
+
+    final_after_original = graph.get_state(config).values
+    assert final_after_original["value"] == [
+        "step_a_done",
+        "ask_1:answer_1",
+        "ask_2:answer_2",
+    ]
+
+    # ONE-CALL time-travel-resume: pair Command(resume=...) with a
+    # non-head parent checkpoint_id in the same invocation.
+    called.clear()
+    result = graph.invoke(Command(resume="new_answer_1"), sub_config_at_first)
+
+    # `step_a` ran before the time-travel target — must not re-execute.
+    assert "step_a" not in called, f"step_a re-executed; called={called}"
+    # The new answer must be applied (not the cached `answer_1`).
+    assert result.get("__interrupt__"), (
+        f"expected to pause at the next interrupt; got result={result}"
+    )
+    assert result["__interrupt__"][0].value == "Question 2?"
+    sub_state_now = graph.get_state(config, subgraphs=True).tasks[0].state
+    assert sub_state_now.values["value"] == ["step_a_done", "ask_1:new_answer_1"], (
+        f"subgraph state at int2 should reflect new_answer_1; "
+        f"got {sub_state_now.values}"
+    )
+
+    # Resume the second interrupt with another new answer; ensure the
+    # whole branch concludes consistently.
+    called.clear()
+    final = graph.invoke(Command(resume="new_answer_2"), config)
+    assert "step_a" not in called
+    assert "ask_1" not in called
+    assert final["value"] == [
+        "step_a_done",
+        "ask_1:new_answer_1",
+        "ask_2:new_answer_2",
+    ]