fix(langgraph): don't replay an abandoned branch into a DeltaChannel fork

[Elior Nataf Lackritz (eliornl)]

Fixes #8443

Problem

Addressing an older checkpoint creates a fork: the shared base ends up with two children, and it keeps the checkpoint_writes of the branch the fork abandons. Nothing in the stored data records which child consumed which write, so the DeltaChannel ancestor walk in get_delta_channel_history collects the abandoned branch's writes too.

Live execution is correct. Only the reconstruction after a reload is wrong:

fork base:     ['in-1', 'first-out']
fork returns:  ['in-1', 'first-out', 'in-3', 'third-out']            # correct
reload gives:  ['in-1', 'first-out', 'in-2', 'in-3', 'third-out']
                                     ^^^^^^ belongs to the branch the fork replaced

full channels are unaffected: their checkpoints carry complete channel_values and need no replay. In a graph carrying both, the plain channel is right and the delta channel is not, which is what the new tests assert against.

This is not one saver's slip. The new test file fails on every backend on main: memory, memory_migrate_sends, sqlite, sqlite_aes, sqlite_aio, and postgres in all three pool modes, sync and async. InMemorySaver, the base walk used by SQLite, and PostgresSaver's paged-SQL override are each wrong independently, because write-to-child ownership simply is not recorded.

Fix

Write side, so no saver changes are needed. A run launched against an explicitly addressed checkpoint forces every DeltaChannel to snapshot into its first checkpoint, which terminates the walk inside the fork instead of at the shared base. This mirrors the existing force-snapshot for Overwrite writes, which exists for the same reason; the set that carried those channels now carries both cases, hence its rename to _delta_channels_forced_snapshot.

• _loop.py: queue every delta channel for a forced snapshot when the addressed checkpoint_id names a real checkpoint. Read off checkpoint_config rather than config, so a subgraph resolved through a checkpoint map during time travel is covered too, matching what __enter__ already does to decide the same question. Checks the value, not key presence like is_replaying, because subgraph task configs always carry an explicit None there. Drained in _put_checkpoint rather than _first, so channels that only get a value later in the run are covered.
• _checkpoint.py: create_checkpoint_plan_for_update_state_api(..., is_fork=True) forces the same full snapshot for update_state against an older checkpoint, for the same reason is_fresh_thread already does. main.py reads this off config, not checkpoint_config, because there checkpoint_config has already been merged with the loaded checkpoint's own id and would be truthy on every call.
• main.py: pass is_fork through both bulk_update_state paths. Two details there are load-bearing. The flag cannot be read off the per-superstep config, because perform_superstep returns the config of the checkpoint it just wrote and the driver feeds that back in, so from the second superstep on it always names a checkpoint whether the caller addressed one or not. And the fork has to be sealed by whichever checkpoint the fork writes first: the as_node INPUT and END paths write one and return before reaching the plan, so create_fork_checkpoint snapshots there too. Sealing later is too late, because by then the next superstep has already reconstructed a wrong in-memory value through the shared base and would just bake it into the snapshot. fork_pending tracks what is still owed, mirroring _delta_channels_awaiting_fork_snapshot in _loop.py.

create_fork_checkpoint passes get_next_version for the same reason exit mode does: those paths apply writes to the input channel rather than the delta channel, so nothing bumps the delta channel's version and put would drop the blob as not-a-new-version. Their new_versions is therefore derived from the returned checkpoint rather than the one passed in.

A channel with no value at the fork base cannot carry a snapshot blob yet (create_checkpoint skips channels absent from channel_versions), so it stays queued until the first superstep that gives it one. Without that, forking off the checkpoint that predates a thread's first input still leaks.

Cost, and why the trigger is coarse

One snapshot per addressed run, not per superstep. Measured on a six-node graph: the forked run writes 8 checkpoints and exactly 1 snapshot blob. Runs with no addressed checkpoint are untouched, and test_unaddressed_run_keeps_snapshot_cadence guards that.

The trigger fires on any truthy checkpoint_id, including a caller that addresses the thread's latest checkpoint for optimistic concurrency rather than to fork. That caller pays one snapshot per run. I considered narrowing it to "the addressed checkpoint has pending writes for this delta channel", which is where the leak actually comes from and would make that case free, and decided against it for two reasons: it would depend on CheckpointTuple.pending_writes being complete, so a saver that leaves them out goes back to leaking silently; and addressing the latest checkpoint genuinely does fork when that checkpoint carries pending writes from an interrupted run. Snapshotting when it was not needed costs a blob; not snapshotting when it was needed is silent corruption. Happy to trade that off differently if you'd rather.

Tests

tests/test_delta_channel_fork.py, parametrized over the shared sync_checkpointer / async_checkpointer fixtures and, for the run paths, over all three durabilities. Every graph carries a DeltaChannel and a plain reducer channel fed the same values, so the plain channel is the oracle: after a fork the two must agree.

• fork by invoke with new input, sync and async
• fork off the checkpoint that predates the thread's first input, sync and async (the deferred-snapshot path)
• fork by update_state / aupdate_state
• fork by bulk_update_state whose first superstep is as_node INPUT, END or __copy__, each of which skips the plan
• cadence guards that neither an unaddressed run nor an unaddressed multi-superstep bulk_update_state writes a snapshot blob

