Merge branch 'main' into fix/delta-fork-abandoned-branch

A DeltaChannel that was never written on the branch being forked has no
value to snapshot and no entry in channel_versions, so create_checkpoint
skipped it and the fork's first checkpoint recorded no boundary at all.
The walk then ran past the fork into the shared base and collected the
abandoned branch's writes, the same failure this branch already fixes for
channels that do have a value.

Two shapes leaked. A run forking off a checkpoint older than the channel's
first value and never writing that channel returned ['in-1'] where the
plain-channel oracle returned []. A bulk update writing the delta key only
in its second superstep returned ['in-1', 's2'] against ['s2'].

No new blob type is needed. _DeltaSnapshot already carries the value and
is already serialized by every saver, and from_checkpoint turns MISSING
into typ(), so _DeltaSnapshot(typ()) reconstructs to the same empty value
the channel would have had. What was missing is a version: without one,
put drops the blob as not-a-new-version, so mint a first one.

Deferring the seal to a later superstep does not work. That superstep
reconstructs through the still-unsealed checkpoint and would only bake the
corrupted value into its own snapshot.

Checked that minting a version does not fire nodes that subscribe to the
channel: a raw Pregel node subscribed directly to the delta channel stays
silent across the fork.

Reported by the Open SWE review bot on #8548.

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_history.py

@@ -208,93 +208,6 @@ async def test_history_migration_plain_value_as_seed(
     assert values == [2], f"Expected [2], got {values}"
 
 
-# Task ids used by the ordering tests below. `build_delta_chain` tags its own
-# writes with a `uuid4`, whose hex digits are all <= "f", so "aaaa..." sorts
-# before every fixture task id and "zzzz..." sorts after every one of them.
-# That makes the expected order fully determined rather than dependent on which
-# uuid4 the fixture happened to draw.
-TASK_ID_SORTS_FIRST = "aaaaaaaa-0000-0000-0000-000000000000"
-TASK_ID_SORTS_LAST = "zzzzzzzz-0000-0000-0000-000000000000"
-
-
-async def test_history_orders_parallel_writes_by_task_path(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Writes from several tasks in one super-step replay in task_path order.
-
-    Live execution sorts a super-step's tasks by `task_path_str(path[:3])`
-    before applying their values, so replay has to recover that order rather
-    than `task_id` order — `task_id` is a hash of the path, so the two
-    disagree, and reducers are only required to be batching-invariant, not
-    order-invariant.
-
-    The two task_ids are assigned so they sort in the *opposite* order from
-    their task_paths. A saver ordering by `(task_id, idx)` therefore returns
-    these writes reversed, rather than passing by happening to agree.
-    """
-    configs = await build_delta_chain(
-        saver,
-        thread_id=str(uuid4()),
-        channel="ch",
-        snapshots_at_steps=[0],
-        total_steps=3,
-    )
-    # The chain is: step 0 snapshot (seed), step 1 write, step 2 write.
-    # `aget_delta_channel_history` walks from the head's parent back to the
-    # seed, so it collects step 1's writes only — step 0 terminates the walk
-    # and step 2 is the head, whose own writes are pending for the next
-    # super-step and excluded. So step 1 is where these writes have to go.
-    step_1, head = configs[1], configs[2]
-    await saver.aput_writes(
-        step_1, [("ch", "second")], TASK_ID_SORTS_FIRST, "~pull, 02"
-    )
-    await saver.aput_writes(step_1, [("ch", "first")], TASK_ID_SORTS_LAST, "~pull, 01")
-
-    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
-    values = [w[2] for w in result["ch"]["writes"]]
-    # 1 is the fixture's own write at step 1. It carries no task_path, so it
-    # sorts ahead of both writes added above.
-    assert values == [1, "first", "second"], (
-        f"Expected task_path order [1, 'first', 'second'], got {values}. "
-        "Ordering by (task_id, idx) alone yields [1, 'second', 'first']."
-    )
-
-
-async def test_history_orders_pathless_writes_first(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """Writes stored without a task_path sort ahead of path-carrying ones.
-
-    A task-less write (graph input) persists `task_path=""`, as does any row
-    written before a saver recorded the column. `""` precedes every
-    `task_path_str` output because that function prefixes tuples with `~`, so
-    those writes replay first — where live execution applies graph input.
-    """
-    configs = await build_delta_chain(
-        saver,
-        thread_id=str(uuid4()),
-        channel="ch",
-        snapshots_at_steps=[0],
-        total_steps=3,
-    )
-    # Same chain shape as above: step 1 is the only step the walk collects.
-    step_1, head = configs[1], configs[2]
-    # Committed in the opposite order to the one they must replay in, so the
-    # assertion cannot pass on insertion order alone.
-    await saver.aput_writes(
-        step_1, [("ch", "from_node")], TASK_ID_SORTS_FIRST, "~pull, a"
-    )
-    await saver.aput_writes(step_1, [("ch", "from_input")], TASK_ID_SORTS_LAST)
-
-    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
-    values = [w[2] for w in result["ch"]["writes"]]
-    # Both 1 (the fixture's write) and "from_input" are pathless, so they sort
-    # by task_id among themselves and both precede the path-carrying write.
-    assert values == [1, "from_input", "from_node"], (
-        f"Expected pathless writes first, got {values}"
-    )
-
-
 ALL_DELTA_CHANNEL_HISTORY_TESTS = [
     test_history_returns_writes_oldest_first,
     test_history_seed_is_nearest_snapshot,
@@ -303,8 +216,6 @@ ALL_DELTA_CHANNEL_HISTORY_TESTS = [
     test_history_empty_channels_returns_empty,
     test_history_walk_to_root_no_seed,
     test_history_migration_plain_value_as_seed,
-    test_history_orders_parallel_writes_by_task_path,
-    test_history_orders_pathless_writes_first,
 ]
 
 

libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py

@@ -168,7 +168,6 @@ class _DeltaStage2Row(TypedDict, total=False):
     type: str | None
     blob: bytes | None
     task_id: str | None  # "w" rows only
-    task_path: str | None  # "w" rows only
     idx: int | None  # "w" rows only
     version: str | None  # "b" rows only
 
@@ -320,7 +319,7 @@ def _build_delta_stage2_sql(
         branches.append(
             "SELECT 'w'::text AS _kind, "
             "checkpoint_id, channel, "
-            "type, blob, task_id, task_path, idx, NULL::text AS version "
+            "type, blob, task_id, idx, NULL::text AS version "
             "FROM checkpoint_writes "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND checkpoint_id = ANY(%s)"
@@ -328,8 +327,7 @@ def _build_delta_stage2_sql(
     for _ in channels_with_seed:
         branches.append(
             "SELECT 'b'::text AS _kind, NULL::text AS checkpoint_id, channel, "
-            "type, blob, NULL::text AS task_id, NULL::text AS task_path, "
-            "NULL::int AS idx, version "
+            "type, blob, NULL::text AS task_id, NULL::int AS idx, version "
             "FROM checkpoint_blobs "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND version = %s"
@@ -494,11 +492,10 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         stored value, or when the seed blob is sentinel "empty" — in both cases
         the consumer treats absence as "start empty".
         """
-        # writes_by_ch_by_cid[channel][cid] = list of
-        # (type, blob, task_id, idx, task_path)
-        writes_by_ch_by_cid: dict[
-            str, dict[str, list[tuple[str, bytes, str, int, str]]]
-        ] = {ch: {} for ch in channels}
+        # writes_by_ch_by_cid[channel][cid] = list of (type, blob, task_id, idx)
+        writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
+            ch: {} for ch in channels
+        }
         # seed_blob_by_ver[(channel, version)] = (type, blob)
         seed_blob_by_ver: dict[tuple[str, str], tuple[str, bytes]] = {}
 
@@ -509,17 +506,8 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
                 cid = cast(str, r["checkpoint_id"])
                 writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
                     cast(
-                        "tuple[str, bytes, str, int, str]",
-                        (
-                            r["type"],
-                            r["blob"],
-                            r["task_id"],
-                            r["idx"],
-                            # `task_path` is NOT NULL DEFAULT '' on "w" rows;
-                            # it is nullable on `_DeltaStage2Row` only because
-                            # the seed branch selects NULL for it.
-                            r["task_path"],
-                        ),
+                        "tuple[str, bytes, str, int]",
+                        (r["type"], r["blob"], r["task_id"], r["idx"]),
                     )
                 )
             else:  # kind == "b"
@@ -528,12 +516,10 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
                     "tuple[str, bytes]", (r["type"], r["blob"])
                 )
 
-        # Sort writes per (channel, cid) newest-first by
-        # (task_path, task_id, idx) — the order `apply_writes` applied them
-        # in live, and the order documented on `DeltaChannelHistory`.
+        # Sort writes per (channel, cid) newest-first by (task_id, idx)
         for cid_map in writes_by_ch_by_cid.values():
             for ws in cid_map.values():
-                ws.sort(key=lambda w: (w[4], w[2], w[3]), reverse=True)
+                ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
 
         result: dict[str, DeltaChannelHistory] = {}
         for ch in channels:
@@ -543,9 +529,7 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
             collected: list[PendingWrite] = []
             cid_writes = writes_by_ch_by_cid.get(ch, {})
             for cid in chain_cids:
-                for type_tag, write_blob, task_id, _idx, _path in cid_writes.get(
-                    cid, []
-                ):
+                for type_tag, write_blob, task_id, _idx in cid_writes.get(cid, []):
                     val = self.serde.loads_typed((type_tag, write_blob))
                     collected.append((task_id, ch, val))
             collected.reverse()

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

