From 799c2f04fd3a56186e24e9fb8bbf3568e543cfbe Mon Sep 17 00:00:00 2001 From: Elior Nataf Lackritz Date: Sun, 11 Oct 2026 08:51:09 -0400 Subject: [PATCH] fix(langgraph): keep an exit run from a checkpoint the thread moved past out of its newer branch (#9269) An exit-durability run that starts from a checkpoint the thread has moved past, addressed by `checkpoint_id`, stores its DeltaChannel writes on that checkpoint. The branch that already grew from it reads the writes stored there, so its checkpoints picked up this run's writes too: ```python graph.invoke({"log": ["in1"]}, config) # nodes a and b add "a1", "b1" turn_1 = graph.get_state(config).config graph.invoke({"log": ["in2"]}, config) turn_2 = graph.get_state(config).config graph.invoke({"log": ["in3"]}, turn_1, durability="exit") graph.get_state(turn_2).values["log"] # ["in1", "a1", "b1", "in3", "a3", "b3", "in2", "a2", "b2"], was ["in1", "a1", "b1", "in2", "a2", "b2"] ``` Sync and async durability store each step's writes on the run's own checkpoints, so they weren't affected. The run now snapshots the DeltaChannels it wrote instead, as a run's input already does on a checkpoint a `checkpoint_id` addressed (#9260). That costs one snapshot of each channel the run wrote, even when the addressed checkpoint is the latest one. Threads that already hit this keep the wrong values on the existing branch. Found by @hntrl in review of langchain-ai/langgraphjs#2974 ([repro](https://gist.github.com/hntrl/2185e489abe22147b3f6f77010c18264)). JS twin: langchain-ai/langgraphjs#2989 ## Tests `test_delta_channel_exit_mode.py`: one and two exit runs from the end of turn 1, after turn 2, leave every checkpoint's DeltaChannel equal to a plain reducer channel. Both fail on main on every saver in the fixture. The `langgraph` suite passes (without Redis), and lint is clean. --- libs/langgraph/langgraph/pregel/_loop.py | 8 +++++ .../tests/test_delta_channel_exit_mode.py | 30 +++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/libs/langgraph/langgraph/pregel/_loop.py b/libs/langgraph/langgraph/pregel/_loop.py index 8e17b8741..e352fc9c0 100644 --- a/libs/langgraph/langgraph/pregel/_loop.py +++ b/libs/langgraph/langgraph/pregel/_loop.py @@ -1360,6 +1360,14 @@ class PregelLoop: ): return + # A checkpoint a `checkpoint_id` addressed may already have children, + # which read the writes stored on it, so then the channels this run + # wrote snapshot instead. + if self._has_persisted_parent and not self._loaded_latest: + self._delta_channels_forced_snapshot.update( + w[3] for w in self._exit_delta_writes + ) + counters = dict( self.checkpoint_metadata.get("counters_since_delta_snapshot") or {} ) diff --git a/libs/langgraph/tests/test_delta_channel_exit_mode.py b/libs/langgraph/tests/test_delta_channel_exit_mode.py index 27aa2e255..70e2627f6 100644 --- a/libs/langgraph/tests/test_delta_channel_exit_mode.py +++ b/libs/langgraph/tests/test_delta_channel_exit_mode.py @@ -625,3 +625,33 @@ def test_exit_resume_retried_after_its_final_checkpoint_fails_reruns_the_resumed state = graph.get_state(config) assert state.values["log"] == state.values["plain"] assert sorted(state.values["plain"]) == ["ask", "done", "in"] + + +class _TurnState(_ResumeState): + turn: int + + +@pytest.mark.parametrize("exit_runs", [1, 2]) +def test_exit_runs_from_a_checkpoint_the_thread_moved_past_leave_its_newer_branch_alone( + sync_checkpointer: BaseCheckpointSaver, exit_runs: int +) -> None: + builder = StateGraph(_TurnState) + builder.add_node("a", lambda state: _both(f"a{state['turn']}")) + builder.add_node("b", lambda state: _both(f"b{state['turn']}")) + builder.add_edge(START, "a") + builder.add_edge("a", "b") + graph = builder.compile(checkpointer=sync_checkpointer) + config = {"configurable": {"thread_id": "t"}} + graph.invoke({"turn": 1, **_both("in1")}, config) + end_of_turn_1 = graph.get_state(config).config + graph.invoke({"turn": 2, **_both("in2")}, config) + + for turn in range(3, 3 + exit_runs): + graph.invoke( + {"turn": turn, **_both(f"in{turn}")}, end_of_turn_1, durability="exit" + ) + + history = list(graph.get_state_history(config)) + assert [s.values.get("log", []) for s in history] == [ + s.values.get("plain", []) for s in history + ]