91 of 133 cases fail on main, all 133 pass with the fix. Of the 42 that pass either way, 21 are the cadence guards and the rest are the END and __copy__ variants, which turn out not to leak: END legitimately absorbs the base's already-run task writes, so the delta and plain channels agree there. They are kept because the oracle is what distinguishes that from a leak, and because both paths now go through the same sealing code.

make format, make lint_package, make lint_tests clean. Full libs/langgraph suite against Docker postgres and redis: 3601 passed, 2 failed, both tests/test_remote_graph.py connection errors that reproduce identically on unpatched main (they need the dev server, which I did not start).

Not covered

update_state persists its writes against the addressed checkpoint, so on a fork those writes also become visible to the abandoned branch's reads. Verified on this branch: after forking with update_state, re-reading the abandoned head returns ['in-1', 'first-out', 'patched', 'in-2', 'second-out'] while its plain channel correctly returns ['in-1', 'first-out', 'in-2', 'second-out']. That is the mirror of this bug rather than this bug, and it predates this change.

It is left out here because it is a separate change, not because it is hard: Aari (@AnnaSuSu) has it written and verified on top of this branch, and it is worth saying that it is blocked on this PR rather than merely adjacent to it. The fix moves those writes onto the checkpoint update_state creates, which is where the INPUT and PUSH writes in bulk_update_state already go. That only works if the new checkpoint carries a snapshot blob, because get_delta_channel_history starts its walk at target_tuple.parent_config and so never collects a checkpoint's own writes for itself. create_fork_checkpoint is what now guarantees the blob. Reproduced locally, deferring the channel_writes loop past the put and targeting next_config if is_fork else checkpoint_config:

                                fork sees "patched"   sibling clean   plain update_state
main @658541c + move             FAIL                  pass            FAIL
this branch + move, conditional  pass                  pass            pass
this branch + move, uncondit'l   pass                  pass            FAIL

On main the fork loses the update that created it. The third row is why the move has to be conditional on the fork flag this PR already threads through.

A delta channel with no value at the fork base, that the fork does not write in the same superstep, is still not sealed. create_checkpoint skips channels absent from channel_versions and get() raises on an unavailable channel, so there is no blob to write and the fork boundary goes unrecorded. Two verified cases, both on this branch:

fork the root checkpoint, fork never writes the delta channel
   delta: ['in-1']          plain: []      <- oracle
bulk_update_state fork at root, delta key only in the second superstep
   delta: ['in-1', 's2']    plain: ['s2']  <- oracle

This is narrower than the bug being fixed (on main every fork leaks) but it is the same class. Closing it needs the fork boundary represented in the first persisted checkpoint even when the channel has no value, which means either a sentinel for "empty as of here" in the checkpoint format, or synthesising a channel_versions entry for a never-written channel, which reaches into versions_seen and task triggering. That felt like a call for a maintainer rather than something to fold in here. Happy to take either direction in this PR or a follow-up.

Threads already forked before this change keep the state they have. The ownership information was never recorded, so there is nothing to recover from at read time.

Thanks to Aari (@AnnaSuSu) for the report, the minimal reproduction, and for working out that ownership is unrecoverable at read time and proposing the force-snapshot approach taken here, and to udit (@UditDewan) for the implementation in #8476, including the case where a channel has no value at the fork base and so cannot carry the blob until a later superstep. Both are credited as co-authors on the commit.

[open-swe]

Open SWE Review found 1 potential issue.

Open in Web • View Open SWE trace

[open-swe]

Open SWE Review found 1 potential issue.

Open in Web • View Open SWE trace

[open-swe]

Open SWE Review found 1 potential issue.

Open in Web • View Open SWE trace

[Aari (AnnaSuSu)]

Reviewed against 834e53d. The fix holds — I could not get the abandoned branch to leak into a fork on this branch on any backend. One finding on the direction you left in Not covered, because it changes the ordering rather than just the scope.

Your fork seal is a precondition for the mirror fix, not just adjacent to it

I built the update_state mirror on top of this branch to see what it takes. The natural fix is to move the channel writes at main.py:2005 off the addressed checkpoint and onto the one update_state creates — which is where the PUSH writes 40 lines below and the as_node == INPUT writes already go. Six write sites in bulk_update_state, and that is the only one targeting checkpoint_config.

That move only works because of this PR. get_delta_channel_history starts its walk at target_tuple.parent_config, so a checkpoint's own pending writes are never collected for itself. Relocating the writes therefore makes them invisible unless the checkpoint carries a snapshot blob — which is exactly what create_fork_checkpoint now guarantees.

Same patch, same two-sided probe, main @ 658541c vs this branch @ 834e53d:

                        sibling (must not see it)   fork (must see it)
main + write move       pass                        FAIL  ['in-1','first-out']
834e53d + write move    pass                        pass  ['in-1','first-out','patched']

On main the fork loses the update that created it. So the mirror is not independent work that happened to be deferred — it is blocked on this landing, and it stops being blocked the moment it does.