@@ -29,11 +29,6 @@ from langgraph.checkpoint.sqlite._delta import (
     build_delta_stage2_sql,
     step_walk_with_row,
 )
-from langgraph.checkpoint.sqlite._schema import (
-    ADD_WRITES_TASK_PATH_SQL,
-    DUPLICATE_COLUMN_ERROR,
-    HAS_WRITES_TASK_PATH_SQL,
-)
 from langgraph.checkpoint.sqlite.utils import search_where
 
 _AIO_ERROR_MSG = (
@@ -159,7 +154,6 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                 checkpoint_ns TEXT NOT NULL DEFAULT '',
                 checkpoint_id TEXT NOT NULL,
                 task_id TEXT NOT NULL,
-                task_path TEXT NOT NULL DEFAULT '',
                 idx INTEGER NOT NULL,
                 channel TEXT NOT NULL,
                 type TEXT,
@@ -168,12 +162,6 @@ class SqliteSaver(BaseCheckpointSaver[str]):
             );
             """
         )
-        if not self.conn.execute(HAS_WRITES_TASK_PATH_SQL).fetchone():
-            try:
-                self.conn.execute(ADD_WRITES_TASK_PATH_SQL)
-            except sqlite3.OperationalError as exc:
-                if DUPLICATE_COLUMN_ERROR not in str(exc):
-                    raise
 
         self.is_setup = True
 
@@ -472,9 +460,9 @@ class SqliteSaver(BaseCheckpointSaver[str]):
             task_path: Path of the task creating the writes.
         """
         query = (
-            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
             if all(w[0] in WRITES_IDX_MAP for w in writes)
-            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
         )
         with self.cursor() as cur:
             cur.executemany(
@@ -485,7 +473,6 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                         str(config["configurable"]["checkpoint_ns"]),
                         str(config["configurable"]["checkpoint_id"]),
                         task_id,
-                        task_path,
                         WRITES_IDX_MAP.get(channel, idx),
                         channel,
                         *self.serde.dumps_typed(value),
@@ -581,7 +568,7 @@ class SqliteSaver(BaseCheckpointSaver[str]):
                     )
                 cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
-                    "list[tuple[str, str, str, int, str, bytes, str]]", cur.fetchall()
+                    "list[tuple[str, str, str, int, str, bytes]]", cur.fetchall()
                 )
             else:
                 stage2_rows = []

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/_delta.py

@@ -57,7 +57,7 @@ def build_delta_stage2_sql(*, chain_lens: Sequence[int]) -> str:
     for n in chain_lens:
         cid_placeholders = ",".join("?" * n)
         branches.append(
-            "SELECT checkpoint_id, channel, task_id, idx, type, value, task_path "
+            "SELECT checkpoint_id, channel, task_id, idx, type, value "
             "FROM writes "
             "WHERE thread_id = ? AND checkpoint_ns = ? AND channel = ? "
             f"AND checkpoint_id IN ({cid_placeholders})"
@@ -130,33 +130,29 @@ def build_delta_channels_writes_history(
     chain_by_ch: Mapping[str, list[str]],
     seed_val_by_ch: Mapping[str, Any],
     seeded: set[str],
-    stage2_rows: Sequence[tuple[str, str, str, int, str, bytes, str]],
+    stage2_rows: Sequence[tuple[str, str, str, int, str, bytes]],
     serde: Any,
 ) -> dict[str, DeltaChannelHistory]:
     """Demux stage-2 rows per channel; produce per-channel histories.
 
-    Stage-2 rows are
-    `(checkpoint_id, channel, task_id, idx, type, value, task_path)`.
-    Final write order is oldest→newest globally and
-    `(task_path, task_id, idx)` within a checkpoint, matching the contract
-    on `DeltaChannelHistory.writes` — that is the order `apply_writes`
-    applied them in live, which `(task_id, idx)` alone does not recover
-    for parallel tasks writing one channel in a single super-step.
+    Stage-2 rows are `(checkpoint_id, channel, task_id, idx, type, value)`.
+    Final write order is oldest→newest globally and `(task_id, idx)` within
+    a checkpoint, matching the contract on `DeltaChannelHistory.writes`.
 
     `seed` is omitted when the walk reached a true root with no snapshot
     found (channel never entered `seeded`); consumers treat absence as
     "start empty".
     """
-    writes_by_ch_by_cid: dict[
-        str, dict[str, list[tuple[str, bytes, str, int, str]]]
-    ] = {ch: {} for ch in channels}
-    for cid, ch, task_id, idx, type_tag, value_blob, task_path in stage2_rows:
+    writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
+        ch: {} for ch in channels
+    }
+    for cid, ch, task_id, idx, type_tag, value_blob in stage2_rows:
         writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
-            (type_tag, value_blob, task_id, idx, task_path)
+            (type_tag, value_blob, task_id, idx)
         )
     for cid_map in writes_by_ch_by_cid.values():
         for ws in cid_map.values():
-            ws.sort(key=lambda w: (w[4], w[2], w[3]))
+            ws.sort(key=lambda w: (w[2], w[3]))
 
     result: dict[str, DeltaChannelHistory] = {}
     for ch in channels:
@@ -165,7 +161,7 @@ def build_delta_channels_writes_history(
         collected: list[PendingWrite] = []
         # Chain is newest-first; iterate oldest-first for the public order.
         for cid in reversed(chain_cids):
-            for type_tag, value_blob, task_id, _idx, _path in cid_writes.get(cid, []):
+            for type_tag, value_blob, task_id, _idx in cid_writes.get(cid, []):
                 collected.append(
                     (task_id, ch, serde.loads_typed((type_tag, value_blob)))
                 )

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/_schema.py

@@ -1,37 +0,0 @@
-"""Additive schema migrations shared by the sqlite savers.
-
-`SqliteSaver.setup` and `AsyncSqliteSaver.setup` create their tables with
-`CREATE TABLE IF NOT EXISTS`, which leaves a database created by an earlier
-version on the earlier schema. Sqlite has no `ADD COLUMN IF NOT EXISTS`
-(the postgres savers rely on that form), and re-running a plain
-`ALTER TABLE ... ADD COLUMN` raises `OperationalError: duplicate column
-name`. So each migration pairs an `ALTER` with a probe against
-`pragma_table_info` that tells us whether this database still needs it.
-
-Databases created fresh already carry every column from the `CREATE TABLE`
-statements, so the probe finds the column and the `ALTER` never runs.
-"""
-
-from __future__ import annotations
-
-# `writes.task_path` records the path of the task that produced a write.
-# Delta channel replay orders a checkpoint's writes by
-# (task_path, task_id, idx) to reproduce the order `apply_writes` applied
-# them in live; without the column, replay can only order by
-# (task_id, idx), which permutes writes made by parallel tasks in the same
-# super-step. Rows written before this migration keep the `''` default and
-# so sort ahead of path-carrying rows within their checkpoint.
-HAS_WRITES_TASK_PATH_SQL = (
-    "SELECT 1 FROM pragma_table_info('writes') WHERE name = 'task_path'"
-)
-
-ADD_WRITES_TASK_PATH_SQL = (
-    "ALTER TABLE writes ADD COLUMN task_path TEXT NOT NULL DEFAULT ''"
-)
-
-# Substring of the `OperationalError` sqlite raises when the column is already
-# there. The probe above is not enough on its own: two connections opening the
-# same file can both pass it and both issue the `ALTER`, and unlike
-# `CREATE TABLE IF NOT EXISTS` the loser of that race raises. Callers treat it
-# as success — whoever won did the same migration.
-DUPLICATE_COLUMN_ERROR = "duplicate column name"

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/aio.py

@@ -30,11 +30,6 @@ from langgraph.checkpoint.sqlite._delta import (
     build_delta_stage2_sql,
     step_walk_with_row,
 )
-from langgraph.checkpoint.sqlite._schema import (
-    ADD_WRITES_TASK_PATH_SQL,
-    DUPLICATE_COLUMN_ERROR,
-    HAS_WRITES_TASK_PATH_SQL,
-)
 from langgraph.checkpoint.sqlite.utils import search_where
 
 T = TypeVar("T", bound=Callable)
@@ -336,7 +331,6 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
                     checkpoint_ns TEXT NOT NULL DEFAULT '',
                     checkpoint_id TEXT NOT NULL,
                     task_id TEXT NOT NULL,
-                    task_path TEXT NOT NULL DEFAULT '',
                     idx INTEGER NOT NULL,
                     channel TEXT NOT NULL,
                     type TEXT,
@@ -347,16 +341,6 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
             ):
                 await self.conn.commit()
 
-            async with self.conn.execute(HAS_WRITES_TASK_PATH_SQL) as cur:
-                has_task_path = await cur.fetchone() is not None
-            if not has_task_path:
-                try:
-                    await self.conn.execute(ADD_WRITES_TASK_PATH_SQL)
-                except aiosqlite.OperationalError as exc:
-                    if DUPLICATE_COLUMN_ERROR not in str(exc):
-                        raise
-                await self.conn.commit()
-
             self.is_setup = True
 
     async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
@@ -592,9 +576,9 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
             task_path: Path of the task creating the writes.
         """
         query = (
-            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
             if all(w[0] in WRITES_IDX_MAP for w in writes)
-            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
         )
         await self.setup()
         async with self.lock, self.conn.cursor() as cur:
@@ -606,7 +590,6 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
                         str(config["configurable"]["checkpoint_ns"]),
                         str(config["configurable"]["checkpoint_id"]),
                         task_id,
-                        task_path,
                         WRITES_IDX_MAP.get(channel, idx),
                         channel,
                         *self.serde.dumps_typed(value),
@@ -698,7 +681,7 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
                     )
                 await cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
-                    "list[tuple[str, str, str, int, str, bytes, str]]",
+                    "list[tuple[str, str, str, int, str, bytes]]",
                     await cur.fetchall(),
                 )
             else:

libs/checkpoint-sqlite/tests/test_writes_task_path_migration.py

@@ -1,214 +0,0 @@
-"""Tests for the additive `writes.task_path` migration (#8382).
-
-`task_path` records the path of the task that produced a write, so delta
-channel replay can restore the order `apply_writes` applied a super-step's
-writes in. The sqlite savers previously accepted `task_path` on `put_writes`
-and dropped it, so the column has to be added to databases created by earlier
-versions as well as to fresh ones.
-
-Sqlite has no `ADD COLUMN IF NOT EXISTS`, so `setup()` probes
-`pragma_table_info` before issuing the `ALTER` — these tests pin that the
-probe makes the migration both effective and repeatable.
-"""
-
-from __future__ import annotations
-
-import sqlite3
-from pathlib import Path
-
-import aiosqlite
-import pytest
-from langgraph.checkpoint.base import empty_checkpoint
-
-from langgraph.checkpoint.sqlite import SqliteSaver
-from langgraph.checkpoint.sqlite._schema import ADD_WRITES_TASK_PATH_SQL
-from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
-
-# The `writes` table as created before `task_path` existed.
-LEGACY_SCHEMA = """
-CREATE TABLE checkpoints (
-    thread_id TEXT NOT NULL,
-    checkpoint_ns TEXT NOT NULL DEFAULT '',
-    checkpoint_id TEXT NOT NULL,
-    parent_checkpoint_id TEXT,
-    type TEXT,
-    checkpoint BLOB,
-    metadata BLOB,
-    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id)
-);
-CREATE TABLE writes (
-    thread_id TEXT NOT NULL,
-    checkpoint_ns TEXT NOT NULL DEFAULT '',
-    checkpoint_id TEXT NOT NULL,
-    task_id TEXT NOT NULL,
-    idx INTEGER NOT NULL,
-    channel TEXT NOT NULL,
-    type TEXT,
-    value BLOB,
-    PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
-);
-"""
-
-
-def _write_legacy_db(path: Path) -> None:
-    conn = sqlite3.connect(path)
-    try:
-        conn.executescript(LEGACY_SCHEMA)
-        conn.commit()
-    finally:
-        conn.close()
-
-
-def _columns(conn: sqlite3.Connection, table: str) -> list[str]:
-    return [row[1] for row in conn.execute(f"PRAGMA table_info({table})")]
-
-
-def test_fresh_database_has_task_path(tmp_path: Path) -> None:
-    with SqliteSaver.from_conn_string(str(tmp_path / "fresh.sqlite")) as saver:
-        saver.setup()
-        assert "task_path" in _columns(saver.conn, "writes")
-
-
-def test_legacy_database_gains_task_path(tmp_path: Path) -> None:
-    db = tmp_path / "legacy.sqlite"
-    _write_legacy_db(db)
-
-    with SqliteSaver.from_conn_string(str(db)) as saver:
-        saver.setup()
-        columns = _columns(saver.conn, "writes")
-
-    assert "task_path" in columns
-    # Existing columns are untouched — this is additive, not a table rebuild.
-    assert columns[:8] == [
-        "thread_id",
-        "checkpoint_ns",
-        "checkpoint_id",
-        "task_id",
-        "idx",
-        "channel",
-        "type",
-        "value",
-    ]
-
-
-def test_setup_is_repeatable_on_migrated_database(tmp_path: Path) -> None:
-    """A second `setup()` must not re-issue the `ALTER`.
-
-    Sqlite raises `duplicate column name` rather than ignoring it, so an
-    unguarded `ALTER` would break every reopen of a migrated database.
-    """
-    db = tmp_path / "legacy.sqlite"
-    _write_legacy_db(db)
-
-    with SqliteSaver.from_conn_string(str(db)) as saver:
-        saver.setup()
-        saver.is_setup = False
-        saver.setup()
-        assert "task_path" in _columns(saver.conn, "writes")
-
-    # And again through a fresh connection to the migrated file.
-    with SqliteSaver.from_conn_string(str(db)) as saver:
-        saver.setup()
-        assert "task_path" in _columns(saver.conn, "writes")
-
-
-def test_legacy_rows_keep_default_and_sort_first(tmp_path: Path) -> None:
-    """Rows predating the column read back as `''` and order ahead of paths.
-
-    `''` precedes every `task_path_str` output, which puts pre-migration
-    writes before path-carrying ones within their checkpoint instead of
-    interleaving them under a rule that never applied to them.
-    """
-    db = tmp_path / "legacy.sqlite"
-    _write_legacy_db(db)
-
-    conn = sqlite3.connect(db)
-    try:
-        conn.execute(
-            "INSERT INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id,"
-            " idx, channel, type, value) VALUES ('t', '', 'c', 'task', 0, 'ch',"
-            " 'null', X'')"
-        )
-        conn.commit()
-    finally:
-        conn.close()
-
-    with SqliteSaver.from_conn_string(str(db)) as saver:
-        saver.setup()
-        stored = saver.conn.execute("SELECT task_path FROM writes").fetchall()
-        assert stored == [("",)]
-
-        ordered = saver.conn.execute(
-            "SELECT task_path FROM writes ORDER BY task_path, task_id, idx"
-        ).fetchall()
-        assert ordered[0] == ("",)
-
-
-def test_setup_survives_losing_the_migration_race(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    """`setup()` succeeds when another connection migrates first.
-
-    The `pragma_table_info` probe is not a lock. Two connections opening the
-    same file can both see the column missing, and whichever issues the `ALTER`
-    second gets `duplicate column name` — `ALTER TABLE ADD COLUMN` has no
-    `IF NOT EXISTS` form to fall back on, unlike the `CREATE TABLE`s above it.
-
-    Stubbing the probe to always report the column missing reproduces exactly
-    the losing interleaving (probe says absent, another connection adds it,
-    then we `ALTER`) without depending on thread timing.
-    """
-    db = tmp_path / "legacy.sqlite"
-    _write_legacy_db(db)
-
-    # Winner of the race: migrates the file out from under the saver below.
-    winner = sqlite3.connect(db)
-    try:
-        winner.execute(ADD_WRITES_TASK_PATH_SQL)
-        winner.commit()
-    finally:
-        winner.close()
-
-    monkeypatch.setattr(
-        "langgraph.checkpoint.sqlite.HAS_WRITES_TASK_PATH_SQL", "SELECT 1 WHERE 0"
-    )
-    with SqliteSaver.from_conn_string(str(db)) as loser:
-        loser.setup()
-        assert "task_path" in _columns(loser.conn, "writes")
-
-
-def test_put_writes_persists_task_path(tmp_path: Path) -> None:
-    with SqliteSaver.from_conn_string(str(tmp_path / "fresh.sqlite")) as saver:
-        config = saver.put(
-            {"configurable": {"thread_id": "t", "checkpoint_ns": ""}},
-            empty_checkpoint(),
-            {},
-            {},
-        )
-        saver.put_writes(config, [("ch", "v")], "task-1", "~__pregel_pull, node")
-
-        stored = saver.conn.execute("SELECT task_id, task_path FROM writes").fetchall()
-        assert stored == [("task-1", "~__pregel_pull, node")]
-
-
-@pytest.mark.asyncio
-async def test_async_saver_migrates_and_persists_task_path(tmp_path: Path) -> None:
-    db = tmp_path / "legacy.sqlite"
-    _write_legacy_db(db)
-
-    async with AsyncSqliteSaver.from_conn_string(str(db)) as saver:
-        await saver.setup()
-        config = await saver.aput(
-            {"configurable": {"thread_id": "t", "checkpoint_ns": ""}},
-            empty_checkpoint(),
-            {},
-            {},
-        )
-        await saver.aput_writes(config, [("ch", "v")], "task-1", "~__pregel_pull, node")
-        # Idempotent for the async saver too.
-        saver.is_setup = False
-        await saver.setup()
-
-    async with aiosqlite.connect(db) as conn:
-        async with conn.execute("SELECT task_id, task_path FROM writes") as cur:
-            assert await cur.fetchall() == [("task-1", "~__pregel_pull, node")]

libs/checkpoint/langgraph/checkpoint/base/__init__.py

@@ -161,23 +161,6 @@ class DeltaChannelHistory(TypedDict):
       Always present; possibly empty. Already filtered to one channel.
       Writes stored at the target checkpoint itself are pending for the
       next super-step and are excluded.
-
-      Within a single checkpoint, writes are ordered by
-      `(task_path, task_id, idx)`. This mirrors the order live execution
-      applied them in: `apply_writes` sorts a super-step's tasks by
-      `task_path_str(task.path[:3])` before handing their values to
-      `channel.update`, so a path-ordered replay reproduces the value
-      `invoke` returned. Ordering by `(task_id, idx)` alone does not —
-      `task_id` is a hash of the path, so for two or more tasks writing
-      the same channel in one super-step it permutes the values against
-      the order the reducer originally saw them in. Reducers are only
-      required to be batching-invariant, not order-invariant, so that
-      permutation changes the reconstructed value.
-
-      Writes persisted without a `task_path` (a task-less write such as
-      graph input, or a row written before the saver recorded the column)
-      sort first within their checkpoint, since `""` precedes every
-      `task_path_str` output — `task_path_str` prefixes tuples with `~`.
     * `seed` — the stored value at the nearest ancestor whose
       `channel_values[ch]` is populated. Omitted if the walk reached the
       root without finding any stored value (consumer treats absence as
@@ -627,14 +610,6 @@ class BaseCheckpointSaver(Generic[V]):
         `PostgresSaver`) override for performance; the return contract is
         fixed here.
 
-        `PendingWrite` carries no `task_path`, so the default takes each
-        ancestor's write order straight from `get_tuple`. Savers relying
-        on it must therefore return `pending_writes` ordered by
-        `(task_path, task_id, idx)` to satisfy the intra-checkpoint order
-        documented on `DeltaChannelHistory`; savers that order
-        `pending_writes` by `(task_id, idx)` alone need to override this
-        method (as the in-tree savers do) rather than inherit it.
-
         Args:
             config: Configuration identifying the target checkpoint.
             channels: Channel names to walk for. Empty → empty mapping.

libs/checkpoint/langgraph/checkpoint/memory/__init__.py

@@ -30,23 +30,6 @@ from langgraph.checkpoint.base import (
 logger = logging.getLogger(__name__)
 
 
-# How `InMemorySaver.writes[thread, ns, checkpoint]` keys and stores one write.
-_WriteKey = tuple[str, int]  # task ID, write idx
-_WriteValue = tuple[str, str, tuple[str, bytes], str]  # + channel, value, path
-_WriteEntry = tuple[_WriteKey, _WriteValue]  # one `dict.items()` pair
-
-
-def _delta_replay_sort_key(entry: _WriteEntry) -> tuple[str, str, int]:
-    """Order one checkpoint's writes as `apply_writes` applied them live.
-
-    Live order is `(task_path, task_id, idx)` — see `DeltaChannelHistory`. It
-    has to be assembled from both halves of the entry: `(task_id, idx)` is the
-    key, `task_path` is the last element of the value.
-    """
-    (task_id, idx), (_, _, _, task_path) = entry
-    return (task_path, task_id, idx)
-
-
 class InMemorySaver(
     BaseCheckpointSaver[str], AbstractContextManager, AbstractAsyncContextManager
 ):
@@ -88,7 +71,10 @@ class InMemorySaver(
         dict[str, dict[str, tuple[tuple[str, bytes], tuple[str, bytes], str | None]]],
     ]
     # (thread ID, checkpoint NS, checkpoint ID) -> (task ID, write idx)
-    writes: defaultdict[tuple[str, str, str], dict[_WriteKey, _WriteValue]]
+    writes: defaultdict[
+        tuple[str, str, str],
+        dict[tuple[str, int], tuple[str, str, tuple[str, bytes], str]],
+    ]
     blobs: dict[
         tuple[
             str, str, str, str | int | float
@@ -214,10 +200,8 @@ class InMemorySaver(
                     terminated_here.add(ch)
 
             step_writes = self.writes.get((thread_id, checkpoint_ns, cp_id), {})
-            # Newest-first; the caller reverses to get the public oldest-first
-            # order.
-            for (task_id, _idx), (_, ch, serialized, _task_path) in sorted(
-                step_writes.items(), key=_delta_replay_sort_key, reverse=True
+            for (_task_id, _idx), (tid, ch, serialized, _) in sorted(
+                step_writes.items(), reverse=True
             ):
                 if ch not in remaining:
                     continue
@@ -227,7 +211,7 @@ class InMemorySaver(
                 ):
                     continue
                 collected_by_ch[ch].append(
-                    (task_id, ch, self.serde.loads_typed(serialized))
+                    (tid, ch, self.serde.loads_typed(serialized))
                 )
 
             for ch in terminated_here:

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -80,12 +80,20 @@ def get_updated_channels_from_tasks(
 
 def get_delta_channels_from_all_channels(
     channels: Mapping[str, BaseChannel],
+    *,
+    include_unavailable: bool = False,
 ) -> set[str]:
-    """DeltaChannels to snapshot on the first update_state of a fresh thread."""
+    """Every available DeltaChannel.
+
+    The set to snapshot whenever no ancestor walk can reconstruct these
+    channels: the first update_state of a fresh thread (no ancestors at all),
+    and the first checkpoint of a fork (whose base also holds the writes of the
+    branch the fork abandons).
+    """
     return {
         k
         for k, ch in channels.items()
-        if isinstance(ch, DeltaChannel) and ch.is_available()
+        if isinstance(ch, DeltaChannel) and (include_unavailable or ch.is_available())
     }
 
 
@@ -122,15 +130,27 @@ def create_checkpoint_plan_for_update_state_api(
     parents: dict[str, Any],
     saved_metadata: Mapping[str, Any] | None,
     is_fresh_thread: bool,
+    is_fork: bool,
 ) -> tuple[set[str], dict[str, Any]]:
-    """Return ``(channels_to_snapshot, metadata)`` for an update_state head."""
+    """Return ``(channels_to_snapshot, metadata)`` for an update_state head.
+
+    ``is_fork`` (the update was addressed at an explicit checkpoint) forces a
+    full snapshot for the same reason ``is_fresh_thread`` does: the ancestor
+    walk cannot reconstruct this head. The base a fork branches off keeps the
+    pending writes of the branch being abandoned, and nothing records which
+    child consumed which write, so the walk would replay them here too.
+    Snapshotting terminates the walk at this checkpoint. Every delta channel
+    snapshots, so no counters carry over.
+    """
     metadata: dict[str, Any] = {
         "source": "update",
         "step": step,
         "parents": parents,
     }
-    if is_fresh_thread:
-        return get_delta_channels_from_all_channels(channels), metadata
+    if is_fresh_thread or is_fork:
+        return get_delta_channels_from_all_channels(
+            channels, include_unavailable=is_fork
+        ), metadata
 
     new_counters = create_metadata_for_update_state_api(
         channels,
@@ -146,6 +166,43 @@ def create_checkpoint_plan_for_update_state_api(
     return channels_to_snapshot, metadata
 
 
+def create_fork_checkpoint(
+    checkpoint: Checkpoint,
+    channels: Mapping[str, BaseChannel],
+    step: int,
+    *,
+    is_fork: bool,
+    get_next_version: GetNextVersion,
+) -> Checkpoint:
+    """``create_checkpoint`` for an update_state path that bypasses the plan.
+
+    The ``as_node`` INPUT and END paths write the fork's first checkpoint and
+    return before ``create_checkpoint_plan_for_update_state_api`` runs. Left
+    without a snapshot that checkpoint does not seal the fork, and the next
+    superstep reconstructs its delta channels by walking through the shared
+    base, picking up the abandoned branch's writes and then baking them into
+    whatever it snapshots. Sealing has to happen on the fork's *first*
+    checkpoint, which is this one.
+
+    ``get_next_version`` is required for the same reason exit mode needs it:
+    these paths apply writes to the input channel, not to the delta channel,
+    so nothing bumps the delta channel's version and ``put`` would drop the
+    blob as not-a-new-version. Callers must derive ``new_versions`` from the
+    returned checkpoint rather than the one they passed in.
+    """
+    if not is_fork:
+        return create_checkpoint(checkpoint, channels, step)
+    return create_checkpoint(
+        checkpoint,
+        channels,
+        step,
+        get_next_version=get_next_version,
+        channels_to_snapshot=get_delta_channels_from_all_channels(
+            channels, include_unavailable=True
+        ),
+    )
+
+
 def create_checkpoint(
     checkpoint: Checkpoint,
     channels: Mapping[str, BaseChannel] | None,
@@ -174,14 +231,27 @@ def create_checkpoint(
         values = {}
         channel_versions = dict(checkpoint["channel_versions"])
         for k in channels:
-            if k not in channel_versions:
-                continue
             ch = channels[k]
+            if k not in channel_versions:
+                # Nothing was ever written to this channel on this branch, so
+                # it has no version and `put` would drop any blob stored for
+                # it. A *forced* snapshot still has to land: it is the only
+                # thing that stops the ancestor walk running past this
+                # checkpoint into a fork base that holds another branch's
+                # writes. Mint a first version so the blob survives.
+                if k in channels_to_snapshot and get_next_version is not None:
+                    channel_versions[k] = get_next_version(None, None)
+                    values[k] = _DeltaSnapshot(
+                        ch.get() if ch.is_available() else ch.typ()
+                    )
+                continue
             if k in channels_to_snapshot:
                 # Callers force a full snapshot blob here: exit mode when a
-                # delta channel reaches its snapshot cadence, and update_state
-                # on a fresh thread (no ancestor to replay writes from). The
-                # manual version-bump below only applies to the exit-mode case.
+                # delta channel reaches its snapshot cadence, update_state on
+                # a fresh thread (no ancestor to replay writes from), and a
+                # fork (whose base also holds the abandoned branch's writes).
+                # The manual version-bump below only applies to the exit-mode
+                # case.
                 #
                 # In exit mode, the snapshot decision is deferred to exit
                 # time (intermediate steps have do_checkpoint=False). The

libs/langgraph/langgraph/pregel/_loop.py

@@ -222,10 +222,32 @@ class PregelLoop:
     # under the saver's `ORDER BY task_id, idx` sorting.
     _exit_delta_writes: list[tuple[int, str, str, Any]] | None = None
 
-    # Delta channels that saw an Overwrite since the last checkpoint. These
-    # channels must snapshot after live update applies overwrite semantics so
-    # sparse replay starts from the same post-overwrite value.
-    _delta_channels_with_overwrite: set[str]
+    # Delta channels that must snapshot at the next checkpoint, whatever their
+    # cadence counters say. Two sources:
+    # * an Overwrite arrived since the last checkpoint, so the snapshot has to
+    #   happen after live update applied overwrite semantics and sparse replay
+    #   starts from the same post-overwrite value;
+    # * this run forked off an explicitly addressed checkpoint, see
+    #   `_delta_channels_awaiting_fork_snapshot`.
+    _delta_channels_forced_snapshot: set[str]
+
+    # Delta channels still owed a fork snapshot, when this run was launched
+    # against an explicitly addressed checkpoint (time travel / fork). That
+    # base keeps the pending writes of the branch the fork abandons, and
+    # nothing records which child consumed which write, so the ancestor walk
+    # would replay them into this branch too. Snapshotting terminates the walk
+    # inside the fork instead of at the shared base. Names drop out once the
+    # blob has landed; a channel with no value yet has nothing to snapshot, so
+    # it waits for the superstep that gives it one.
+    #
+    # The trigger is deliberately coarse: any addressed checkpoint, not only
+    # one that turns out to have abandoned writes on it. Which writes belong to
+    # which child is exactly what is not recorded, so a narrower test would
+    # have to trust the base's `pending_writes` to be complete, and a saver
+    # that leaves them out would silently go back to leaking. Snapshotting when
+    # it was not needed costs one blob per addressed run; not snapshotting when
+    # it was needed is silent corruption.
+    _delta_channels_awaiting_fork_snapshot: set[str]
 
     # The checkpoint_config that points at the parent loaded at `__enter__`
     # (or the synthetic-empty checkpoint, on first run). We capture it
@@ -369,6 +391,16 @@ class PregelLoop:
             if self.config[CONF].get(CONFIG_KEY_CHECKPOINT_NS)
             else ()
         )
+        # Checks the value, not just key presence like `is_replaying` above:
+        # subgraph task configs always carry an explicit `None` here, and only
+        # a real id means the caller addressed one specific checkpoint. Read
+        # off `checkpoint_config` so subgraphs resolved through a checkpoint
+        # map during time travel are covered too, matching `__enter__`.
+        self._delta_channels_awaiting_fork_snapshot = (
+            {k for k, spec in specs.items() if isinstance(spec, DeltaChannel)}
+            if self.checkpoint_config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
+            else set()
+        )
         self.prev_checkpoint_config = None
         runtime = self.config[CONF].get(CONFIG_KEY_RUNTIME)
         self.control = runtime.control if isinstance(runtime, Runtime) else None
@@ -683,7 +715,7 @@ class PregelLoop:
     def after_tick(self) -> None:
         # finish superstep
         writes = [w for t in self.tasks.values() for w in t.writes]
-        self._delta_channels_with_overwrite.update(
+        self._delta_channels_forced_snapshot.update(
             ch
             for ch, v in writes
             if isinstance(self.specs.get(ch), DeltaChannel) and _get_overwrite(v)[0]
@@ -991,7 +1023,7 @@ class PregelLoop:
                 manager=None,
                 updated_channels=updated_channels,
             )
-            self._delta_channels_with_overwrite.update(
+            self._delta_channels_forced_snapshot.update(
                 c
                 for c, v in input_writes
                 if isinstance(self.specs.get(c), DeltaChannel) and _get_overwrite(v)[0]
@@ -1133,10 +1165,21 @@ class PregelLoop:
         do_checkpoint = self._checkpointer_put_after_previous is not None and (
             exiting or self.durability != "exit"
         )
+        # Fork: make this checkpoint self-contained, so the ancestor walk stops
+        # inside the fork instead of reaching the base this run forked off and
+        # collecting the abandoned branch's writes from it. Resolved here
+        # rather than in `_first` so channels that only got a value this
+        # superstep are covered too.
+        if self._delta_channels_awaiting_fork_snapshot:
+            self._delta_channels_forced_snapshot.update(
+                k
+                for k in self._delta_channels_awaiting_fork_snapshot
+                if k in self.channels
+            )
         # create new checkpoint
         channels_to_snapshot = (
             delta_channels_to_snapshot(self.channels, new_counters)
-            | self._delta_channels_with_overwrite
+            | self._delta_channels_forced_snapshot
             if do_checkpoint
             else set()
         )
@@ -1154,7 +1197,13 @@ class PregelLoop:
         for k in channels_to_snapshot:
             new_counters[k] = (0, 0)
         if do_checkpoint:
-            self._delta_channels_with_overwrite.difference_update(channels_to_snapshot)
+            self._delta_channels_forced_snapshot.difference_update(channels_to_snapshot)
+            # `create_checkpoint` drops a requested snapshot for a channel with
+            # no version in this checkpoint yet (nothing was ever written to it
+            # on this branch), so keep asking until the blob really landed.
+            self._delta_channels_awaiting_fork_snapshot.difference_update(
+                self.checkpoint["channel_values"]
+            )
         non_zero = {k: v for k, v in new_counters.items() if v != (0, 0)}
         if non_zero:
             self.checkpoint_metadata["counters_since_delta_snapshot"] = non_zero
@@ -1239,7 +1288,7 @@ class PregelLoop:
         )
         channels_to_snapshot = (
             delta_channels_to_snapshot(self.channels, counters)
-            | self._delta_channels_with_overwrite
+            | self._delta_channels_forced_snapshot
         )
 
         pending = [
@@ -1684,7 +1733,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         )
         self._delta_write_futs = []
         self._error_handler_write_futs = []
-        self._delta_channels_with_overwrite = set()
+        self._delta_channels_forced_snapshot = set()
         self._exit_delta_writes = (
             [] if self.durability == "exit" and self.checkpointer is not None else None
         )
@@ -1942,7 +1991,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         )
         self._delta_write_futs = []
         self._error_handler_write_futs = []
-        self._delta_channels_with_overwrite = set()
+        self._delta_channels_forced_snapshot = set()
         self._exit_delta_writes = (
             [] if self.durability == "exit" and self.checkpointer is not None else None
         )

libs/langgraph/langgraph/pregel/main.py

@@ -108,6 +108,7 @@ from langgraph.callbacks import (
     get_sync_graph_callback_manager_for_config,
 )
 from langgraph.channels.base import BaseChannel
+from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.topic import Topic
 from langgraph.config import get_config
 from langgraph.constants import END
@@ -133,6 +134,7 @@ from langgraph.pregel._checkpoint import (
     copy_checkpoint,
     create_checkpoint,
     create_checkpoint_plan_for_update_state_api,
+    create_fork_checkpoint,
     empty_checkpoint,
     get_updated_channels_from_tasks,
 )
@@ -1637,8 +1639,24 @@ class Pregel(
             else:
                 raise ValueError(f"Subgraph {recast} not found")
 
+        # Delta channels still owed a fork snapshot. Mirrors
+        # `_delta_channels_awaiting_fork_snapshot` in `_loop.py`: a fork is only
+        # sealed once a checkpoint actually carries the blob. Several superstep
+        # paths (`as_node` of INPUT, END or `__copy__`) write a checkpoint and
+        # return before reaching the plan, so a flag cleared after the first
+        # superstep would leave the branch unsealed and the next superstep would
+        # reconstruct through the shared base.
+        fork_pending: set[str] = (
+            {k for k, v in self.channels.items() if isinstance(v, DeltaChannel)}
+            if config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
+            else set()
+        )
+
         def perform_superstep(
-            input_config: RunnableConfig, updates: Sequence[StateUpdate]
+            input_config: RunnableConfig,
+            updates: Sequence[StateUpdate],
+            *,
+            is_fork: bool,
         ) -> RunnableConfig:
             # get last checkpoint
             config = ensure_config(self.config, input_config)
@@ -1726,9 +1744,17 @@ class Pregel(
                         self.trigger_to_nodes,
                     )
                 # save checkpoint
+                next_checkpoint = create_fork_checkpoint(
+                    checkpoint,
+                    channels,
+                    step,
+                    is_fork=is_fork,
+                    get_next_version=checkpointer.get_next_version,
+                )
+                fork_pending.difference_update(next_checkpoint["channel_values"])
                 next_config = checkpointer.put(
                     checkpoint_config,
-                    create_checkpoint(checkpoint, channels, step),
+                    next_checkpoint,
                     {
                         "source": "update",
                         "step": step + 1,
@@ -1736,7 +1762,7 @@ class Pregel(
                     },
                     get_new_channel_versions(
                         checkpoint_previous_versions,
-                        checkpoint["channel_versions"],
+                        next_checkpoint["channel_versions"],
                     ),
                 )
                 return patch_checkpoint_map(
@@ -1765,9 +1791,17 @@ class Pregel(
                         if saved and saved.metadata.get("step") is not None
                         else -1
                     )
+                    next_checkpoint = create_fork_checkpoint(
+                        checkpoint,
+                        channels,
+                        next_step,
+                        is_fork=is_fork,
+                        get_next_version=checkpointer.get_next_version,
+                    )
+                    fork_pending.difference_update(next_checkpoint["channel_values"])
                     next_config = checkpointer.put(
                         checkpoint_config,
-                        create_checkpoint(checkpoint, channels, next_step),
+                        next_checkpoint,
                         {
                             "source": "input",
                             "step": next_step,
@@ -1777,7 +1811,7 @@ class Pregel(
                         },
                         get_new_channel_versions(
                             checkpoint_previous_versions,
-                            checkpoint["channel_versions"],
+                            next_checkpoint["channel_versions"],
                         ),
                     )
 
@@ -1873,6 +1907,9 @@ class Pregel(
                     return perform_superstep(
                         patch_checkpoint_map(next_config, saved.metadata),
                         [item for lst in user_group_by.values() for item in lst],
+                        # The checkpoint just written clears tasks and carries
+                        # no delta snapshot, so a fork is still unsealed here.
+                        is_fork=is_fork,
                     )
 
                 return patch_checkpoint_map(next_config, saved.metadata)
@@ -2020,6 +2057,7 @@ class Pregel(
                     parents=saved.metadata.get("parents", {}) if saved else {},
                     saved_metadata=saved.metadata if saved else None,
                     is_fresh_thread=saved is None,
+                    is_fork=is_fork,
                 )
             )
             checkpoint = create_checkpoint(
@@ -2032,6 +2070,8 @@ class Pregel(
                 else None,
                 channels_to_snapshot=channels_to_snapshot,
             )
+            if is_fork:
+                fork_pending.difference_update(checkpoint["channel_values"])
             next_config = checkpointer.put(
                 checkpoint_config,
                 checkpoint,
@@ -2049,8 +2089,14 @@ class Pregel(
         current_config = patch_configurable(
             config, {CONFIG_KEY_THREAD_ID: str(config[CONF][CONFIG_KEY_THREAD_ID])}
         )
+        # The flag cannot be derived from `current_config`: `perform_superstep`
+        # returns the config of the checkpoint it just wrote and the loop feeds
+        # that back in, so from the second superstep on it always names a
+        # checkpoint whether or not the caller addressed one.
         for superstep in supersteps:
-            current_config = perform_superstep(current_config, superstep)
+            current_config = perform_superstep(
+                current_config, superstep, is_fork=bool(fork_pending)
+            )
         return current_config
 
     async def abulk_update_state(
@@ -2103,8 +2149,24 @@ class Pregel(
             else:
                 raise ValueError(f"Subgraph {recast} not found")
 
+        # Delta channels still owed a fork snapshot. Mirrors
+        # `_delta_channels_awaiting_fork_snapshot` in `_loop.py`: a fork is only
+        # sealed once a checkpoint actually carries the blob. Several superstep
+        # paths (`as_node` of INPUT, END or `__copy__`) write a checkpoint and
+        # return before reaching the plan, so a flag cleared after the first
+        # superstep would leave the branch unsealed and the next superstep would
+        # reconstruct through the shared base.
+        fork_pending: set[str] = (
+            {k for k, v in self.channels.items() if isinstance(v, DeltaChannel)}
+            if config[CONF].get(CONFIG_KEY_CHECKPOINT_ID)
+            else set()
+        )
+
         async def aperform_superstep(
-            input_config: RunnableConfig, updates: Sequence[StateUpdate]
+            input_config: RunnableConfig,
+            updates: Sequence[StateUpdate],
+            *,
+            is_fork: bool,
         ) -> RunnableConfig:
             # get last checkpoint
             config = ensure_config(self.config, input_config)
@@ -2190,16 +2252,25 @@ class Pregel(
                         self.trigger_to_nodes,
                     )
                 # save checkpoint
+                next_checkpoint = create_fork_checkpoint(
+                    checkpoint,
+                    channels,
+                    step,
+                    is_fork=is_fork,
+                    get_next_version=checkpointer.get_next_version,
+                )
+                fork_pending.difference_update(next_checkpoint["channel_values"])
                 next_config = await checkpointer.aput(
                     checkpoint_config,
-                    create_checkpoint(checkpoint, channels, step),
+                    next_checkpoint,
                     {
                         "source": "update",
                         "step": step + 1,
                         "parents": saved.metadata.get("parents", {}) if saved else {},
                     },
                     get_new_channel_versions(
-                        checkpoint_previous_versions, checkpoint["channel_versions"]
+                        checkpoint_previous_versions,
+                        next_checkpoint["channel_versions"],
                     ),
                 )
                 return patch_checkpoint_map(
@@ -2228,9 +2299,17 @@ class Pregel(
                         if saved and saved.metadata.get("step") is not None
                         else -1
                     )
+                    next_checkpoint = create_fork_checkpoint(
+                        checkpoint,
+                        channels,
+                        next_step,
+                        is_fork=is_fork,
+                        get_next_version=checkpointer.get_next_version,
+                    )
+                    fork_pending.difference_update(next_checkpoint["channel_values"])
                     next_config = await checkpointer.aput(
                         checkpoint_config,
-                        create_checkpoint(checkpoint, channels, next_step),
+                        next_checkpoint,
                         {
                             "source": "input",
                             "step": next_step,
@@ -2240,7 +2319,7 @@ class Pregel(
                         },
                         get_new_channel_versions(
                             checkpoint_previous_versions,
-                            checkpoint["channel_versions"],
+                            next_checkpoint["channel_versions"],
                         ),
                     )
 
@@ -2335,6 +2414,9 @@ class Pregel(
                     return await aperform_superstep(
                         patch_checkpoint_map(next_config, saved.metadata),
                         [item for lst in user_group_by.values() for item in lst],
+                        # The checkpoint just written clears tasks and carries
+                        # no delta snapshot, so a fork is still unsealed here.
+                        is_fork=is_fork,
                     )
 
                 return patch_checkpoint_map(
@@ -2480,6 +2562,7 @@ class Pregel(
                     parents=saved.metadata.get("parents", {}) if saved else {},
                     saved_metadata=saved.metadata if saved else None,
                     is_fresh_thread=saved is None,
+                    is_fork=is_fork,
                 )
             )
             checkpoint = create_checkpoint(
@@ -2492,6 +2575,8 @@ class Pregel(
                 else None,
                 channels_to_snapshot=channels_to_snapshot,
             )
+            if is_fork:
+                fork_pending.difference_update(checkpoint["channel_values"])
             next_config = await checkpointer.aput(
                 checkpoint_config,
                 checkpoint,
@@ -2508,8 +2593,14 @@ class Pregel(
         current_config = patch_configurable(
             config, {CONFIG_KEY_THREAD_ID: str(config[CONF][CONFIG_KEY_THREAD_ID])}
         )
+        # The flag cannot be derived from `current_config`: `aperform_superstep`
+        # returns the config of the checkpoint it just wrote and the loop feeds
+        # that back in, so from the second superstep on it always names a
+        # checkpoint whether or not the caller addressed one.
         for superstep in supersteps:
-            current_config = await aperform_superstep(current_config, superstep)
+            current_config = await aperform_superstep(
+                current_config, superstep, is_fork=bool(fork_pending)
+            )
         return current_config
 
     def update_state(

libs/langgraph/tests/memory_assert.py

@@ -85,11 +85,19 @@ class MemorySaverAssertImmutable(InMemorySaver):
                 )
                 == saved
             ), config["configurable"]["checkpoint_ns"]
-        self.storage_for_copies[thread_id][checkpoint_ns][checkpoint["id"]] = (
-            self.serde.dumps_typed(checkpoint)
-        )
         # call super to write checkpoint
-        return super().put(config, checkpoint, metadata, new_versions)
+        next_config = super().put(config, checkpoint, metadata, new_versions)
+        # Record the checkpoint as the saver reads it back, not the object it
+        # was handed. `channel_values` are stored per (channel, version), so a
+        # channel a step did not write is refilled from the blob its inherited
+        # version still points at. A `DeltaChannel` omits its value except at a
+        # snapshot, which makes the two representations differ for reasons that
+        # are not mutation. Comparing read-back against read-back still catches
+        # a checkpoint whose stored data actually changed.
+        self.storage_for_copies[thread_id][checkpoint_ns][checkpoint["id"]] = (
+            self.serde.dumps_typed(super().get(next_config))
+        )
+        return next_config
 
 
 class MemorySaverNoPending(InMemorySaver):

libs/langgraph/tests/test_delta_channel_fork.py

@@ -0,0 +1,414 @@
+"""Forking a thread must not replay the abandoned branch into the fork.
+
+Regression suite for #8443. 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 used to collect the
+abandoned branch's writes as well.
+
+Every graph here carries a ``DeltaChannel`` and a plain reducer channel fed the
+same values. ``full`` channels store complete ``channel_values`` and need no
+replay, so the plain channel is the oracle: after a fork the two must agree.
+
+Coverage: fork by ``invoke`` with new input (sync/async, all durabilities),
+fork off the checkpoint that predates the thread's first input (sync/async),
+fork by ``update_state`` / ``aupdate_state``, fork before the delta channel
+ever had a value, and guards that neither an
+unaddressed run nor an unaddressed multi-superstep ``bulk_update_state``
+departs from the normal ``snapshot_frequency`` cadence.
+"""
+
+from collections.abc import Sequence
+from operator import add
+from typing import Annotated, Any
+
+import pytest
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.base import BaseCheckpointSaver
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
+from typing_extensions import TypedDict
+
+from langgraph._internal._constants import INPUT
+from langgraph.channels.delta import DeltaChannel
+from langgraph.graph import END, StateGraph
+from langgraph.types import Durability, StateSnapshot, StateUpdate
+
+pytestmark = pytest.mark.anyio
+
+
+def _append(current: list | None, writes: Sequence[Any]) -> list:
+    """DeltaChannel reducer: extend the list with every batched write."""
+    out = list(current or [])
+    for write in writes:
+        out.extend(write if isinstance(write, list) else [write])
+    return out
+
+
+class _State(TypedDict):
+    log: Annotated[list, DeltaChannel(_append, snapshot_frequency=1000)]
+    plain: Annotated[list, add]
+    # Written only by the tests that fork before `log` ever has a value, so the
+    # fork advances without touching the delta channel.
+    other: Annotated[list, add]
+
+
+def _build(checkpointer: BaseCheckpointSaver, tag: str) -> Any:
+    """Compile a one-node graph whose node appends ``{tag}-out`` to both channels.
+
+    ``snapshot_frequency=1000`` keeps the cadence from masking the bug: without
+    a forced snapshot the fork's ancestor walk always runs past the fork base.
+    """
+
+    def node(state: _State) -> dict:
+        return {"log": [f"{tag}-out"], "plain": [f"{tag}-out"]}
+
+    builder = StateGraph(_State)
+    builder.add_node("n", node)
+    builder.set_entry_point("n")
+    builder.set_finish_point("n")
+    return builder.compile(checkpointer=checkpointer)
+
+
+def _build_without_delta_writes(checkpointer: BaseCheckpointSaver, tag: str) -> Any:
+    """Compile a graph whose node writes only ``other``, never the delta channel."""
+
+    def node(state: _State) -> dict:
+        return {"other": [f"{tag}-other"]}
+
+    builder = StateGraph(_State)
+    builder.add_node("n", node)
+    builder.set_entry_point("n")
+    builder.set_finish_point("n")
+    return builder.compile(checkpointer=checkpointer)
+
+
+def _thread(thread_id: str) -> RunnableConfig:
+    return {"configurable": {"thread_id": thread_id}}
+
+
+def _at(config: RunnableConfig, snapshot: StateSnapshot) -> RunnableConfig:
+    """Config addressing one specific checkpoint of ``config``'s thread."""
+    return {
+        "configurable": {
+            **config["configurable"],
+            "checkpoint_ns": "",
+            "checkpoint_id": snapshot.config["configurable"]["checkpoint_id"],
+        }
+    }
+
+
+def _input(marker: str) -> dict:
+    return {"log": [marker], "plain": [marker]}
+
+
+def _snapshotted_checkpoints(
+    checkpointer: BaseCheckpointSaver, config: RunnableConfig
+) -> list[str]:
+    """Ids of this thread's checkpoints carrying a ``log`` snapshot blob."""
+    return [
+        tuple_.config["configurable"]["checkpoint_id"]
+        for tuple_ in checkpointer.list(config)
+        if isinstance(tuple_.checkpoint["channel_values"].get("log"), _DeltaSnapshot)
+    ]
+
+
+def _assert_fork_is_clean(state: StateSnapshot, abandoned: str) -> None:
+    """The delta channel must match the plain channel and drop ``abandoned``."""
+    assert state.values["log"] == state.values["plain"], (
+        f"delta channel diverged from the plain channel: "
+        f"{state.values['log']} != {state.values['plain']}"
+    )
+    assert abandoned not in state.values["log"], (
+        f"{abandoned!r} belongs to the branch the fork replaced, "
+        f"but was replayed into {state.values['log']}"
+    )
+
+
+def test_fork_by_invoke(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    _build(sync_checkpointer, "first").invoke(
+        _input("in-1"), config, durability=durability
+    )
+    graph = _build(sync_checkpointer, "second")
+    graph.invoke(_input("in-2"), config, durability=durability)
+
+    # The last checkpoint that predates "in-2" entering state: forking here
+    # abandons the "in-2" branch, whose writes still hang off this checkpoint.
+    base = next(
+        snapshot
+        for snapshot in graph.get_state_history(config)
+        if "in-2" not in snapshot.values["log"]
+    )
+    _build(sync_checkpointer, "third").invoke(
+        _input("in-3"), _at(config, base), durability=durability
+    )
+
+    state = graph.get_state(config)
+    _assert_fork_is_clean(state, "in-2")
+    assert state.values["log"] == [*base.values["log"], "in-3", "third-out"]
+
+
+async def test_afork_by_invoke(
+    async_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    await _build(async_checkpointer, "first").ainvoke(
+        _input("in-1"), config, durability=durability
+    )
+    graph = _build(async_checkpointer, "second")
+    await graph.ainvoke(_input("in-2"), config, durability=durability)
+
+    base = await anext(
+        snapshot
+        async for snapshot in graph.aget_state_history(config)
+        if "in-2" not in snapshot.values["log"]
+    )
+    await _build(async_checkpointer, "third").ainvoke(
+        _input("in-3"), _at(config, base), durability=durability
+    )
+
+    state = await graph.aget_state(config)
+    _assert_fork_is_clean(state, "in-2")
+    assert state.values["log"] == [*base.values["log"], "in-3", "third-out"]
+
+
+def test_fork_off_checkpoint_before_first_input(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    """Fork off the root checkpoint, which predates any value for ``log``.
+
+    ``create_checkpoint`` drops a requested snapshot for a channel absent from
+    ``channel_versions``, so the fork's own first checkpoint cannot carry the
+    blob. The request has to stay queued until a superstep gives the channel a
+    value, otherwise the root's ``in-1`` write still leaks into the fork.
+    """
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "first")
+    graph.invoke(_input("in-1"), config, durability=durability)
+
+    root = list(graph.get_state_history(config))[-1]
+    assert root.values["log"] == []
+
+    _build(sync_checkpointer, "third").invoke(
+        _input("in-9"), _at(config, root), durability=durability
+    )
+
+    state = graph.get_state(config)
+    _assert_fork_is_clean(state, "in-1")
+    assert state.values["log"] == ["in-9", "third-out"]
+
+
+async def test_afork_off_checkpoint_before_first_input(
+    async_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    """Async twin of ``test_fork_off_checkpoint_before_first_input``."""
+    config = _thread("t")
+    graph = _build(async_checkpointer, "first")
+    await graph.ainvoke(_input("in-1"), config, durability=durability)
+
+    root = [snapshot async for snapshot in graph.aget_state_history(config)][-1]
+    assert root.values["log"] == []
+
+    await _build(async_checkpointer, "third").ainvoke(
+        _input("in-9"), _at(config, root), durability=durability
+    )
+
+    state = await graph.aget_state(config)
+    _assert_fork_is_clean(state, "in-1")
+    assert state.values["log"] == ["in-9", "third-out"]
+
+
+def test_fork_by_update_state(sync_checkpointer: BaseCheckpointSaver) -> None:
+    config = _thread("t")
+    _build(sync_checkpointer, "first").invoke(_input("in-1"), config)
+    graph = _build(sync_checkpointer, "second")
+    graph.invoke(_input("in-2"), config)
+
+    base = next(
+        snapshot
+        for snapshot in graph.get_state_history(config)
+        if "in-2" not in snapshot.values["log"]
+    )
+    forked = graph.update_state(_at(config, base), _input("patched"))
+
+    state = graph.get_state(forked)
+    _assert_fork_is_clean(state, "in-2")
+    assert state.values["log"] == [*base.values["log"], "patched"]
+
+
+async def test_afork_by_update_state(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config = _thread("t")
+    await _build(async_checkpointer, "first").ainvoke(_input("in-1"), config)
+    graph = _build(async_checkpointer, "second")
+    await graph.ainvoke(_input("in-2"), config)
+
+    base = await anext(
+        snapshot
+        async for snapshot in graph.aget_state_history(config)
+        if "in-2" not in snapshot.values["log"]
+    )
+    forked = await graph.aupdate_state(_at(config, base), _input("patched"))
+
+    state = await graph.aget_state(forked)
+    _assert_fork_is_clean(state, "in-2")
+    assert state.values["log"] == [*base.values["log"], "patched"]
+
+
+def test_unaddressed_run_keeps_snapshot_cadence(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    """A run with no explicitly addressed checkpoint writes no snapshot blob.
+
+    Guards the cost of the fix: the forced snapshot is one per addressed run,
+    not a change to the normal ``snapshot_frequency`` cadence.
+    """
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "first")
+    graph.invoke(_input("in-1"), config, durability=durability)
+    graph.invoke(_input("in-2"), config, durability=durability)
+
+    assert not _snapshotted_checkpoints(sync_checkpointer, config)
+
+
+def test_fork_before_first_value_when_fork_never_writes_the_channel(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    """Seal the fork even when the channel has no value to snapshot.
+
+    Forking before ``log`` was ever written leaves nothing to copy into the
+    fork's first checkpoint, so without a minted version and an empty blob the
+    boundary goes unrecorded and the walk runs into the base. The fork here
+    never writes ``log`` at all, so no later superstep can seal it either.
+    """
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "first")
+    graph.invoke(_input("in-1"), config, durability=durability)
+
+    root = list(graph.get_state_history(config))[-1]
+    assert root.values["log"] == []
+
+    _build_without_delta_writes(sync_checkpointer, "third").invoke(
+        {"other": ["in-9"]}, _at(config, root), durability=durability
+    )
+
+    state = graph.get_state(config)
+    _assert_fork_is_clean(state, "in-1")
+    assert state.values["log"] == []
+
+
+async def test_afork_before_first_value_when_fork_never_writes_the_channel(
+    async_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    """Async twin of ``test_fork_before_first_value_when_fork_never_writes_the_channel``."""
+    config = _thread("t")
+    graph = _build(async_checkpointer, "first")
+    await graph.ainvoke(_input("in-1"), config, durability=durability)
+
+    root = [snapshot async for snapshot in graph.aget_state_history(config)][-1]
+    assert root.values["log"] == []
+
+    await _build_without_delta_writes(async_checkpointer, "third").ainvoke(
+        {"other": ["in-9"]}, _at(config, root), durability=durability
+    )
+
+    state = await graph.aget_state(config)
+    _assert_fork_is_clean(state, "in-1")
+    assert state.values["log"] == []
+
+
+def test_fork_before_first_value_by_bulk_update(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """The fork's first superstep touches another key, the delta key comes later.
+
+    The first checkpoint has to seal the boundary on its own. Deferring until
+    the superstep that finally writes ``log`` is too late, because that
+    superstep reconstructs through the unsealed checkpoint first.
+    """
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "first")
+    graph.invoke(_input("in-1"), config)
+
+    root = list(graph.get_state_history(config))[-1]
+    assert root.values["log"] == []
+
+    forked = graph.bulk_update_state(
+        _at(config, root),
+        [
+            [StateUpdate({"other": ["s1"]}, "n")],
+            [StateUpdate(_input("s2"), "n")],
+        ],
+    )
+
+    state = graph.get_state(forked)
+    _assert_fork_is_clean(state, "in-1")
+    assert state.values["log"] == ["s2"]
+
+
+@pytest.mark.parametrize("first_as_node", [INPUT, END, "__copy__"])
+def test_fork_by_bulk_update_whose_first_superstep_skips_the_plan(
+    sync_checkpointer: BaseCheckpointSaver, first_as_node: str
+) -> None:
+    """The fork must be sealed by whichever checkpoint the fork writes first.
+
+    ``as_node`` of INPUT, END or ``__copy__`` writes a checkpoint and returns
+    before ``create_checkpoint_plan_for_update_state_api`` runs. If that
+    checkpoint carries no snapshot the branch is still unsealed, so the next
+    superstep reconstructs through the shared base, picks up the abandoned
+    writes, and bakes them into whatever it snapshots. Sealing later is too
+    late: by then the in-memory value is already wrong.
+    """
+    config = _thread("t")
+    _build(sync_checkpointer, "first").invoke(_input("in-1"), config)
+    graph = _build(sync_checkpointer, "second")
+    graph.invoke(_input("in-2"), config)
+
+    base = next(
+        snapshot
+        for snapshot in graph.get_state_history(config)
+        if "in-2" not in snapshot.values["log"]
+    )
+    first = (
+        StateUpdate(_input("first-step"), first_as_node)
+        if first_as_node == INPUT
+        else StateUpdate(None, first_as_node)
+    )
+    forked = graph.bulk_update_state(
+        _at(config, base),
+        [[first], [StateUpdate(_input("second-step"), "n")]],
+    )
+
+    state = graph.get_state(forked)
+    # END legitimately absorbs the base's already-run task writes, so "in-2"
+    # belongs there; the plain channel is the oracle for which is which.
+    assert state.values["log"] == state.values["plain"], (
+        f"delta channel diverged from the plain channel: "
+        f"{state.values['log']} != {state.values['plain']}"
+    )
+
+
+def test_unaddressed_bulk_update_keeps_snapshot_cadence(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    """A multi-superstep ``bulk_update_state`` forks at most once, at the head.
+
+    ``perform_superstep`` returns the config of the checkpoint it just wrote
+    and the driver feeds that back in, so every superstep after the first
+    receives a config naming a checkpoint even when the caller addressed none.
+    Deriving the fork flag from that config snapshots the whole growing value
+    once per superstep.
+    """
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "first")
+    graph.invoke(_input("in-1"), config)
+
+    graph.bulk_update_state(
+        config,
+        [[StateUpdate(_input(f"u{i}"), "n")] for i in range(4)],
+    )
+
+    assert not _snapshotted_checkpoints(sync_checkpointer, config)

libs/langgraph/tests/test_delta_channel_parallel_order.py

@@ -1,184 +0,0 @@
-"""Tests that `DeltaChannel` replay preserves live parallel-write order.
-
-Regression suite for #8382.
-
-`apply_writes` sorts a super-step's tasks by `task_path_str(task.path[:3])`
-before handing their values to `channel.update`, so the order a reducer sees is
-deterministic and independent of which parallel task finishes first. Replay has
-to recover that same order. Ordering a checkpoint's writes by `(task_id, idx)`
-does not: `task_id` is a hash of the path, so for two or more tasks writing one
-`DeltaChannel` in a single super-step it yields an effectively arbitrary
-permutation. Reducers are required to be batching-invariant, not
-order-invariant, so the permutation changes the reconstructed value.
-
-Every test runs against the full `async_checkpointer` matrix — memory, sqlite,
-and postgres in three pool modes — because each saver reconstructs delta
-channels through its own `aget_delta_channel_history` override rather than a
-shared code path, and the three stored `task_path` differently before this fix.
-"""
-
-from itertools import pairwise
-from typing import Annotated, Any
-
-import pytest
-from langgraph.checkpoint.base import BaseCheckpointSaver
-from typing_extensions import TypedDict
-
-from langgraph.channels.delta import DeltaChannel
-from langgraph.graph import END, START, StateGraph
-
-pytestmark = pytest.mark.anyio
-
-# Node names double as the values written. They are listed in sorted order,
-# which is also the order live execution applies them: each node is a PULL task
-# whose path is `("__pregel_pull", name)`, so sorting paths sorts by name.
-FAN_OUT_NAMES = ["a", "b", "c", "d", "e", "f", "g", "h"]
-
-
-def _append_reducer(current: list, updates: list) -> list:
-    """Order-sensitive list accumulation, as in the `DeltaChannel` docstring."""
-    result = list(current)
-    for update in updates:
-        if isinstance(update, list):
-            result.extend(update)
-        else:
-            result.append(update)
-    return result
-
-
-def _build_graph(checkpointer: BaseCheckpointSaver, *, sequential: bool = False) -> Any:
-    """Compile a `DeltaChannel`-backed `items` graph over `FAN_OUT_NAMES`.
-
-    By default every node is wired off `START`, so they all write `items` in one
-    super-step — the shape #8382 is about. `sequential=True` chains them
-    instead, giving one writer per super-step as a control.
-
-    `snapshot_frequency` is far above the number of updates these tests make, so
-    no snapshot is ever written and the value has to come from replaying
-    ancestor writes — the path under test.
-    """
-
-    class State(TypedDict):
-        items: Annotated[
-            list, DeltaChannel(_append_reducer, list, snapshot_frequency=10_000)
-        ]
-
-    def make_node(label: str) -> Any:
-        def node(state: State) -> dict:
-            return {"items": [label]}
-
-        return node
-
-    builder = StateGraph(State)
-    for name in FAN_OUT_NAMES:
-        builder.add_node(name, make_node(name))
-    if sequential:
-        for source, target in pairwise([START, *FAN_OUT_NAMES, END]):
-            builder.add_edge(source, target)
-    else:
-        for name in FAN_OUT_NAMES:
-            builder.add_edge(START, name)
-            builder.add_edge(name, END)
-    return builder.compile(checkpointer=checkpointer)
-
-
-async def test_get_state_matches_live_invoke_order(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """A cold read reports the same order `invoke` returned."""
-    graph = _build_graph(async_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-
-    live = (await graph.ainvoke({"items": []}, config))["items"]
-    replayed = (await graph.aget_state(config)).values["items"]
-
-    assert live == FAN_OUT_NAMES
-    assert replayed == live
-
-
-async def test_continuing_thread_preserves_committed_prefix(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """A second run appends without reordering the first run's items.
-
-    The more serious half of #8382: the reordered replay becomes the base that
-    later writes build on, so the corruption is persisted rather than confined
-    to a read.
-    """
-    graph = _build_graph(async_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-
-    first = (await graph.ainvoke({"items": []}, config))["items"]
-    second = (await graph.ainvoke({"items": []}, config))["items"]
-
-    assert second == first + first
-    assert (await graph.aget_state(config)).values["items"] == second
-
-
-async def test_order_stable_across_many_supersteps(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Order holds over a chain spanning several supersteps with no snapshot."""
-    runs = 5
-    graph = _build_graph(async_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-
-    for _ in range(runs):
-        live = (await graph.ainvoke({"items": []}, config))["items"]
-
-    assert live == FAN_OUT_NAMES * runs
-    assert (await graph.aget_state(config)).values["items"] == live
-
-
-async def test_state_history_reports_live_order_at_every_step(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Every checkpoint in the history replays in live order.
-
-    Guards the walk at intermediate depths, not just from the head. Entries are
-    checked against the order live execution produced rather than against the
-    replayed head — comparing replayed values only to each other passes even
-    when every one of them is permuted the same wrong way.
-    """
-    runs = 3
-    graph = _build_graph(async_checkpointer)
-    config = {"configurable": {"thread_id": "1"}}
-
-    for _ in range(runs):
-        await graph.ainvoke({"items": []}, config)
-    live_expected = FAN_OUT_NAMES * runs
-
-    seen = [
-        snapshot.values["items"]
-        async for snapshot in graph.aget_state_history(config)
-        if "items" in snapshot.values
-    ]
-
-    assert seen, "expected at least one snapshot carrying `items`"
-    # The deepest entry is the head, so the matrix below covers the full value
-    # as well as every partial prefix.
-    assert max(len(values) for values in seen) == len(live_expected)
-    for values in seen:
-        assert values == live_expected[: len(values)], (
-            f"history entry {values} is not the live order "
-            f"{live_expected[: len(values)]}"
-        )
-
-
-async def test_sequential_graph_unaffected(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """One writer per super-step replays correctly with or without the fix.
-
-    Control: it localises #8382 to multiple tasks writing one channel in a
-    single super-step, rather than to delta replay in general. This is the one
-    test here that passes on main.
-    """
-    graph = _build_graph(async_checkpointer, sequential=True)
-    config = {"configurable": {"thread_id": "1"}}
-
-    live = (await graph.ainvoke({"items": []}, config))["items"]
-    replayed = (await graph.aget_state(config)).values["items"]
-
-    assert live == FAN_OUT_NAMES
-    assert replayed == live