One trap in it, in case it saves you a round

The move has to be conditional on is_fork. Doing it unconditionally breaks unforked update_state: no fork means no seal, so the new checkpoint has no blob, the walk starts at its parent, and the update vanishes the same way — ['in-1','first-out'] against a plain-channel oracle reading ['in-1','first-out','patched']. writes_config = next_config if is_fork else checkpoint_config is enough, and it reuses the flag you already thread through.

Worth noting the deferred case is already covered by this PR and needs nothing extra: an unforked update_state leaves its writes on the addressed checkpoint, but forking off that checkpoint later still reads clean, because the walk skips the target's own writes when seeding.

Cost trigger

Agree with keeping the trigger coarse. Addressing the latest checkpoint for optimistic concurrency does pay one snapshot, but that checkpoint genuinely can have pending writes from an interrupted run, and a blob you did not need is cheaper than silent corruption. The same reasoning is what makes writes_config safe on that path.

---

I have the mirror fix written and verified on top of this branch — sync and async, 26 parametrized cases green across memory/sqlite/postgres, red on 834e53d without it, and no change to the stable failure set of the full suite. Happy to open it as a follow-up once this merges, or to hand you the diff if you would rather fold it in. Either is fine by me; it is your call as the author of the thing it depends on.

[Elior Nataf Lackritz (eliornl)]

Thanks for digging in, and for testing both directions. The sibling half is what would catch a fix that overshoots.

You are right about the ordering and I will fix the PR body. I had it down as adjacent work when it is actually blocked on this landing. I rebuilt your write move locally to check it rather than take it on trust, deferring the channel_writes loop until after the put and targeting next_config if is_fork else checkpoint_config:

                                    fork sees "patched"   sibling clean   plain update_state
main @658541c + move                 FAIL                  pass            FAIL
834e53d + move, conditional          pass                  pass            pass
834e53d + move, unconditional        pass                  pass            FAIL

On main the fork comes back ['in-1', 'first-out', 'in-2'] and loses the update that created it, exactly as you had it. Row three is your trap reproduced: unconditional drops plain update_state to ['in-1', 'first-out'] against a plain-channel oracle of ['in-1', 'first-out', 'patched'].

Confirmed the two pieces the reasoning rests on as well. The walk starts at target_tuple.parent_config (libs/checkpoint/langgraph/checkpoint/base/__init__.py:627), so a checkpoint's own pending writes are never collected for itself, and of the six put_writes sites in bulk_update_state the sync/async pair at 2042 and 2547 is the only one still pointing at checkpoint_config while the INPUT and PUSH writes already go to next_config.

Please open it as your own PR once this merges. You wrote it and verified it, so it should carry your name rather than a trailer in mine.

On the gate that closed #8476: you do not need a new issue to get around it. Reading require_issue_link.yml, a maintainer either reopening the PR or removing the missing-issue-link label adds bypass-issue-check and clears the gate for good, and trusted-contributor skips it up front. So just open it and ping me here, and I will sort that out and review.

Separately, the empty-channel case still listed under Not covered is a different gap and your patch does not touch it, so that one stays open either way. A delta channel with no value at the fork base cannot carry a snapshot blob, so the boundary goes unrecorded and the walk runs past it.

[PRABHU KIRAN VANDRANKI (VANDRANKI)]

Community review, does not clear the merge gate.

This is a large, intricate patch (regression for #8443) touching checkpoint invariants across _checkpoint.py, _loop.py and main.py's bulk_update_state/abulk_update_state. I read the whole diff and the new test suite in tests/test_delta_channel_fork.py.

Core mechanism as I traced it: when a run is addressed at an explicit checkpoint (a fork/time-travel), the fork's shared base still carries the abandoned branch's pending writes, and nothing records which child consumed which write. Previously only is_fresh_thread forced a full delta-channel snapshot to stop the ancestor walk; this adds is_fork for the same reason, snapshotting every delta channel (including unavailable ones, via the new include_unavailable flag) on the fork's first checkpoint so the walk terminates inside the fork instead of reaching the base.

I specifically checked the "several superstep paths write a checkpoint and return before the plan runs" problem the comments call out (as_node of INPUT/END/copy). The new create_fork_checkpoint helper and the fork_pending/_delta_channels_awaiting_fork_snapshot bookkeeping both persist the "still owed a fork snapshot" state across supersteps rather than clearing it after the first one, which is what test_fork_by_bulk_update_whose_first_superstep_skips_the_plan (parametrized over INPUT/END/copy) is checking. I also confirmed the rename _delta_channels_with_overwrite -> _delta_channels_forced_snapshot is applied consistently everywhere it's read or written.

I did not run the test suite myself, and I don't have deep enough working knowledge of DeltaChannel's ancestor-walk/versioning internals to independently re-derive every invariant here from scratch, so I'm reviewing the reasoning and consistency of the diff rather than certifying the whole mechanism. The test coverage looks genuinely thorough: sync/async, all durabilities, fork before any value exists, fork via update_state/bulk_update_state, and explicit guards that an unaddressed run does not change the normal snapshot_frequency cadence (cost containment).