release(langgraph): 1.2.7 (#8223)

## Summary

Releases `langgraph` 1.2.7.

Bumps the package version `1.2.6` -> `1.2.7` and propagates it into the
`langgraph`, `prebuilt`, and `sdk-py` lockfiles. No dependency floor or
source changes.

## Changes

- Update `libs/langgraph/pyproject.toml` to version `1.2.7`.
- Update the editable `langgraph` package entries in
`libs/langgraph/uv.lock`, `libs/prebuilt/uv.lock`, and
`libs/sdk-py/uv.lock`.

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

@@ -54,12 +54,6 @@ async def build_delta_chain(
     stored: list[RunnableConfig] = []
     parent_cfg: RunnableConfig | None = None
 
-    # Track supersteps since the channel's last snapshot, mirroring Pregel's
-    # `counters_since_delta_snapshot` bookkeeping: every step increments the
-    # counter; a snapshot step resets it (and carries no counter entry, so
-    # the walk locates the seed exactly `supersteps` hops back).
-    supersteps_since_snapshot = 0
-
     for step in range(total_steps):
         config: RunnableConfig = {
             "configurable": {
@@ -74,19 +68,11 @@ async def build_delta_chain(
 
         channel_values: dict[str, Any] = {}
         channel_versions: dict[str, int] = {}
-        supersteps_since_snapshot += 1
-        counters: dict[str, tuple[int, int]] = {}
         if step in snapshot_set:
             channel_values[channel] = _DeltaSnapshot(
                 write_value_fn(step),
             )
             channel_versions[channel] = step + 1
-            supersteps_since_snapshot = 0
-        else:
-            counters[channel] = (
-                supersteps_since_snapshot,
-                supersteps_since_snapshot,
-            )
 
         cp = Checkpoint(
             v=1,
@@ -98,10 +84,9 @@ async def build_delta_chain(
             updated_channels=None,
         )
         new_versions = dict(channel_versions)
-        md = generate_metadata(step=step)
-        if counters:
-            md["counters_since_delta_snapshot"] = counters
-        parent_cfg = await saver.aput(config, cp, md, new_versions)
+        parent_cfg = await saver.aput(
+            config, cp, generate_metadata(step=step), new_versions
+        )
         stored.append(parent_cfg)
 
         # Write a pending write for non-snapshot steps so the walk has

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

@@ -87,10 +87,6 @@ async def test_history_multi_channel(
 
     from langgraph.checkpoint.conformance.test_utils import generate_metadata
 
-    # Per-channel supersteps since last snapshot ("a" snapshots at step 1,
-    # "b" at step 3). Drives the per-channel walk depth from the head.
-    s_a = 0
-    s_b = 0
     for step in range(5):
         config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
         if parent_cfg:
@@ -99,21 +95,12 @@ async def test_history_multi_channel(
             ]
         cv: dict = {}
         cvs: dict = {}
-        s_a += 1
-        s_b += 1
-        counters: dict = {}
         if step == 1:
             cv["a"] = _DeltaSnapshot("snap_a")
             cvs["a"] = step + 1
-            s_a = 0
-        else:
-            counters["a"] = (s_a, s_a)
         if step == 3:
             cv["b"] = _DeltaSnapshot("snap_b")
             cvs["b"] = step + 1
-            s_b = 0
-        else:
-            counters["b"] = (s_b, s_b)
         cp = Checkpoint(
             v=1,
             id=str(uuid6(clock_seq=-1)),
@@ -123,10 +110,7 @@ async def test_history_multi_channel(
             versions_seen={},
             updated_channels=None,
         )
-        md = generate_metadata(step=step)
-        if counters:
-            md["counters_since_delta_snapshot"] = counters
-        parent_cfg = await saver.aput(config, cp, md, cvs)
+        parent_cfg = await saver.aput(config, cp, generate_metadata(step=step), cvs)
         configs.append(parent_cfg)
         await saver.aput_writes(parent_cfg, [("a", step), ("b", step)], str(uuid4()))
 
@@ -164,12 +148,7 @@ async def test_history_walk_to_root_no_seed(
     )
     head = configs[-1]
     result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
-    # The head's supersteps counter (4) runs one hop past the real root —
-    # the snapshot checkpoint was never persisted (implicit empty baseline).
-    # No seed, and the full chain (steps 0..2) replays from empty.
     assert "seed" not in result["ch"], f"Expected no seed, got {result['ch']}"
-    values = [w[2] for w in result["ch"]["writes"]]
-    assert values == [0, 1, 2], f"Expected full-chain writes [0,1,2], got {values}"
 
 
 async def test_history_migration_plain_value_as_seed(
@@ -191,10 +170,6 @@ async def test_history_migration_plain_value_as_seed(
     configs: list = []
     parent_cfg = None
 
-    # The channel was a non-delta channel through step 1 (plain value, no
-    # delta counter), then migrated to DeltaChannel at step 2. Supersteps
-    # count from the migration boundary: step 2 -> 1, step 3 -> 2, so the
-    # head (step 3) walks 2 hops back to the plain-value seed at step 1.
     for step in range(4):
         config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
         if parent_cfg:
@@ -207,9 +182,6 @@ async def test_history_migration_plain_value_as_seed(
         if step == 1:
             cv["ch"] = [10, 20, 30]
             cvs["ch"] = step + 1
-        md = generate_metadata(step=step)
-        if step >= 2:
-            md["counters_since_delta_snapshot"] = {"ch": (step - 1, step - 1)}
         cp = Checkpoint(
             v=1,
             id=str(uuid6(clock_seq=-1)),
@@ -219,7 +191,7 @@ async def test_history_migration_plain_value_as_seed(
             versions_seen={},
             updated_channels=None,
         )
-        parent_cfg = await saver.aput(config, cp, md, cvs)
+        parent_cfg = await saver.aput(config, cp, generate_metadata(step=step), cvs)
         configs.append(parent_cfg)
         if step != 1:
             await saver.aput_writes(parent_cfg, [("ch", step)], str(uuid4()))
@@ -236,67 +208,6 @@ async def test_history_migration_plain_value_as_seed(
     assert values == [2], f"Expected [2], got {values}"
 
 
-async def test_history_migration_skips_seed_checkpoint_writes(
-    saver: BaseCheckpointSaver,
-) -> None:
-    """A migrated plain-value seed already incorporates its own checkpoint's
-    writes, so those writes must NOT be re-emitted in the history.
-
-    This is the discriminating case for non-additive reducers (e.g. an
-    even-only filter): re-applying the seed checkpoint's writes on top of
-    the seed would corrupt the reconstructed value. The saver contract is
-    to skip them — only writes strictly after the seed are returned.
-    """
-    from langgraph.checkpoint.base import Checkpoint
-    from langgraph.checkpoint.base.id import uuid6
-
-    from langgraph.checkpoint.conformance.test_utils import generate_metadata
-
-    tid = str(uuid4())
-    configs: list = []
-    parent_cfg = None
-
-    for step in range(4):
-        config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
-        if parent_cfg:
-            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
-                "checkpoint_id"
-            ]
-        cv: dict = {}
-        cvs: dict = {}
-        # Step 1: pre-delta plain accumulated value AND its own writes.
-        if step == 1:
-            cv["ch"] = [2, 4]
-            cvs["ch"] = step + 1
-        md = generate_metadata(step=step)
-        if step >= 2:
-            md["counters_since_delta_snapshot"] = {"ch": (step - 1, step - 1)}
-        cp = Checkpoint(
-            v=1,
-            id=str(uuid6(clock_seq=-1)),
-            ts="",
-            channel_values=cv,
-            channel_versions=cvs,
-            versions_seen={},
-            updated_channels=None,
-        )
-        parent_cfg = await saver.aput(config, cp, md, cvs)
-        configs.append(parent_cfg)
-        # The seed checkpoint (step 1) carries writes that the plain value
-        # already subsumes; later steps carry post-migration delta writes.
-        write_value = 99 if step == 1 else step * 10
-        await saver.aput_writes(parent_cfg, [("ch", write_value)], str(uuid4()))
-
-    head = configs[-1]
-    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
-    assert result["ch"].get("seed") == [2, 4], (
-        f"Expected plain seed [2, 4], got {result['ch'].get('seed')}"
-    )
-    values = [w[2] for w in result["ch"]["writes"]]
-    # Seed checkpoint's own write (99) is skipped; step 2's write (20) kept.
-    assert values == [20], f"Expected only post-seed writes [20], got {values}"
-
-
 ALL_DELTA_CHANNEL_HISTORY_TESTS = [
     test_history_returns_writes_oldest_first,
     test_history_seed_is_nearest_snapshot,
@@ -305,7 +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_migration_skips_seed_checkpoint_writes,
 ]
 
 

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

@@ -14,7 +14,6 @@ from langgraph.checkpoint.base import (
     CheckpointMetadata,
     CheckpointTuple,
     DeltaChannelHistory,
-    _parse_supersteps_since_last_snapshot_by_channel,
     get_checkpoint_id,
     get_serializable_checkpoint_metadata,
 )
@@ -29,11 +28,9 @@ from langgraph.checkpoint.postgres import _internal
 from langgraph.checkpoint.postgres.base import (
     _DELTA_PAGE_SIZE,
     BasePostgresSaver,
-    _advance_shared_chain,
-    _build_delta_fetch_sql,
-    _build_delta_walk_sql,
+    _build_delta_stage1_sql,
+    _build_delta_stage2_sql,
     _DeltaStage2Row,
-    _ingest_walk_page,
 )
 from langgraph.checkpoint.postgres.shallow import ShallowPostgresSaver
 
@@ -449,137 +446,107 @@ class PostgresSaver(BasePostgresSaver):
     ) -> Mapping[str, DeltaChannelHistory]:
         """Fast-path override of `BaseCheckpointSaver.get_delta_channel_history`.
 
-        Reconstructs each delta channel's state for the target checkpoint by
-        walking the parent chain `supersteps` hops (read from the target's
-        `counters_since_delta_snapshot` metadata) to its seed snapshot, then
-        collecting the writes between the seed and the target.
+        Two-stage query, both stages cover ALL requested channels:
 
-        Two passes (see the `# Multi-channel two-pass` comment in `base.py`):
+        * Stage 1 (paged): dynamic SELECT over `checkpoints` with K parallel
+          JSONB key lookups (one column pair per channel) — no subquery, no
+          aggregation. Pages newest-first by `checkpoint_id` with a cursor;
+          page size is `_DELTA_PAGE_SIZE`. Stops paging when every channel
+          has found its seed or the chain is exhausted.
 
-        * WALK (paged): dynamic SELECT over `checkpoints` with K parallel
-          JSONB lookups for `channel_versions[ch]` (the seed blob version),
-          following `parent_checkpoint_id` newest-first. Page size is
-          `_DELTA_PAGE_SIZE`; stops once the shared chain reaches the deepest
-          requested `supersteps`, the root is reached, or the chain is
-          exhausted.
-
-        * FETCH (per-channel UNION ALL): one branch per channel reading
-          `checkpoint_writes` for its `chain_cids`, plus one branch per
-          channel that located a seed reading `checkpoint_blobs` at the
-          seed's version.
+        * Stage 2 (per-channel UNION ALL): one branch per channel reading
+          `checkpoint_writes` filtered to that channel's specific
+          `chain_cids`, plus one branch per channel that has a seed reading
+          `checkpoint_blobs` for that channel + version. Avoids the
+          over-fetch of a single `channel = ANY(channels)` filter when
+          channels have different chain depths.
         """
         if not channels:
             return {}
         channels = list(channels)
         thread_id = config["configurable"]["thread_id"]
-        if not thread_id:
-            raise ValueError("empty thread ID")
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
-
-        # Resolve the target checkpoint id + its metadata (for supersteps).
         checkpoint_id = get_checkpoint_id(config)
-        with self._cursor() as cur:
-            if checkpoint_id is None:
-                cur.execute(
-                    "SELECT checkpoint_id, metadata FROM checkpoints "
-                    "WHERE thread_id = %s AND checkpoint_ns = %s "
-                    "ORDER BY checkpoint_id DESC LIMIT 1",
-                    (thread_id, checkpoint_ns),
-                )
-            else:
-                cur.execute(
-                    "SELECT checkpoint_id, metadata FROM checkpoints "
-                    "WHERE thread_id = %s AND checkpoint_ns = %s "
-                    "AND checkpoint_id = %s",
-                    (thread_id, checkpoint_ns, checkpoint_id),
-                )
-            target_row = cur.fetchone()
-        if target_row is None:
-            return {ch: {"writes": []} for ch in channels}
-        target_id = cast(str, target_row["checkpoint_id"])
-        supersteps_by_ch = _parse_supersteps_since_last_snapshot_by_channel(
-            target_row["metadata"] or {}, channels
-        )
-        max_supersteps = max(supersteps_by_ch.values(), default=0)
+        if checkpoint_id is None:
+            target = self.get_tuple(config)
+            if target is None:
+                return {ch: {"writes": []} for ch in channels}
+            checkpoint_id = target.config["configurable"]["checkpoint_id"]
 
-        # WALK: page the parent chain, bounded by the deepest supersteps.
-        walk_sql = _build_delta_walk_sql(channels)
+        # Stage 1: paged K-JSONB-lookup scan, walking the parent chain in
+        # Python after each page. Stops as soon as every channel has its seed.
+        stage1_sql = _build_delta_stage1_sql(channels, paged=True)
         parent_of: dict[str, str | None] = {}
         ver_by_i_by_cid: list[dict[str, str | None]] = [{} for _ in channels]
-        shared_cpid_chain: list[str] = []
-        has_reached_root = False
+        hs_by_i_by_cid: list[dict[str, bool]] = [{} for _ in channels]
+        chain_by_ch: dict[str, list[str]] = {ch: [] for ch in channels}
+        seed_ver_by_ch: dict[str, str | None] = {ch: None for ch in channels}
+        walk_cursor_by_ch: dict[str, str | None] = {}
+        seeded: set[str] = set()
         cursor: str | None = None
 
-        while max_supersteps > 0:
-            walk_params: list[Any] = [
-                *channels,
-                thread_id,
-                checkpoint_ns,
-                cursor,
-                cursor,
-                _DELTA_PAGE_SIZE,
-            ]
-            with self._cursor() as cur:
-                cur.execute(walk_sql, walk_params)
+        with self._cursor() as cur:
+            while True:
+                stage1_params: list[Any] = []
+                for ch in channels:
+                    stage1_params.extend([ch, ch])
+                stage1_params.extend(
+                    [thread_id, checkpoint_ns, cursor, cursor, _DELTA_PAGE_SIZE]
+                )
+                cur.execute(stage1_sql, stage1_params)
                 page = cur.fetchall()
-            if not page:
-                break
-            oldest = _ingest_walk_page(
-                cast("list[Mapping[str, Any]]", page),
-                channels,
-                parent_of,
-                ver_by_i_by_cid,
-            )
-            has_reached_root = _advance_shared_chain(
-                target_id, parent_of, shared_cpid_chain, max_supersteps
-            )
-            if (
-                has_reached_root
-                or len(shared_cpid_chain) >= max_supersteps
-                or len(page) < _DELTA_PAGE_SIZE
-            ):
-                break
-            cursor = oldest
+                if not page:
+                    break
+                oldest = self._ingest_stage1_page(
+                    cast("list[Mapping[str, Any]]", page),
+                    channels,
+                    parent_of,
+                    ver_by_i_by_cid,
+                    hs_by_i_by_cid,
+                )
+                self._try_advance_walks(
+                    checkpoint_id,
+                    channels,
+                    parent_of,
+                    ver_by_i_by_cid,
+                    hs_by_i_by_cid,
+                    chain_by_ch,
+                    seed_ver_by_ch,
+                    walk_cursor_by_ch,
+                    seeded,
+                )
+                # Stop if every channel is seeded, or the page was short
+                # (chain exhausted — no more rows to fetch).
+                if len(seeded) == len(channels) or len(page) < _DELTA_PAGE_SIZE:
+                    break
+                cursor = oldest
 
-        chained_cpid_by_ch, seed_cpid_by_ch, seed_ver_by_ch = (
-            self._resolve_delta_chains(
-                channels,
-                supersteps_by_ch,
-                shared_cpid_chain,
-                ver_by_i_by_cid,
-                has_reached_root,
-                thread_id,
-            )
-        )
-
-        # FETCH: per-channel UNION ALL — writes branch per chained channel,
-        # blob branch per seeded channel.
-        channels_with_chain = [ch for ch in channels if chained_cpid_by_ch[ch]]
+        # Stage 2: per-channel UNION ALL — one writes branch per channel
+        # with non-empty chain, plus one blob branch per seeded channel.
+        channels_with_chain = [ch for ch in channels if chain_by_ch[ch]]
         channels_with_seed = [ch for ch in channels if seed_ver_by_ch[ch] is not None]
-        fetch_sql = _build_delta_fetch_sql(
+        stage2_sql = _build_delta_stage2_sql(
             channels_with_chain=channels_with_chain,
             channels_with_seed=channels_with_seed,
         )
-        if fetch_sql:
-            fetch_params: list[Any] = []
-            for ch in channels_with_chain:
-                fetch_params.extend(
-                    [thread_id, checkpoint_ns, ch, chained_cpid_by_ch[ch]]
-                )
-            for ch in channels_with_seed:
-                fetch_params.extend([thread_id, checkpoint_ns, ch, seed_ver_by_ch[ch]])
-            with self._cursor() as cur:
-                cur.execute(fetch_sql, fetch_params)
-                fetch_rows = cur.fetchall()
-        else:
-            fetch_rows = []
 
-        return self._assemble_delta_history(
+        if stage2_sql:
+            stage2_params: list[Any] = []
+            for ch in channels_with_chain:
+                stage2_params.extend([thread_id, checkpoint_ns, ch, chain_by_ch[ch]])
+            for ch in channels_with_seed:
+                stage2_params.extend([thread_id, checkpoint_ns, ch, seed_ver_by_ch[ch]])
+            with self._cursor() as cur:
+                cur.execute(stage2_sql, stage2_params)
+                stage2_rows = cur.fetchall()
+        else:
+            stage2_rows = []
+
+        return self._build_delta_channels_writes_history(
             channels=channels,
-            chained_cpid_by_ch=chained_cpid_by_ch,
-            seed_cpid_by_ch=seed_cpid_by_ch,
+            chain_by_ch=chain_by_ch,
             seed_ver_by_ch=seed_ver_by_ch,
-            fetch_rows=cast("list[_DeltaStage2Row]", fetch_rows),
+            stage2_rows=cast("list[_DeltaStage2Row]", stage2_rows),
         )
 
     def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:

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

@@ -14,7 +14,6 @@ from langgraph.checkpoint.base import (
     CheckpointMetadata,
     CheckpointTuple,
     DeltaChannelHistory,
-    _parse_supersteps_since_last_snapshot_by_channel,
     get_checkpoint_id,
     get_serializable_checkpoint_metadata,
 )
@@ -29,11 +28,9 @@ from langgraph.checkpoint.postgres import _ainternal
 from langgraph.checkpoint.postgres.base import (
     _DELTA_PAGE_SIZE,
     BasePostgresSaver,
-    _advance_shared_chain,
-    _build_delta_fetch_sql,
-    _build_delta_walk_sql,
+    _build_delta_stage1_sql,
+    _build_delta_stage2_sql,
     _DeltaStage2Row,
-    _ingest_walk_page,
 )
 from langgraph.checkpoint.postgres.shallow import AsyncShallowPostgresSaver
 
@@ -411,122 +408,89 @@ class AsyncPostgresSaver(BasePostgresSaver):
         """Fast-path override of `BaseCheckpointSaver.aget_delta_channel_history`.
 
         See `PostgresSaver.get_delta_channel_history` for design notes; this is
-        the async equivalent: paged supersteps-bounded WALK + per-channel
-        UNION ALL FETCH.
+        the async equivalent with internal stage-1 paging and per-channel
+        UNION ALL stage-2.
         """
         if not channels:
             return {}
         channels = list(channels)
         thread_id = config["configurable"]["thread_id"]
-        if not thread_id:
-            raise ValueError("empty thread ID")
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
-
-        # Resolve the target checkpoint id + its metadata (for supersteps).
         checkpoint_id = get_checkpoint_id(config)
-        async with self._cursor() as cur:
-            if checkpoint_id is None:
-                await cur.execute(
-                    "SELECT checkpoint_id, metadata FROM checkpoints "
-                    "WHERE thread_id = %s AND checkpoint_ns = %s "
-                    "ORDER BY checkpoint_id DESC LIMIT 1",
-                    (thread_id, checkpoint_ns),
-                )
-            else:
-                await cur.execute(
-                    "SELECT checkpoint_id, metadata FROM checkpoints "
-                    "WHERE thread_id = %s AND checkpoint_ns = %s "
-                    "AND checkpoint_id = %s",
-                    (thread_id, checkpoint_ns, checkpoint_id),
-                )
-            target_row = await cur.fetchone()
-        if target_row is None:
-            return {ch: {"writes": []} for ch in channels}
-        target_id = cast(str, target_row["checkpoint_id"])
-        supersteps_by_ch = _parse_supersteps_since_last_snapshot_by_channel(
-            target_row["metadata"] or {}, channels
-        )
-        max_supersteps = max(supersteps_by_ch.values(), default=0)
+        if checkpoint_id is None:
+            target = await self.aget_tuple(config)
+            if target is None:
+                return {ch: {"writes": []} for ch in channels}
+            checkpoint_id = target.config["configurable"]["checkpoint_id"]
 
-        # WALK: page the parent chain, bounded by the deepest supersteps.
-        walk_sql = _build_delta_walk_sql(channels)
+        stage1_sql = _build_delta_stage1_sql(channels, paged=True)
         parent_of: dict[str, str | None] = {}
         ver_by_i_by_cid: list[dict[str, str | None]] = [{} for _ in channels]
-        shared_cpid_chain: list[str] = []
-        has_reached_root = False
+        hs_by_i_by_cid: list[dict[str, bool]] = [{} for _ in channels]
+        chain_by_ch: dict[str, list[str]] = {ch: [] for ch in channels}
+        seed_ver_by_ch: dict[str, str | None] = {ch: None for ch in channels}
+        walk_cursor_by_ch: dict[str, str | None] = {}
+        seeded: set[str] = set()
         cursor: str | None = None
 
-        while max_supersteps > 0:
-            walk_params: list[Any] = [
-                *channels,
-                thread_id,
-                checkpoint_ns,
-                cursor,
-                cursor,
-                _DELTA_PAGE_SIZE,
-            ]
-            async with self._cursor() as cur:
-                await cur.execute(walk_sql, walk_params)
+        async with self._cursor() as cur:
+            while True:
+                stage1_params: list[Any] = []
+                for ch in channels:
+                    stage1_params.extend([ch, ch])
+                stage1_params.extend(
+                    [thread_id, checkpoint_ns, cursor, cursor, _DELTA_PAGE_SIZE]
+                )
+                await cur.execute(stage1_sql, stage1_params)
                 page = await cur.fetchall()
-            if not page:
-                break
-            oldest = _ingest_walk_page(
-                cast("list[Mapping[str, Any]]", page),
-                channels,
-                parent_of,
-                ver_by_i_by_cid,
-            )
-            has_reached_root = _advance_shared_chain(
-                target_id, parent_of, shared_cpid_chain, max_supersteps
-            )
-            if (
-                has_reached_root
-                or len(shared_cpid_chain) >= max_supersteps
-                or len(page) < _DELTA_PAGE_SIZE
-            ):
-                break
-            cursor = oldest
+                if not page:
+                    break
+                oldest = self._ingest_stage1_page(
+                    cast("list[Mapping[str, Any]]", page),
+                    channels,
+                    parent_of,
+                    ver_by_i_by_cid,
+                    hs_by_i_by_cid,
+                )
+                self._try_advance_walks(
+                    checkpoint_id,
+                    channels,
+                    parent_of,
+                    ver_by_i_by_cid,
+                    hs_by_i_by_cid,
+                    chain_by_ch,
+                    seed_ver_by_ch,
+                    walk_cursor_by_ch,
+                    seeded,
+                )
+                if len(seeded) == len(channels) or len(page) < _DELTA_PAGE_SIZE:
+                    break
+                cursor = oldest
 
-        chained_cpid_by_ch, seed_cpid_by_ch, seed_ver_by_ch = (
-            self._resolve_delta_chains(
-                channels,
-                supersteps_by_ch,
-                shared_cpid_chain,
-                ver_by_i_by_cid,
-                has_reached_root,
-                thread_id,
-            )
-        )
-
-        # FETCH: per-channel UNION ALL — writes branch per chained channel,
-        # blob branch per seeded channel.
-        channels_with_chain = [ch for ch in channels if chained_cpid_by_ch[ch]]
+        channels_with_chain = [ch for ch in channels if chain_by_ch[ch]]
         channels_with_seed = [ch for ch in channels if seed_ver_by_ch[ch] is not None]
-        fetch_sql = _build_delta_fetch_sql(
+        stage2_sql = _build_delta_stage2_sql(
             channels_with_chain=channels_with_chain,
             channels_with_seed=channels_with_seed,
         )
 
-        if fetch_sql:
-            fetch_params: list[Any] = []
+        if stage2_sql:
+            stage2_params: list[Any] = []
             for ch in channels_with_chain:
-                fetch_params.extend(
-                    [thread_id, checkpoint_ns, ch, chained_cpid_by_ch[ch]]
-                )
+                stage2_params.extend([thread_id, checkpoint_ns, ch, chain_by_ch[ch]])
             for ch in channels_with_seed:
-                fetch_params.extend([thread_id, checkpoint_ns, ch, seed_ver_by_ch[ch]])
+                stage2_params.extend([thread_id, checkpoint_ns, ch, seed_ver_by_ch[ch]])
             async with self._cursor() as cur:
-                await cur.execute(fetch_sql, fetch_params)
-                fetch_rows = await cur.fetchall()
+                await cur.execute(stage2_sql, stage2_params)
+                stage2_rows = await cur.fetchall()
         else:
-            fetch_rows = []
+            stage2_rows = []
 
-        return self._assemble_delta_history(
+        return self._build_delta_channels_writes_history(
             channels=channels,
-            chained_cpid_by_ch=chained_cpid_by_ch,
-            seed_cpid_by_ch=seed_cpid_by_ch,
+            chain_by_ch=chain_by_ch,
             seed_ver_by_ch=seed_ver_by_ch,
-            fetch_rows=cast("list[_DeltaStage2Row]", fetch_rows),
+            stage2_rows=cast("list[_DeltaStage2Row]", stage2_rows),
         )
 
     async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:

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

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import logging
 import random
 import warnings
 from collections.abc import Mapping, Sequence
@@ -16,12 +15,10 @@ from langgraph.checkpoint.base import (
     PendingWrite,
     get_checkpoint_id,
 )
-from langgraph.checkpoint.serde.types import TASKS, _DeltaSnapshot
+from langgraph.checkpoint.serde.types import TASKS
 from psycopg.types.json import Jsonb
 
-logger = logging.getLogger(__name__)
-
-# Page size for the paged WALK scan in `get_delta_channel_history`. Internal
+# Page size for stage-1 paged scan in `get_delta_channel_history`. Internal
 # constant — exposing this as a kwarg is left as a follow-up.
 _DELTA_PAGE_SIZE = 1024
 
@@ -163,7 +160,7 @@ INSERT_CHECKPOINT_WRITES_SQL = """
 
 
 class _DeltaStage2Row(TypedDict, total=False):
-    """One row from `_build_delta_fetch_sql` (a UNION ALL of writes and blobs)."""
+    """One row from `_build_delta_stage2_sql` (a UNION ALL of writes and blobs)."""
 
     _kind: str  # "w" or "b"
     checkpoint_id: str | None  # "w" rows only
@@ -175,58 +172,42 @@ class _DeltaStage2Row(TypedDict, total=False):
     version: str | None  # "b" rows only
 
 
-# Multi-channel two-pass DeltaChannel reconstruction.
+# Multi-channel two-stage DeltaChannel reconstruction.
 #
-# A `DeltaChannel` does not store its full value at every checkpoint — it
-# stores periodic full-value *snapshots* and accumulates intermediate
-# *writes* between snapshots. To rebuild a channel's value at a target
-# checkpoint we need:
+# Stage 1 scans checkpoint metadata (no blob bytes) and emits one row per
+# checkpoint with K parallel JSONB key lookups (one column pair per
+# requested delta channel: ver_i / hs_i).  No subqueries, no aggregation.
+# Python walks the parent chain once across all channels.
 #
-#   - the **seed** — the most recent snapshot at-or-before the target
-#     (a single blob row in `checkpoint_blobs`); and
-#   - the **chain writes** — every write for this channel committed
-#     between that snapshot and the target, in order
-#     (rows in `checkpoint_writes`).
+# Stage 2 fetches all writes and the seed blobs for ALL channels in a
+# single roundtrip via `channel = ANY(%s)` and chain/seed-version
+# filtering.
 #
-# Two passes, in order:
+# Empirical comparison vs an alternative "ship full channel_versions /
+# channel_values JSONB and let Python pick" form (1000 checkpoints,
+# 8 total channels in graph, 3 delta channels requested):
 #
-#   1. WALK — scan checkpoint metadata only (no blob bytes). For each
-#      requested channel we read `channel_versions[ch]` (the seed's blob
-#      version pointer). Then follow `parent_checkpoint_id` from the
-#      target backwards in pages of `_DELTA_PAGE_SIZE` rows.
+#   Postgres execution:    A=0.24ms vs B=0.38ms   (both negligible)
+#   End-to-end latency:    A=6.83ms vs B=2.28ms   (B is 3.0x faster)
+#   Wire payload:          A=836KB  vs B=330KB    (61% smaller)
+#   Buffer hits:           identical (167 blocks)
 #
-#      Walk depth is driven by the *supersteps since last snapshot*
-#      counter — `metadata.counters_since_delta_snapshot[ch][1]` — read
-#      from the target checkpoint. A channel's seed snapshot sits exactly
-#      `supersteps` hops back along the parent chain; walking by the
-#      counter (rather than scanning `channel_values` for the snapshot
-#      marker) is the only reliable way to locate seeds that aren't a
-#      `_DeltaSnapshot` sentinel — e.g. legacy plain-value blobs left by a
-#      thread that migrated from a non-delta channel, which `put` stores
-#      out of the inline `channel_values` map.
-#
-#   2. FETCH — given each channel's chained checkpoint ids and seed
-#      version from WALK, pull only the rows we need: writes for those
-#      exact checkpoint_ids and the seed blob at that exact version. One
-#      roundtrip, per-channel UNION ALL — no over-fetch.
-#
-# Walking the *parent chain* (not `list(before=...)`) matters: forked
-# threads have multiple branches, and only on-path ancestors contribute.
+# B (this dynamic-columns design) wins because it avoids JSONB
+# serialization on the wire and JSONB-to-dict deserialization in
+# psycopg.  Even at K=8 (8 delta channels = 16 dynamic columns), B
+# still beats A end-to-end (4.2ms vs 6.8ms).
 
 
-def _build_delta_walk_sql(channels: Sequence[str]) -> str:
-    """Build the paged WALK SQL — scans checkpoint metadata only.
+def _build_delta_stage1_sql(channels: Sequence[str], *, paged: bool) -> str:
+    """Build stage 1 SQL with 2K parallel JSONB key lookups.
 
-    Emits one row per checkpoint with K parallel JSONB key lookups (one
-    `ver_i` column per requested channel — the channel's blob version, the
-    pointer we dereference in FETCH if this checkpoint is the seed). No
-    blob bytes; the result set fits a paged `LIMIT` cleanly.
-
-    For channels=["messages", "files"] the result is::
+    For channels=["messages", "files"] (with `paged=True`) the result is::
 
         SELECT checkpoint_id, parent_checkpoint_id,
                checkpoint -> 'channel_versions' ->> %s AS ver_0,
-               checkpoint -> 'channel_versions' ->> %s AS ver_1
+               (checkpoint -> 'channel_values' -> %s) IS NOT NULL AS hs_0,
+               checkpoint -> 'channel_versions' ->> %s AS ver_1,
+               (checkpoint -> 'channel_values' -> %s) IS NOT NULL AS hs_1
         FROM checkpoints
         WHERE thread_id = %s AND checkpoint_ns = %s
           AND (%s::text IS NULL OR checkpoint_id < %s)
@@ -234,38 +215,45 @@ def _build_delta_walk_sql(channels: Sequence[str]) -> str:
         LIMIT %s
 
     Channel names are passed as `%s` parameters (safe from SQL injection).
-    Only the column aliases `ver_i` are interpolated into the SQL string
-    (i is bounded by len(channels) and uses safe identifiers).
+    Only the column aliases `ver_i` / `hs_i` are interpolated into the
+    SQL string (i is bounded by len(channels) and uses safe identifiers).
 
-    Caller must extend params with `[ch_0, ch_1, ..., thread_id, ns,
-    cursor, cursor, page_size]`. The `cursor` is the smallest
-    `checkpoint_id` from the previous page (or `None` on the first page);
-    `(%s::text IS NULL OR ...)` makes the first-page `WHERE` a no-op.
+    Caller must extend params with `[ch_0, ch_0, ch_1, ch_1, ...,
+    thread_id, ns, cursor, cursor, page_size]` when `paged=True`.
+
+    When `paged=False`, the WHERE has no cursor predicate and there's no
+    LIMIT/ORDER BY — kept as a non-public helper for tests/diagnostics.
     """
-    cols = [
-        f"checkpoint -> 'channel_versions' ->> %s AS ver_{i}"
-        for i in range(len(channels))
-    ]
-    return (
+    cols = []
+    for i in range(len(channels)):
+        cols.append(
+            f"checkpoint -> 'channel_versions' ->> %s AS ver_{i}, "
+            f"(checkpoint -> 'channel_values' -> %s) IS NOT NULL AS hs_{i}"
+        )
+    sql = (
         "SELECT checkpoint_id, parent_checkpoint_id, "
         + ", ".join(cols)
         + " FROM checkpoints WHERE thread_id = %s AND checkpoint_ns = %s"
-        " AND (%s::text IS NULL OR checkpoint_id < %s)"
-        " ORDER BY checkpoint_id DESC LIMIT %s"
     )
+    if paged:
+        sql += (
+            " AND (%s::text IS NULL OR checkpoint_id < %s)"
+            " ORDER BY checkpoint_id DESC LIMIT %s"
+        )
+    return sql
 
 
-def _build_delta_fetch_sql(
+def _build_delta_stage2_sql(
     *,
     channels_with_chain: Sequence[str],
     channels_with_seed: Sequence[str],
 ) -> str:
-    """Build the FETCH SQL as a per-channel UNION ALL.
+    """Build stage 2 SQL as a per-channel UNION ALL.
 
     For each channel with a non-empty chain, emit one branch reading
     `checkpoint_writes` for that specific channel + chain_cids. For each
     channel with a seed_version, emit one branch reading `checkpoint_blobs`
-    for that channel + version. This avoids the over-fetch of a single
+    for that channel + version. This avoids the over-fetch of the prior
     `channel = ANY(channels) AND checkpoint_id = ANY(union)` form when
     channels have different chain depths.
 
@@ -281,7 +269,6 @@ def _build_delta_fetch_sql(
     """
     branches: list[str] = []
     for _ in channels_with_chain:
-        # NOTE: no ORDER BY on this branch — writes are sorted in assembly.
         branches.append(
             "SELECT 'w'::text AS _kind, "
             "checkpoint_id, channel, "
@@ -301,59 +288,10 @@ def _build_delta_fetch_sql(
     return " UNION ALL ".join(branches)
 
 
-def _ingest_walk_page(
-    page_rows: Sequence[Mapping[str, Any]],
-    channels: Sequence[str],
-    parent_of: dict[str, str | None],
-    ver_by_i_by_cid: list[dict[str, str | None]],
-) -> str | None:
-    """Fold one WALK page into `parent_of` + per-channel `ver_by_cid`.
-
-    Returns the oldest checkpoint_id seen on this page (smallest, since
-    pages come back DESC). Caller uses it as the cursor for the next page
-    (`AND checkpoint_id < cursor`).
-    """
-    oldest: str | None = None
-    for r in page_rows:
-        cid = cast(str, r["checkpoint_id"])
-        parent_of[cid] = cast("str | None", r["parent_checkpoint_id"])
-        for i in range(len(channels)):
-            ver_by_i_by_cid[i][cid] = cast("str | None", r.get(f"ver_{i}"))
-        # Rows are DESC; the last one is the smallest cid in the page.
-        oldest = cid
-    return oldest
-
-
-def _advance_shared_chain(
-    target_id: str,
-    parent_of: Mapping[str, str | None],
-    shared_cpid_chain: list[str],
-    max_supersteps: int,
-) -> bool:
-    """Extend the shared parent chain as far as ingested pages allow.
-
-    The chain holds ancestors of the target, newest first: `chain[0]` is
-    the target's parent, `chain[1]` its grandparent, and so on. A single
-    chain is shared across all channels and grown to the maximum requested
-    depth; each channel later slices `chain[:supersteps]`.
-
-    Stops when:
-      - the chain reaches `max_supersteps` hops, OR
-      - the root is reached (parent is None) — returns True, OR
-      - the next ancestor's cid isn't in `parent_of` yet (waits for the
-        next page).
-
-    Returns True iff the root was reached.
-    """
-    while len(shared_cpid_chain) < max_supersteps:
-        top_of_chain = shared_cpid_chain[-1] if shared_cpid_chain else target_id
-        if top_of_chain not in parent_of:
-            return False  # wait for the next page
-        parent = parent_of[top_of_chain]
-        if parent is None:
-            return True  # hit the root
-        shared_cpid_chain.append(parent)
-    return False
+# Stage 1 rows are dynamic-shape dicts: {checkpoint_id, parent_checkpoint_id,
+# ver_0, hs_0, ver_1, hs_1, ...}.  Walking is parameterized by the channel
+# list to map indices back to channel names — no static TypedDict here.
+# `dict[str, Any]` is the practical signature.
 
 
 class BasePostgresSaver(BaseCheckpointSaver[str]):
@@ -398,92 +336,107 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
             if t.decode() != "empty"
         }
 
-    def _resolve_delta_chains(
-        self,
+    @staticmethod
+    def _ingest_stage1_page(
+        stage1_rows: Sequence[Mapping[str, Any]],
         channels: Sequence[str],
-        supersteps_by_ch: Mapping[str, int],
-        shared_cpid_chain: Sequence[str],
-        ver_by_i_by_cid: Sequence[Mapping[str, str | None]],
-        has_reached_root: bool,
-        thread_id: str,
-    ) -> tuple[
-        dict[str, list[str]],
-        dict[str, str | None],
-        dict[str, str | None],
-    ]:
-        """Slice the shared parent chain into per-channel chain/seed mappings.
+        parent_of: dict[str, str | None],
+        ver_by_i_by_cid: list[dict[str, str | None]],
+        hs_by_i_by_cid: list[dict[str, bool]],
+    ) -> str | None:
+        """Fold one stage-1 page into the running walk-state mappings.
 
-        For each channel the seed snapshot sits `supersteps` hops back, so
-        the seed checkpoint is `shared_cpid_chain[supersteps - 1]` and the
-        chain is `shared_cpid_chain[:supersteps]` (newest first).
-
-        When the chain is shorter than `supersteps` but the walk reached the
-        root, the persisted chain is "compressed" relative to the logical
-        superstep count — either because intermediate supersteps were never
-        persisted (`durability="exit"`) or because the thread never produced
-        a snapshot at all. In both cases the seed candidate is the oldest
-        persisted checkpoint (`shared_cpid_chain[-1]`): FETCH loads its blob,
-        and assembly keeps it only if non-empty (a real snapshot or migrated
-        value) — otherwise it omits `seed` and replays the full chain on an
-        empty baseline.
+        Returns the oldest checkpoint_id seen on this page (smallest, since
+        pages come back DESC). Caller uses it as the cursor for the next
+        page (`AND checkpoint_id < cursor`).
         """
-        chained_cpid_by_ch: dict[str, list[str]] = {ch: [] for ch in channels}
-        seed_cpid_by_ch: dict[str, str | None] = {ch: None for ch in channels}
-        seed_ver_by_ch: dict[str, str | None] = {ch: None for ch in channels}
-        for i, ch in enumerate(channels):
-            bound = supersteps_by_ch.get(ch, 0)
-            if bound <= 0:
-                continue
-            if len(shared_cpid_chain) >= bound:
-                seed_depth = bound
-            elif has_reached_root:
-                seed_depth = len(shared_cpid_chain)
-            else:
-                logger.warning(
-                    "cannot find seed snapshot for delta channel "
-                    "(thread_id=%s, channel=%s)",
-                    thread_id,
-                    ch,
-                )
-                continue
-            if seed_depth <= 0:
-                continue
-            chained_cpid_by_ch[ch] = list(shared_cpid_chain[:seed_depth])
-            seed_cpid_by_ch[ch] = shared_cpid_chain[seed_depth - 1]
-            seed_ver_by_ch[ch] = ver_by_i_by_cid[i].get(seed_cpid_by_ch[ch])
-        return chained_cpid_by_ch, seed_cpid_by_ch, seed_ver_by_ch
+        oldest: str | None = None
+        for r in stage1_rows:
+            cid = cast(str, r["checkpoint_id"])
+            parent_of[cid] = cast("str | None", r["parent_checkpoint_id"])
+            for i in range(len(channels)):
+                ver_by_i_by_cid[i][cid] = cast("str | None", r.get(f"ver_{i}"))
+                hs_by_i_by_cid[i][cid] = bool(r.get(f"hs_{i}"))
+            # Rows are DESC; the last one is the smallest cid in the page.
+            oldest = cid
+        return oldest
 
-    def _assemble_delta_history(
+    @staticmethod
+    def _try_advance_walks(
+        target_id: str,
+        channels: Sequence[str],
+        parent_of: Mapping[str, str | None],
+        ver_by_i_by_cid: Sequence[Mapping[str, str | None]],
+        hs_by_i_by_cid: Sequence[Mapping[str, bool]],
+        chain_by_ch: dict[str, list[str]],
+        seed_ver_by_ch: dict[str, str | None],
+        walk_cursor_by_ch: dict[str, str | None],
+        seeded: set[str],
+    ) -> None:
+        """Advance each not-yet-seeded channel's walk as far as possible.
+
+        Uses the partial `parent_of` map accumulated so far. A walk stops
+        either because:
+          (a) it found a snapshot for its channel (channel becomes seeded),
+          (b) it reached a real root (parent_of[cid] is None — fully
+              materialized at this point), or
+          (c) the next ancestor cid isn't in `parent_of` yet (waiting for
+              a later page; the cursor stays put).
+
+        Mutates `chain_by_ch`, `seed_ver_by_ch`, `walk_cursor_by_ch`, and
+        `seeded` in place.
+        """
+        for i, ch in enumerate(channels):
+            if ch in seeded:
+                continue
+            # First-time entry: cursor starts at the target's parent.
+            if ch not in walk_cursor_by_ch:
+                walk_cursor_by_ch[ch] = parent_of.get(target_id)
+            cur_cid = walk_cursor_by_ch[ch]
+            ch_chain = chain_by_ch[ch]
+            hs_i = hs_by_i_by_cid[i]
+            ver_i = ver_by_i_by_cid[i]
+            while cur_cid is not None:
+                if cur_cid not in parent_of:
+                    # Need more pages to continue this walk.
+                    break
+                ch_chain.append(cur_cid)
+                if hs_i.get(cur_cid, False):
+                    seed_ver_by_ch[ch] = ver_i.get(cur_cid)
+                    seeded.add(ch)
+                    cur_cid = None
+                    break
+                cur_cid = parent_of[cur_cid]
+            walk_cursor_by_ch[ch] = cur_cid
+
+    def _build_delta_channels_writes_history(
         self,
         *,
         channels: Sequence[str],
-        chained_cpid_by_ch: Mapping[str, Sequence[str]],
-        seed_cpid_by_ch: Mapping[str, str | None],
+        chain_by_ch: Mapping[str, list[str]],
         seed_ver_by_ch: Mapping[str, str | None],
-        fetch_rows: Sequence[_DeltaStage2Row],
+        stage2_rows: Sequence[_DeltaStage2Row],
     ) -> dict[str, DeltaChannelHistory]:
-        """Demux FETCH rows per channel and produce per-channel histories.
+        """Demux stage 2 rows per channel; produce per-channel histories.
 
-        `fetch_rows` carry `channel` on every row. Write rows (`_kind = 'w'`)
-        are bucketed per channel per checkpoint; seed-blob rows
-        (`_kind = 'b'`) give each channel its snapshot value.
-
-        The seed checkpoint's own writes are replayed on top of a
-        `_DeltaSnapshot` seed (the snapshot is the value *prior* to its own
-        writes), but skipped for a migrated plain-value seed (a legacy
-        non-delta blob already incorporates those writes). The `seed` key is
-        omitted when no seed was located or the blob is the "empty"
-        tombstone — the consumer treats absence as "start empty".
+        stage2_rows carry `channel` on every row. We build per-channel
+        `writes_by_cid` and per-channel `seed_blob` dicts, then assemble
+        a `DeltaChannelHistory` per requested channel. The `seed` key is omitted
+        when the walk reached root with no snapshot found, 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)
         writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
             ch: {} for ch in channels
         }
-        seed_blob_by_ch: dict[str, tuple[str, bytes]] = {}
+        # seed_blob_by_ver[(channel, version)] = (type, blob)
+        seed_blob_by_ver: dict[tuple[str, str], tuple[str, bytes]] = {}
 
-        for r in fetch_rows:
+        for r in stage2_rows:
             ch = cast(str, r["channel"])
-            if r["_kind"] == "w":
+            kind = r["_kind"]
+            if kind == "w":
                 cid = cast(str, r["checkpoint_id"])
                 writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
                     cast(
@@ -491,39 +444,35 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
                         (r["type"], r["blob"], r["task_id"], r["idx"]),
                     )
                 )
-            else:  # _kind == "b" — the seed blob for this channel.
-                seed_blob_by_ch[ch] = cast("tuple[str, bytes]", (r["type"], r["blob"]))
+            else:  # kind == "b"
+                ver = cast(str, r["version"])
+                seed_blob_by_ver[(ch, ver)] = cast(
+                    "tuple[str, bytes]", (r["type"], r["blob"])
+                )
 
-        # Within a checkpoint, writes apply oldest→newest by (task_id, idx).
+        # 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[2], w[3]))
+                ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
 
         result: dict[str, DeltaChannelHistory] = {}
         for ch in channels:
-            entry: DeltaChannelHistory = {"writes": []}
-
-            skip_seed_checkpoint_writes = False
-            seed_blob = seed_blob_by_ch.get(ch)
-            if seed_blob is not None and seed_blob[0] != "empty":
-                seed_value = self.serde.loads_typed(seed_blob)
-                entry["seed"] = seed_value
-                # A migrated (non-delta) seed already includes the writes on
-                # its own checkpoint; a `_DeltaSnapshot` does not.
-                skip_seed_checkpoint_writes = not isinstance(seed_value, _DeltaSnapshot)
+            chain_cids = chain_by_ch.get(ch, [])
+            seed_version = seed_ver_by_ch.get(ch)
 
+            collected: list[PendingWrite] = []
             cid_writes = writes_by_ch_by_cid.get(ch, {})
-            if cid_writes:
-                collected: list[PendingWrite] = []
-                seed_cpid = seed_cpid_by_ch.get(ch)
-                # Chain is newest→oldest; replay oldest→newest.
-                for cid in reversed(chained_cpid_by_ch.get(ch, [])):
-                    if skip_seed_checkpoint_writes and cid == seed_cpid:
-                        continue
-                    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))
-                entry["writes"] = collected
+            for cid in chain_cids:
+                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()
+
+            entry: DeltaChannelHistory = {"writes": collected}
+            if seed_version is not None:
+                blob = seed_blob_by_ver.get((ch, seed_version))
+                if blob is not None and blob[0] != "empty":
+                    entry["seed"] = self.serde.loads_typed(blob)
             result[ch] = entry
         return result
 

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

@@ -24,12 +24,10 @@ from langgraph.checkpoint.base import (
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
 
 from langgraph.checkpoint.sqlite._delta import (
-    DELTA_WALK_SQL,
+    DELTA_STAGE1_SQL,
     build_delta_channels_writes_history,
-    build_delta_writes_fetch_sql,
-    parse_supersteps_since_last_snapshot_by_channel,
-    resolve_delta_chains,
-    step_walk_supersteps,
+    build_delta_stage2_sql,
+    step_walk_with_row,
 )
 from langgraph.checkpoint.sqlite.utils import search_where
 
@@ -507,106 +505,68 @@ class SqliteSaver(BaseCheckpointSaver[str]):
     ) -> Mapping[str, DeltaChannelHistory]:
         """Fast-path override of `BaseCheckpointSaver.get_delta_channel_history`.
 
-        Walks the parent chain `supersteps` hops (read from the target's
-        `counters_since_delta_snapshot` metadata) to each channel's seed
-        snapshot, then collects the writes between the seed and the target.
+        Two-stage query:
 
-        * WALK: stream a newest-first slice of `checkpoints` returning
+        * Stage 1 (paged): newest-first slice of `checkpoints` returning
           `(checkpoint_id, parent_checkpoint_id, type, checkpoint)` per
-          ancestor. Sqlite has no JSONB, so we deserialize only the seed
-          checkpoints to read their inline `channel_values`. Stops once the
-          shared chain reaches the deepest requested `supersteps` or the
-          root is reached.
+          ancestor. Sqlite has no JSONB, so we ship the full serialized
+          checkpoint blob and inspect `channel_values` in Python. Pages
+          newest-first by `checkpoint_id` with a `< cursor` predicate;
+          page size is `DELTA_PAGE_SIZE`. Stops paging when every channel
+          has found its seed or the chain is exhausted.
 
-        * FETCH (per-channel UNION ALL): one branch per channel reading
-          `writes` filtered to that channel's `chain_cids`. No separate
-          seed-blob fetch — sqlite stores `channel_values` inline.
+        * Stage 2 (per-channel UNION ALL): one branch per channel reading
+          `writes` filtered to that channel's specific `chain_cids`. No
+          separate seed-blob fetch — sqlite stores `channel_values` inline
+          in the checkpoint blob, so seeds come back from stage 1.
         """
         if not channels:
             return {}
         channels = list(channels)
         thread_id = str(config["configurable"]["thread_id"])
-        if not thread_id:
-            raise ValueError("empty thread ID")
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
-
-        # Resolve the target checkpoint id + its metadata (for supersteps).
         checkpoint_id = get_checkpoint_id(config)
-        with self.cursor(transaction=False) as cur:
-            if checkpoint_id is None:
-                cur.execute(
-                    "SELECT checkpoint_id, metadata FROM checkpoints "
-                    "WHERE thread_id = ? AND checkpoint_ns = ? "
-                    "ORDER BY checkpoint_id DESC LIMIT 1",
-                    (thread_id, checkpoint_ns),
-                )
-            else:
-                cur.execute(
-                    "SELECT checkpoint_id, metadata FROM checkpoints "
-                    "WHERE thread_id = ? AND checkpoint_ns = ? "
-                    "AND checkpoint_id = ?",
-                    (thread_id, checkpoint_ns, checkpoint_id),
-                )
-            target_row = cur.fetchone()
-        if target_row is None:
-            return {ch: {"writes": []} for ch in channels}
-        target_id = str(target_row[0])
-        metadata = json.loads(target_row[1]) if target_row[1] is not None else {}
-        supersteps_by_ch = parse_supersteps_since_last_snapshot_by_channel(
-            metadata, channels
-        )
-        max_supersteps = max(supersteps_by_ch.values(), default=0)
-        needed_depths = set(supersteps_by_ch.values())
+        if checkpoint_id is None:
+            target = self.get_tuple(config)
+            if target is None:
+                return {ch: {"writes": []} for ch in channels}
+            checkpoint_id = target.config["configurable"]["checkpoint_id"]
 
-        shared_cpid_chain: list[str] = []
+        chain_by_ch: dict[str, list[str]] = {ch: [] for ch in channels}
+        seed_val_by_ch: dict[str, Any] = {}
         walk_state: dict[str, Any] = {}
-        seed_values_by_depth: dict[int, dict[str, Any]] = {}
+        seeded: set[str] = set()
 
         with self.cursor(transaction=False) as cur:
-            if max_supersteps > 0:
-                cur.execute(DELTA_WALK_SQL, (thread_id, checkpoint_ns, target_id))
-                for row in cur:
-                    cid, parent_cid, type_tag, blob = row
-                    if step_walk_supersteps(
-                        cid=cid,
-                        parent_cid=parent_cid,
-                        type_tag=type_tag,
-                        blob=blob,
-                        target_id=target_id,
-                        serde=self.serde,
-                        shared_cpid_chain=shared_cpid_chain,
-                        walk_state=walk_state,
-                        max_supersteps=max_supersteps,
-                        needed_depths=needed_depths,
-                        seed_values_by_depth=seed_values_by_depth,
-                        channels=channels,
-                    ):
-                        break
+            cur.execute(DELTA_STAGE1_SQL, (thread_id, checkpoint_ns, checkpoint_id))
+            for row in cur:
+                cid, parent_cid, type_tag, blob = row
+                if step_walk_with_row(
+                    cid=cid,
+                    parent_cid=parent_cid,
+                    type_tag=type_tag,
+                    blob=blob,
+                    target_id=checkpoint_id,
+                    serde=self.serde,
+                    chain_by_ch=chain_by_ch,
+                    seed_val_by_ch=seed_val_by_ch,
+                    walk_state=walk_state,
+                    seeded=seeded,
+                    channels=channels,
+                ):
+                    break
 
-            (
-                chained_cpid_by_ch,
-                seed_cpid_by_ch,
-                seed_value_by_ch,
-            ) = resolve_delta_chains(
-                channels=channels,
-                supersteps_by_ch=supersteps_by_ch,
-                shared_cpid_chain=shared_cpid_chain,
-                seed_values_by_depth=seed_values_by_depth,
-                has_reached_root=bool(walk_state.get("reached_root")),
-                thread_id=thread_id,
+            channels_with_chain = [ch for ch in channels if chain_by_ch[ch]]
+            stage2_sql = build_delta_stage2_sql(
+                chain_lens=[len(chain_by_ch[ch]) for ch in channels_with_chain],
             )
-
-            channels_with_chain = [ch for ch in channels if chained_cpid_by_ch[ch]]
-            fetch_sql = build_delta_writes_fetch_sql(
-                chain_lens=[len(chained_cpid_by_ch[ch]) for ch in channels_with_chain],
-            )
-            if fetch_sql:
-                fetch_params: list[Any] = []
+            if stage2_sql:
+                stage2_params: list[Any] = []
                 for ch in channels_with_chain:
-                    fetch_params.extend(
-                        [thread_id, checkpoint_ns, ch, *chained_cpid_by_ch[ch]]
+                    stage2_params.extend(
+                        [thread_id, checkpoint_ns, ch, *chain_by_ch[ch]]
                     )
-                cur.execute(fetch_sql, fetch_params)
+                cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
                     "list[tuple[str, str, str, int, str, bytes]]", cur.fetchall()
                 )
@@ -615,9 +575,9 @@ class SqliteSaver(BaseCheckpointSaver[str]):
 
         return build_delta_channels_writes_history(
             channels=channels,
-            chained_cpid_by_ch=chained_cpid_by_ch,
-            seed_cpid_by_ch=seed_cpid_by_ch,
-            seed_value_by_ch=seed_value_by_ch,
+            chain_by_ch=chain_by_ch,
+            seed_val_by_ch=seed_val_by_ch,
+            seeded=seeded,
             stage2_rows=stage2_rows,
             serde=self.serde,
         )

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

@@ -1,50 +1,37 @@
 """Shared helpers for `get_delta_channel_history` on sqlite savers.
 
-Mirrors the supersteps-based two-pass shape of `BasePostgresSaver`
-(ancestor walk bounded by `counters_since_delta_snapshot` + per-channel
-UNION ALL writes fetch), adapted for sqlite's constraints:
+Mirrors the two-stage shape of `BasePostgresSaver` (ancestor walk +
+per-channel UNION ALL writes fetch), but adapted for sqlite's
+constraints. The structural differences:
 
 * No JSONB — to inspect `channel_values` for a checkpoint we must
-  deserialize the full blob. The WALK streams the cursor row-by-row and
-  deserializes only the seed checkpoints (the ones at a channel's
-  `supersteps` depth), freeing each blob before advancing.
+  deserialize the full blob. Stage 1 streams the cursor row-by-row and
+  deserializes only the rows the merged walk visits, freeing each blob
+  before advancing.
 * No separate blob table — `channel_values` lives inline in the
-  checkpoint, so seeds come back from the WALK with no second fetch.
-* Single shared parent-chain walk: each requested channel slices the
-  same chain to its own `supersteps` depth.
+  checkpoint, so seeds come back from stage 1 with no second fetch.
+* Single merged walk (not K independent walks): each visited cid is
+  deserialized exactly once, regardless of how many channels are still
+  seeking their seed.
 
-Walk depth is driven by the *supersteps since last snapshot* counter,
-not by scanning `channel_values` for the snapshot marker — the only
-reliable way to locate seeds that aren't a `_DeltaSnapshot` sentinel
-(e.g. legacy plain-value blobs from a thread migrated off a non-delta
-channel).
+The streaming design keeps peak in-flight memory at roughly one
+deserialized checkpoint at a time, instead of holding the entire
+ancestor chain's worth of raw blobs as a `fetchall()`-materialized list.
 """
 
 from __future__ import annotations
 
-import logging
 from collections.abc import Mapping, Sequence
 from typing import Any
 
-from langgraph.checkpoint.base import (
-    DeltaChannelHistory,
-    PendingWrite,
-    _parse_supersteps_since_last_snapshot_by_channel,
-)
-from langgraph.checkpoint.serde.types import _DeltaSnapshot
+from langgraph.checkpoint.base import DeltaChannelHistory, PendingWrite
 
-logger = logging.getLogger(__name__)
-
-# Re-exported under the package-local name used by the sqlite savers.
-parse_supersteps_since_last_snapshot_by_channel = (
-    _parse_supersteps_since_last_snapshot_by_channel
-)
-
-# The WALK streams ancestors of `target_id` newest-first. The `<=`
+# Stage 1 streams ancestors of `target_cid` newest-first. The `<=`
 # predicate keeps target itself in the stream so we can read its
-# `parent_checkpoint_id` from the first matching row without a separate
-# lookup; target's own writes/seed are not part of the contract.
-DELTA_WALK_SQL = (
+# `parent_checkpoint_id` from the first row without a separate lookup;
+# the caller skips target's own writes/seed (matches the
+# `BaseCheckpointSaver` contract).
+DELTA_STAGE1_SQL = (
     "SELECT checkpoint_id, parent_checkpoint_id, type, checkpoint "
     "FROM checkpoints "
     "WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id <= ? "
@@ -52,12 +39,12 @@ DELTA_WALK_SQL = (
 )
 
 
-def build_delta_writes_fetch_sql(*, chain_lens: Sequence[int]) -> str:
-    """Per-channel UNION ALL fetching writes from `writes`.
+def build_delta_stage2_sql(*, chain_lens: Sequence[int]) -> str:
+    """Stage-2 per-channel UNION ALL fetching writes from `writes`.
 
     One branch per channel with a non-empty chain. Each branch inlines its
     own `IN (?, ?, ...)` placeholder list because sqlite has no array-bind
-    equivalent of postgres's `= ANY(?)`. Caller passes parameters in
+    equivalent of postgres's `= ANY(%s)`. Caller passes parameters in
     matching order: `[thread_id, checkpoint_ns, channel, *chain_cids]` per
     branch.
 
@@ -78,7 +65,7 @@ def build_delta_writes_fetch_sql(*, chain_lens: Sequence[int]) -> str:
     return " UNION ALL ".join(branches)
 
 
-def step_walk_supersteps(
+def step_walk_with_row(
     *,
     cid: str,
     parent_cid: str | None,
@@ -86,142 +73,75 @@ def step_walk_supersteps(
     blob: bytes,
     target_id: str,
     serde: Any,
-    shared_cpid_chain: list[str],
+    chain_by_ch: dict[str, list[str]],
+    seed_val_by_ch: dict[str, Any],
     walk_state: dict[str, Any],
-    max_supersteps: int,
-    needed_depths: set[int],
-    seed_values_by_depth: dict[int, dict[str, Any]],
+    seeded: set[str],
     channels: Sequence[str],
 ) -> bool:
-    """Process one streamed WALK row, extending the shared parent chain.
+    """Process one streamed stage-1 row in the merged ancestor walk.
 
-    The cursor returns `(cid, parent_cid, type, checkpoint)` rows in
-    `checkpoint_id` DESC order starting at target. The first row is target
-    itself; we read its `parent_cid` to seed the walk and skip it (target's
-    own writes/seed are not part of the contract). Off-path rows (a sibling
-    branch on the same thread) advance the cursor without doing work.
+    The cursor returns (cid, parent_cid, type, blob) rows in
+    `checkpoint_id` DESC order starting at target. The first row is
+    target itself; we read its parent_cid to seed the walk and otherwise
+    skip it (target's own writes/seed are not part of the contract).
 
-    For each on-path ancestor we append its cid to `shared_cpid_chain`
-    (newest first). When the ancestor sits at a depth some channel needs as
-    its seed (`len(chain)` ∈ `needed_depths`), we deserialize it once and
-    record its `channel_values` for the requested channels. The
-    deserialized checkpoint is dropped immediately — peak in-flight is one
-    deserialized checkpoint.
+    For each subsequent row, if `cid` matches the walk's current
+    position, we deserialize the blob, append the cid to every
+    not-yet-seeded channel's chain, and check `channel_values` for
+    seeds. The deserialized checkpoint is dropped before advancing — no
+    cross-row cache, so peak in-flight is one deserialized checkpoint.
 
-    Sets `walk_state["reached_root"]` when an ancestor has no parent.
-    Returns True when the walk can stop: chain reached `max_supersteps`, or
-    the root was reached.
+    Off-path rows (different branch on the same thread) advance the
+    cursor without doing any work.
+
+    Returns True when every requested channel is seeded — the caller
+    can stop iterating and close the cursor.
     """
     if "started" not in walk_state:
         if cid == target_id:
             walk_state["started"] = True
             walk_state["cur_cid"] = parent_cid
-            if parent_cid is None:
-                walk_state["reached_root"] = True
-                return True
+            walk_state["active"] = {ch for ch in channels if ch not in seeded}
         # Not target yet (or target not present): keep streaming.
         return False
-    if len(shared_cpid_chain) >= max_supersteps:
+    active: set[str] = walk_state["active"]
+    if not active:
         return True
     if cid != walk_state["cur_cid"]:
         # Off-path row from a sibling branch — skip without deserializing.
         return False
-    shared_cpid_chain.append(cid)
-    depth = len(shared_cpid_chain)  # 1-indexed position along the chain
-    # Capture channel_values at any depth a channel may use as its seed: the
-    # exact `supersteps` depths, plus the root-most checkpoint (the seed
-    # candidate when the chain is shorter than `supersteps`).
-    if depth in needed_depths or parent_cid is None:
-        ckpt = serde.loads_typed((type_tag, blob))
-        channel_values: Mapping[str, Any] = ckpt.get("channel_values") or {}
-        seed_values_by_depth[depth] = {
-            ch: channel_values[ch] for ch in channels if ch in channel_values
-        }
-        del ckpt, channel_values
-    if parent_cid is None:
-        walk_state["reached_root"] = True
-        return True
+    for ch in active:
+        chain_by_ch[ch].append(cid)
+    ckpt = serde.loads_typed((type_tag, blob))
+    channel_values: Mapping[str, Any] = ckpt.get("channel_values") or {}
+    for ch in [ch for ch in active if ch in channel_values]:
+        seed_val_by_ch[ch] = channel_values[ch]
+        seeded.add(ch)
+        active.discard(ch)
+    del ckpt, channel_values
     walk_state["cur_cid"] = parent_cid
-    return len(shared_cpid_chain) >= max_supersteps
-
-
-def resolve_delta_chains(
-    *,
-    channels: Sequence[str],
-    supersteps_by_ch: Mapping[str, int],
-    shared_cpid_chain: Sequence[str],
-    seed_values_by_depth: Mapping[int, Mapping[str, Any]],
-    has_reached_root: bool,
-    thread_id: str,
-) -> tuple[
-    dict[str, list[str]],
-    dict[str, str | None],
-    dict[str, Any],
-]:
-    """Slice the shared parent chain into per-channel chain/seed mappings.
-
-    For each channel the seed snapshot sits `supersteps` hops back: the seed
-    checkpoint is `shared_cpid_chain[supersteps - 1]` and the chain is
-    `shared_cpid_chain[:supersteps]` (newest first), with the seed's inline
-    `channel_values[ch]` captured during the walk.
-
-    When the chain is shorter than `supersteps` but the walk reached the
-    root, the persisted chain is "compressed" relative to the logical
-    superstep count (`durability="exit"`, or a thread that never
-    snapshotted). The seed candidate is then the oldest persisted checkpoint
-    (`shared_cpid_chain[-1]`); if its `channel_values[ch]` is absent the
-    channel has no seed and replays the full chain on an empty baseline.
-    """
-    chained_cpid_by_ch: dict[str, list[str]] = {ch: [] for ch in channels}
-    seed_cpid_by_ch: dict[str, str | None] = {ch: None for ch in channels}
-    seed_value_by_ch: dict[str, Any] = {}
-    for ch in channels:
-        bound = supersteps_by_ch.get(ch, 0)
-        if bound <= 0:
-            continue
-        if len(shared_cpid_chain) >= bound:
-            seed_depth = bound
-        elif has_reached_root:
-            seed_depth = len(shared_cpid_chain)
-        else:
-            logger.warning(
-                "cannot find seed snapshot for delta channel "
-                "(thread_id=%s, channel=%s)",
-                thread_id,
-                ch,
-            )
-            continue
-        if seed_depth <= 0:
-            continue
-        chained_cpid_by_ch[ch] = list(shared_cpid_chain[:seed_depth])
-        seed_cpid_by_ch[ch] = shared_cpid_chain[seed_depth - 1]
-        seed_vals = seed_values_by_depth.get(seed_depth, {})
-        if ch in seed_vals:
-            seed_value_by_ch[ch] = seed_vals[ch]
-    return chained_cpid_by_ch, seed_cpid_by_ch, seed_value_by_ch
+    return not active
 
 
 def build_delta_channels_writes_history(
     *,
     channels: Sequence[str],
-    chained_cpid_by_ch: Mapping[str, Sequence[str]],
-    seed_cpid_by_ch: Mapping[str, str | None],
-    seed_value_by_ch: Mapping[str, Any],
+    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]],
     serde: Any,
 ) -> dict[str, DeltaChannelHistory]:
-    """Demux writes rows per channel; produce per-channel histories.
+    """Demux stage-2 rows per channel; produce per-channel histories.
 
-    `stage2_rows` are `(checkpoint_id, channel, task_id, idx, type, value)`.
+    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`.
 
-    The seed checkpoint's own writes are replayed on top of a
-    `_DeltaSnapshot` seed (the snapshot is the value *prior* to its own
-    writes), but skipped for a migrated plain-value seed (a legacy
-    non-delta blob already incorporates those writes). `seed` is omitted
-    when no seed was located (implicit empty baseline); consumers treat
-    absence as "start empty".
+    `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]]]] = {
         ch: {} for ch in channels
@@ -236,26 +156,17 @@ def build_delta_channels_writes_history(
 
     result: dict[str, DeltaChannelHistory] = {}
     for ch in channels:
-        entry: DeltaChannelHistory = {"writes": []}
-
-        skip_seed_checkpoint_writes = False
-        if ch in seed_value_by_ch:
-            seed_value = seed_value_by_ch[ch]
-            entry["seed"] = seed_value
-            skip_seed_checkpoint_writes = not isinstance(seed_value, _DeltaSnapshot)
-
+        chain_cids = chain_by_ch.get(ch, [])
         cid_writes = writes_by_ch_by_cid.get(ch, {})
-        if cid_writes:
-            collected: list[PendingWrite] = []
-            seed_cpid = seed_cpid_by_ch.get(ch)
-            # Chain is newest→oldest; replay oldest→newest.
-            for cid in reversed(list(chained_cpid_by_ch.get(ch, []))):
-                if skip_seed_checkpoint_writes and cid == seed_cpid:
-                    continue
-                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)))
-                    )
-            entry["writes"] = collected
+        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 in cid_writes.get(cid, []):
+                collected.append(
+                    (task_id, ch, serde.loads_typed((type_tag, value_blob)))
+                )
+        entry: DeltaChannelHistory = {"writes": collected}
+        if ch in seeded:
+            entry["seed"] = seed_val_by_ch[ch]
         result[ch] = entry
     return result

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

@@ -25,12 +25,10 @@ from langgraph.checkpoint.base import (
 from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
 
 from langgraph.checkpoint.sqlite._delta import (
-    DELTA_WALK_SQL,
+    DELTA_STAGE1_SQL,
     build_delta_channels_writes_history,
-    build_delta_writes_fetch_sql,
-    parse_supersteps_since_last_snapshot_by_channel,
-    resolve_delta_chains,
-    step_walk_supersteps,
+    build_delta_stage2_sql,
+    step_walk_with_row,
 )
 from langgraph.checkpoint.sqlite.utils import search_where
 
@@ -627,98 +625,61 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
         """Fast-path override of `BaseCheckpointSaver.aget_delta_channel_history`.
 
         See `SqliteSaver.get_delta_channel_history` for design notes; this
-        is the async equivalent using `aiosqlite` cursors. The WALK streams
-        the parent chain newest-first, bounded by the target's
-        `counters_since_delta_snapshot` supersteps, and deserializes only
-        seed checkpoints; FETCH pulls the relevant writes via per-channel
-        UNION ALL.
+        is the async equivalent using `aiosqlite` cursors. Stage 1 pages
+        the parent chain newest-first and Python-deserializes each
+        checkpoint blob to find per-channel snapshots; stage 2 fetches
+        only the relevant writes via per-channel UNION ALL.
         """
         if not channels:
             return {}
         channels = list(channels)
         await self.setup()
         thread_id = str(config["configurable"]["thread_id"])
-        if not thread_id:
-            raise ValueError("empty thread ID")
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
-
-        # Resolve the target checkpoint id + its metadata (for supersteps).
         checkpoint_id = get_checkpoint_id(config)
-        async with self.lock, self.conn.cursor() as cur:
-            if checkpoint_id is None:
-                await cur.execute(
-                    "SELECT checkpoint_id, metadata FROM checkpoints "
-                    "WHERE thread_id = ? AND checkpoint_ns = ? "
-                    "ORDER BY checkpoint_id DESC LIMIT 1",
-                    (thread_id, checkpoint_ns),
-                )
-            else:
-                await cur.execute(
-                    "SELECT checkpoint_id, metadata FROM checkpoints "
-                    "WHERE thread_id = ? AND checkpoint_ns = ? "
-                    "AND checkpoint_id = ?",
-                    (thread_id, checkpoint_ns, checkpoint_id),
-                )
-            target_row = await cur.fetchone()
-        if target_row is None:
-            return {ch: {"writes": []} for ch in channels}
-        target_id = str(target_row[0])
-        metadata = json.loads(target_row[1]) if target_row[1] is not None else {}
-        supersteps_by_ch = parse_supersteps_since_last_snapshot_by_channel(
-            metadata, channels
-        )
-        max_supersteps = max(supersteps_by_ch.values(), default=0)
-        needed_depths = set(supersteps_by_ch.values())
+        if checkpoint_id is None:
+            target = await self.aget_tuple(config)
+            if target is None:
+                return {ch: {"writes": []} for ch in channels}
+            checkpoint_id = target.config["configurable"]["checkpoint_id"]
 
-        shared_cpid_chain: list[str] = []
+        chain_by_ch: dict[str, list[str]] = {ch: [] for ch in channels}
+        seed_val_by_ch: dict[str, Any] = {}
         walk_state: dict[str, Any] = {}
-        seed_values_by_depth: dict[int, dict[str, Any]] = {}
+        seeded: set[str] = set()
 
         async with self.lock, self.conn.cursor() as cur:
-            if max_supersteps > 0:
-                await cur.execute(DELTA_WALK_SQL, (thread_id, checkpoint_ns, target_id))
-                async for row in cur:
-                    cid, parent_cid, type_tag, blob = row
-                    if step_walk_supersteps(
-                        cid=cid,
-                        parent_cid=parent_cid,
-                        type_tag=type_tag,
-                        blob=blob,
-                        target_id=target_id,
-                        serde=self.serde,
-                        shared_cpid_chain=shared_cpid_chain,
-                        walk_state=walk_state,
-                        max_supersteps=max_supersteps,
-                        needed_depths=needed_depths,
-                        seed_values_by_depth=seed_values_by_depth,
-                        channels=channels,
-                    ):
-                        break
-
-            (
-                chained_cpid_by_ch,
-                seed_cpid_by_ch,
-                seed_value_by_ch,
-            ) = resolve_delta_chains(
-                channels=channels,
-                supersteps_by_ch=supersteps_by_ch,
-                shared_cpid_chain=shared_cpid_chain,
-                seed_values_by_depth=seed_values_by_depth,
-                has_reached_root=bool(walk_state.get("reached_root")),
-                thread_id=thread_id,
+            await cur.execute(
+                DELTA_STAGE1_SQL, (thread_id, checkpoint_ns, checkpoint_id)
             )
+            async for row in cur:
+                cid, parent_cid, type_tag, blob = row
+                if step_walk_with_row(
+                    cid=cid,
+                    parent_cid=parent_cid,
+                    type_tag=type_tag,
+                    blob=blob,
+                    target_id=checkpoint_id,
+                    serde=self.serde,
+                    chain_by_ch=chain_by_ch,
+                    seed_val_by_ch=seed_val_by_ch,
+                    walk_state=walk_state,
+                    seeded=seeded,
+                    channels=channels,
+                ):
+                    break
 
-            channels_with_chain = [ch for ch in channels if chained_cpid_by_ch[ch]]
-            fetch_sql = build_delta_writes_fetch_sql(
-                chain_lens=[len(chained_cpid_by_ch[ch]) for ch in channels_with_chain],
+            channels_with_chain = [ch for ch in channels if chain_by_ch[ch]]
+            stage2_sql = build_delta_stage2_sql(
+                chain_lens=[len(chain_by_ch[ch]) for ch in channels_with_chain],
             )
-            if fetch_sql:
-                fetch_params: list[Any] = []
+            if stage2_sql:
+                stage2_params: list[Any] = []
                 for ch in channels_with_chain:
-                    fetch_params.extend(
-                        [thread_id, checkpoint_ns, ch, *chained_cpid_by_ch[ch]]
+                    stage2_params.extend(
+                        [thread_id, checkpoint_ns, ch, *chain_by_ch[ch]]
                     )
-                await cur.execute(fetch_sql, fetch_params)
+                await cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
                     "list[tuple[str, str, str, int, str, bytes]]",
                     await cur.fetchall(),
@@ -728,9 +689,9 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):
 
         return build_delta_channels_writes_history(
             channels=channels,
-            chained_cpid_by_ch=chained_cpid_by_ch,
-            seed_cpid_by_ch=seed_cpid_by_ch,
-            seed_value_by_ch=seed_value_by_ch,
+            chain_by_ch=chain_by_ch,
+            seed_val_by_ch=seed_val_by_ch,
+            seeded=seeded,
             stage2_rows=stage2_rows,
             serde=self.serde,
         )

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

@@ -34,37 +34,6 @@ PendingWrite = tuple[str, str, Any]
 logger = logging.getLogger(__name__)
 
 
-def _parse_supersteps_since_last_snapshot_by_channel(
-    metadata: Mapping[str, Any],
-    channels: Sequence[str],
-) -> dict[str, int]:
-    """Per-channel supersteps-since-last-snapshot, parsed from metadata.
-
-    Reads `metadata.counters_since_delta_snapshot[ch]`, a `(updates,
-    supersteps)` pair, and returns `{ch: supersteps}` for channels whose
-    supersteps count is positive. Channels with no counter entry (just
-    snapshotted, or never written) are omitted — their seed, if any, is the
-    target checkpoint itself and needs no ancestor walk.
-
-    Used to drive the ancestor-walk depth in `get_delta_channel_history`:
-    a channel's seed snapshot sits exactly `supersteps` hops back along the
-    parent chain.
-    """
-    counters = metadata.get("counters_since_delta_snapshot") or {}
-    result: dict[str, int] = {}
-    for ch in channels:
-        entry = counters.get(ch)
-        if not isinstance(entry, (list, tuple)) or len(entry) < 2:
-            continue
-        try:
-            supersteps = int(entry[1])
-        except (TypeError, ValueError):
-            continue
-        if supersteps > 0:
-            result[ch] = supersteps
-    return result
-
-
 # Marked as total=False to allow for future expansion.
 class CheckpointMetadata(TypedDict, total=False):
     """Metadata associated with a checkpoint."""
@@ -650,30 +619,34 @@ class BaseCheckpointSaver(Generic[V]):
         """
         if not channels:
             return {}
-        channels = list(channels)
+        collected_by_ch: dict[str, list[PendingWrite]] = {c: [] for c in channels}
+        seed_by_ch: dict[str, Any] = {}
+        remaining: set[str] = set(channels)
         target_tuple = self.get_tuple(config)
-        if target_tuple is None:
-            return {ch: {"writes": []} for ch in channels}
-        supersteps_by_ch = _parse_supersteps_since_last_snapshot_by_channel(
-            target_tuple.metadata or {}, channels
+        cursor_config: RunnableConfig | None = (
+            target_tuple.parent_config if target_tuple else None
         )
-        max_supersteps = max(supersteps_by_ch.values(), default=0)
-
-        chain: list[CheckpointTuple] = []
-        has_reached_root = False
-        cursor_config: RunnableConfig | None = target_tuple.parent_config
-        while cursor_config is not None and len(chain) < max_supersteps:
+        while cursor_config is not None and remaining:
             tup = self.get_tuple(cursor_config)
             if tup is None:
                 break
-            chain.append(tup)
-            if tup.parent_config is None:
-                has_reached_root = True
-                break
+            if tup.pending_writes:
+                for write in reversed(tup.pending_writes):
+                    ch = write[1]
+                    if ch in remaining:
+                        collected_by_ch[ch].append(write)
+            for ch in list(remaining):
+                if ch in tup.checkpoint["channel_values"]:
+                    seed_by_ch[ch] = tup.checkpoint["channel_values"][ch]
+                    remaining.discard(ch)
             cursor_config = tup.parent_config
-        return self._assemble_default_delta_history(
-            channels, supersteps_by_ch, chain, has_reached_root, config
-        )
+        result: dict[str, DeltaChannelHistory] = {}
+        for ch in channels:
+            entry: DeltaChannelHistory = {"writes": list(reversed(collected_by_ch[ch]))}
+            if ch in seed_by_ch:
+                entry["seed"] = seed_by_ch[ch]
+            result[ch] = entry
+        return result
 
     async def aget_delta_channel_history(
         self, *, config: RunnableConfig, channels: Sequence[str]
@@ -687,106 +660,32 @@ class BaseCheckpointSaver(Generic[V]):
         """
         if not channels:
             return {}
-        channels = list(channels)
+        collected_by_ch: dict[str, list[PendingWrite]] = {c: [] for c in channels}
+        seed_by_ch: dict[str, Any] = {}
+        remaining: set[str] = set(channels)
         target_tuple = await self.aget_tuple(config)
-        if target_tuple is None:
-            return {ch: {"writes": []} for ch in channels}
-        supersteps_by_ch = _parse_supersteps_since_last_snapshot_by_channel(
-            target_tuple.metadata or {}, channels
+        cursor_config: RunnableConfig | None = (
+            target_tuple.parent_config if target_tuple else None
         )
-        max_supersteps = max(supersteps_by_ch.values(), default=0)
-
-        chain: list[CheckpointTuple] = []
-        has_reached_root = False
-        cursor_config: RunnableConfig | None = target_tuple.parent_config
-        while cursor_config is not None and len(chain) < max_supersteps:
+        while cursor_config is not None and remaining:
             tup = await self.aget_tuple(cursor_config)
             if tup is None:
                 break
-            chain.append(tup)
-            if tup.parent_config is None:
-                has_reached_root = True
-                break
+            if tup.pending_writes:
+                for write in reversed(tup.pending_writes):
+                    ch = write[1]
+                    if ch in remaining:
+                        collected_by_ch[ch].append(write)
+            for ch in list(remaining):
+                if ch in tup.checkpoint["channel_values"]:
+                    seed_by_ch[ch] = tup.checkpoint["channel_values"][ch]
+                    remaining.discard(ch)
             cursor_config = tup.parent_config
-        return self._assemble_default_delta_history(
-            channels, supersteps_by_ch, chain, has_reached_root, config
-        )
-
-    def _assemble_default_delta_history(
-        self,
-        channels: Sequence[str],
-        supersteps_by_ch: Mapping[str, int],
-        chain: Sequence[CheckpointTuple],
-        has_reached_root: bool,
-        config: RunnableConfig,
-    ) -> dict[str, DeltaChannelHistory]:
-        """Slice the walked parent chain into per-channel histories.
-
-        For each channel the seed snapshot sits `supersteps` hops back along
-        `chain` (newest first): the seed checkpoint is `chain[supersteps - 1]`
-        and its `channel_values[ch]` is the seed value. The seed checkpoint's
-        own writes are replayed on top of a `_DeltaSnapshot` seed but skipped
-        for a migrated plain-value seed (the legacy non-delta blob already
-        incorporates them).
-
-        When the chain is shorter than `supersteps` but the root was reached,
-        the persisted chain is "compressed" relative to the logical superstep
-        count (`durability="exit"`, or a thread that never snapshotted). The
-        seed candidate is then the oldest checkpoint (`chain[-1]`); if its
-        `channel_values[ch]` is absent the channel has no seed and replays
-        the full chain on an empty baseline.
-        """
-        # Imported lazily to avoid a hard checkpoint→serde-types coupling at
-        # module import; only the delta surface needs the runtime check.
-        from langgraph.checkpoint.serde.types import _DeltaSnapshot
-
-        thread_id = config["configurable"].get("thread_id")
         result: dict[str, DeltaChannelHistory] = {}
         for ch in channels:
-            entry: DeltaChannelHistory = {"writes": []}
-            bound = supersteps_by_ch.get(ch, 0)
-            if bound <= 0:
-                result[ch] = entry
-                continue
-            if len(chain) >= bound:
-                seed_depth = bound
-            elif has_reached_root:
-                seed_depth = len(chain)
-            else:
-                logger.warning(
-                    "cannot find seed snapshot for delta channel "
-                    "(thread_id=%s, channel=%s)",
-                    thread_id,
-                    ch,
-                )
-                result[ch] = entry
-                continue
-            if seed_depth <= 0:
-                result[ch] = entry
-                continue
-            chain_slice: Sequence[CheckpointTuple] = chain[:seed_depth]
-            seed_tuple: CheckpointTuple | None = chain[seed_depth - 1]
-
-            skip_seed_checkpoint_writes = False
-            channel_values = seed_tuple.checkpoint["channel_values"]
-            if ch in channel_values:
-                seed_value = channel_values[ch]
-                entry["seed"] = seed_value
-                skip_seed_checkpoint_writes = not isinstance(seed_value, _DeltaSnapshot)
-            else:
-                # No stored value at the oldest checkpoint → empty baseline,
-                # so the chain's own writes are all replayed from empty.
-                seed_tuple = None
-
-            collected: list[PendingWrite] = []
-            # Chain is newest→oldest; replay oldest→newest.
-            for tup in reversed(list(chain_slice)):
-                if skip_seed_checkpoint_writes and tup is seed_tuple:
-                    continue
-                for write in tup.pending_writes or []:
-                    if write[1] == ch:
-                        collected.append(write)
-            entry["writes"] = collected
+            entry: DeltaChannelHistory = {"writes": list(reversed(collected_by_ch[ch]))}
+            if ch in seed_by_ch:
+                entry["seed"] = seed_by_ch[ch]
             result[ch] = entry
         return result
 

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

@@ -23,7 +23,6 @@ from langgraph.checkpoint.base import (
     DeltaChannelHistory,
     PendingWrite,
     SerializerProtocol,
-    _parse_supersteps_since_last_snapshot_by_channel,
     get_checkpoint_id,
     get_checkpoint_metadata,
 )
@@ -145,20 +144,14 @@ class InMemorySaver(
     ) -> Mapping[str, DeltaChannelHistory]:
         """Override: walk the parent chain ONCE for all requested channels.
 
-        Walk depth is driven by the target checkpoint's
-        `counters_since_delta_snapshot[ch]` supersteps counter: each
-        channel's seed snapshot sits exactly `supersteps` hops back along
-        the parent chain. This locates seeds reliably even when they aren't
-        a `_DeltaSnapshot` sentinel — e.g. a legacy plain-value blob from a
-        thread migrated off a non-delta channel.
+        Each channel terminates independently at the nearest ancestor
+        whose stored blob is non-empty. Other channels keep walking until
+        they find their own terminator or hit the root.
 
-        Pre-delta plain-value seeds subsume their own checkpoint's pending
-        writes (the value already includes them), so those are skipped;
-        `_DeltaSnapshot` seeds do not (the snapshot is the value AT that
-        ancestor, prior to its own pending writes), so they are replayed.
-        When the chain reaches the root short of `supersteps`, the snapshot
-        was never persisted (implicit empty baseline) — replay the full
-        chain with no seed.
+        Pre-delta plain-value blobs subsume their ancestor's pending
+        writes (the value already includes them); `_DeltaSnapshot` blobs
+        do not (snapshot is the value AT that ancestor, prior to its own
+        pending writes that produce the child).
         """
         if not channels:
             return {}
@@ -166,98 +159,72 @@ class InMemorySaver(
         # module import; only this override needs the runtime check.
         from langgraph.checkpoint.serde.types import _DeltaSnapshot
 
-        channels = list(channels)
         thread_id = config["configurable"]["thread_id"]
         checkpoint_ns = config["configurable"].get("checkpoint_ns", "")
+        checkpoint_id = config["configurable"].get("checkpoint_id", "")
         ns_storage = self.storage.get(thread_id, {}).get(checkpoint_ns, {})
 
-        # Resolve the target checkpoint id + its metadata (for supersteps).
-        checkpoint_id = get_checkpoint_id(config)
-        if checkpoint_id is None:
-            checkpoint_id = next(reversed(ns_storage), None)
-        target_entry = ns_storage.get(checkpoint_id) if checkpoint_id else None
-        if target_entry is None:
-            return {ch: {"writes": []} for ch in channels}
-        metadata = self.serde.loads_typed(target_entry[1])
-        supersteps_by_ch = _parse_supersteps_since_last_snapshot_by_channel(
-            metadata or {}, channels
-        )
-        max_supersteps = max(supersteps_by_ch.values(), default=0)
-
-        # Walk the parent chain (newest first) up to the deepest supersteps.
         chain: list[str] = []
-        has_reached_root = False
-        current: str | None = target_entry[2]
-        while current is not None and len(chain) < max_supersteps:
+        target_entry = ns_storage.get(checkpoint_id)
+        current: str | None = target_entry[2] if target_entry is not None else None
+        while current is not None:
             entry = ns_storage.get(current)
             if entry is None:
                 break
             chain.append(current)
-            parent = entry[2]
-            if parent is None:
-                has_reached_root = True
-                break
+            _, _, parent = entry
             current = parent
 
+        collected_by_ch: dict[str, list[PendingWrite]] = {c: [] for c in channels}
+        seed_by_ch: dict[str, Any] = {}
+        remaining: set[str] = set(channels)
+
+        for cp_id in chain:
+            if not remaining:
+                break
+            entry = ns_storage.get(cp_id)
+            ckpt = self.serde.loads_typed(entry[0]) if entry is not None else None
+
+            terminated_here: set[str] = set()
+            blob_value_by_ch: dict[str, Any] = {}
+            if ckpt is not None:
+                versions = ckpt.get("channel_versions", {})
+                for ch in remaining:
+                    ver = versions.get(ch)
+                    if ver is None:
+                        continue
+                    blob_entry = self.blobs.get((thread_id, checkpoint_ns, ch, ver))
+                    if blob_entry is None or blob_entry[0] == "empty":
+                        continue
+                    blob_value_by_ch[ch] = self.serde.loads_typed(blob_entry)
+                    terminated_here.add(ch)
+
+            step_writes = self.writes.get((thread_id, checkpoint_ns, cp_id), {})
+            for (_task_id, _idx), (tid, ch, serialized, _) in sorted(
+                step_writes.items(), reverse=True
+            ):
+                if ch not in remaining:
+                    continue
+                blob_value = blob_value_by_ch.get(ch)
+                if blob_value is not None and not isinstance(
+                    blob_value, _DeltaSnapshot
+                ):
+                    continue
+                collected_by_ch[ch].append(
+                    (tid, ch, self.serde.loads_typed(serialized))
+                )
+
+            for ch in terminated_here:
+                seed_by_ch[ch] = blob_value_by_ch[ch]
+                remaining.discard(ch)
+
         result: dict[str, DeltaChannelHistory] = {}
         for ch in channels:
-            entry_h: DeltaChannelHistory = {"writes": []}
-            bound = supersteps_by_ch.get(ch, 0)
-            if bound <= 0:
-                result[ch] = entry_h
-                continue
-            if len(chain) >= bound:
-                seed_depth = bound
-            elif has_reached_root:
-                seed_depth = len(chain)
-            else:
-                logger.warning(
-                    "cannot find seed snapshot for delta channel "
-                    "(thread_id=%s, channel=%s)",
-                    thread_id,
-                    ch,
-                )
-                result[ch] = entry_h
-                continue
-            if seed_depth <= 0:
-                result[ch] = entry_h
-                continue
-            chain_slice: list[str] = chain[:seed_depth]
-            seed_cpid: str | None = chain[seed_depth - 1]
-
-            skip_seed_checkpoint_writes = False
-            seed_entry = ns_storage.get(seed_cpid)
-            if seed_entry is not None:
-                ckpt = self.serde.loads_typed(seed_entry[0])
-                ver = (ckpt.get("channel_versions") or {}).get(ch)
-                blob_entry = (
-                    self.blobs.get((thread_id, checkpoint_ns, ch, ver))
-                    if ver is not None
-                    else None
-                )
-                if blob_entry is not None and blob_entry[0] != "empty":
-                    seed_value = self.serde.loads_typed(blob_entry)
-                    entry_h["seed"] = seed_value
-                    skip_seed_checkpoint_writes = not isinstance(
-                        seed_value, _DeltaSnapshot
-                    )
-                else:
-                    # No stored value at the oldest checkpoint → empty
-                    # baseline; the chain's own writes replay from empty.
-                    seed_cpid = None
-
-            collected: list[PendingWrite] = []
-            # Chain is newest→oldest; replay oldest→newest.
-            for cp_id in reversed(chain_slice):
-                if skip_seed_checkpoint_writes and cp_id == seed_cpid:
-                    continue
-                step_writes = self.writes.get((thread_id, checkpoint_ns, cp_id), {})
-                for (_task_id, _idx), (tid, w_ch, serialized, _) in sorted(
-                    step_writes.items()
-                ):
-                    if w_ch == ch:
-                        collected.append((tid, ch, self.serde.loads_typed(serialized)))
-            entry_h["writes"] = collected
+            entry_h: DeltaChannelHistory = {
+                "writes": list(reversed(collected_by_ch[ch]))
+            }
+            if ch in seed_by_ch:
+                entry_h["seed"] = seed_by_ch[ch]
             result[ch] = entry_h
         return result
 

libs/checkpoint/tests/test_memory.py

@@ -351,12 +351,9 @@ class TestInMemorySaverDeltaChannel:
         cp1["id"] = "cp1"
         cp2 = empty_checkpoint()
         cp2["id"] = "cp2"
-        # Target (cp2) carries the supersteps counter; no snapshot was ever
-        # taken, so the walk runs back to the root → no seed, empty baseline.
-        cp2_md = {"counters_since_delta_snapshot": {channel: [1, 2]}}
         saver.storage[thread_id][ns] = {
             "cp1": (serde.dumps_typed(cp1), serde.dumps_typed({}), None),
-            "cp2": (serde.dumps_typed(cp2), serde.dumps_typed(cp2_md), "cp1"),
+            "cp2": (serde.dumps_typed(cp2), serde.dumps_typed({}), "cp1"),
         }
         # Writes stored at cp1 produced the cp1 snapshot; part of history.
         saver.writes[(thread_id, ns, "cp1")][("task1", 0)] = (
@@ -459,15 +456,10 @@ class TestBaseFallbackGetChannelWrites:
         cp1["id"] = "00000000000000000000000000000002.0000000000000000"
         cp2 = empty_checkpoint()
         cp2["id"] = "00000000000000000000000000000003.0000000000000000"
-        # The target (cp2) carries `counters_since_delta_snapshot` — the
-        # supersteps-since-last-snapshot driving the ancestor-walk depth. No
-        # snapshot was ever taken, so the count runs back past the oldest
-        # persisted checkpoint (root) → implicit empty baseline, no seed.
-        cp2_md = {"counters_since_delta_snapshot": {channel: [2, 3]}}
         saver.storage[thread_id][ns] = {
             cp0["id"]: (serde.dumps_typed(cp0), serde.dumps_typed({}), None),
             cp1["id"]: (serde.dumps_typed(cp1), serde.dumps_typed({}), cp0["id"]),
-            cp2["id"]: (serde.dumps_typed(cp2), serde.dumps_typed(cp2_md), cp1["id"]),
+            cp2["id"]: (serde.dumps_typed(cp2), serde.dumps_typed({}), cp1["id"]),
         }
         # Writes under cp0 produced cp1's state; writes under cp1 produced cp2's.
         saver.writes[(thread_id, ns, cp0["id"])][("task1", 0)] = (
@@ -615,14 +607,10 @@ class TestPreDeltaBlobTerminator:
         cp3["id"] = "cp3"
         cp3["channel_versions"][channel] = v3
 
-        # cp1 is the pre-delta (migration) seed; the channel became a
-        # DeltaChannel at cp2, so its supersteps counter starts there: cp2 -> 1,
-        # cp3 -> 2. The target (cp3) walks 2 hops back to the cp1 seed blob.
-        cp3_md = {"counters_since_delta_snapshot": {channel: [2, 2]}}
         saver.storage[thread_id][ns] = {
             "cp1": (serde.dumps_typed(cp1), serde.dumps_typed({}), None),
             "cp2": (serde.dumps_typed(cp2), serde.dumps_typed({}), "cp1"),
-            "cp3": (serde.dumps_typed(cp3), serde.dumps_typed(cp3_md), "cp2"),
+            "cp3": (serde.dumps_typed(cp3), serde.dumps_typed({}), "cp2"),
         }
         # Write under cp1 would be from the pre-delta era and MUST be ignored
         # (the blob already captures it). We add one and assert it is not

libs/langgraph/langgraph/channels/binop.py

@@ -29,11 +29,25 @@ def _strip_extras(t):  # type: ignore[no-untyped-def]
 
 
 def _get_overwrite(value: Any) -> tuple[bool, Any]:
-    """Inspects the given value and returns (is_overwrite, overwrite_value)."""
+    """Inspects the given value and returns (is_overwrite, overwrite_value).
+
+    Recognises three forms:
+
+    * The typed `Overwrite` dataclass instance.
+    * The sentinel-keyed `{"__overwrite__": value}` dict form.
+    * The dataclass-erased `{"value": ..., "type": "__overwrite__"}` form that
+      results from JSON-serialising an `Overwrite` (e.g. an `orjson`-encoded
+      state update routed through the LangGraph API server). This keeps the
+      `Overwrite` semantics intact across JSON boundaries that strip dataclass
+      types.
+    """
     if isinstance(value, Overwrite):
         return True, value.value
-    if isinstance(value, dict) and len(value) == 1 and OVERWRITE in value:
-        return True, value[OVERWRITE]
+    if isinstance(value, dict):
+        if len(value) == 1 and OVERWRITE in value:
+            return True, value[OVERWRITE]
+        if value.get("type") == OVERWRITE and "value" in value:
+            return True, value["value"]
     return False, None
 
 

libs/langgraph/langgraph/channels/delta.py

@@ -172,13 +172,11 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
                 overwrite_idx = i
         if overwrite_idx is not None:
             _, overwrite_value = _get_overwrite(values[overwrite_idx])
-            base = (
+            self.value = (
                 _copy.copy(overwrite_value)
                 if overwrite_value is not None
                 else self.typ()
             )
-            remaining = [v for i, v in enumerate(values) if i != overwrite_idx]
-            self.value = self.reducer(base, remaining) if remaining else base
             return True
         base = self.typ() if self.value is MISSING else self.value
         self.value = self.reducer(base, list(values))

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -70,42 +70,6 @@ def delta_channels_to_snapshot(
     return result
 
 
-def advance_delta_counters(
-    channels: Mapping[str, BaseChannel],
-    prev_metadata: Mapping[str, Any] | None,
-    updated_channels: set[str] | None,
-) -> tuple[dict[str, tuple[int, int]], set[str]]:
-    """Advance per-delta-channel `(updates, supersteps)` counters one step.
-
-    Mirrors the counter bookkeeping in `PregelLoop._put_checkpoint`: every
-    superstep bumps `supersteps` for all delta channels and `updates` for
-    those written this step, then channels that hit their snapshot cadence
-    are reset to `(0, 0)`.
-
-    Used by `update_state` (which creates a delta checkpoint outside the
-    main loop) so the written checkpoint carries a correct
-    `counters_since_delta_snapshot` — without it, the supersteps-based
-    `get_delta_channel_history` walk can't locate the seed. Returns
-    `(new_counters, channels_to_snapshot)`; callers drop `(0, 0)` entries
-    before writing the metadata field.
-    """
-    prev = dict((prev_metadata or {}).get("counters_since_delta_snapshot") or {})
-    updated = updated_channels or set()
-    new_counters: dict[str, tuple[int, int]] = {}
-    for name, ch in channels.items():
-        if not isinstance(ch, DeltaChannel):
-            continue
-        u, s = prev.get(name, (0, 0))
-        s += 1
-        if name in updated:
-            u += 1
-        new_counters[name] = (u, s)
-    channels_to_snapshot = delta_channels_to_snapshot(channels, new_counters)
-    for k in channels_to_snapshot:
-        new_counters[k] = (0, 0)
-    return new_counters, channels_to_snapshot
-
-
 def create_checkpoint(
     checkpoint: Checkpoint,
     channels: Mapping[str, BaseChannel] | None,

libs/langgraph/langgraph/pregel/_loop.py

@@ -70,6 +70,7 @@ from langgraph.callbacks import (
     GraphResumeEvent,
 )
 from langgraph.channels.base import BaseChannel
+from langgraph.channels.binop import _get_overwrite
 from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.untracked_value import UntrackedValue
 from langgraph.constants import TAG_HIDDEN
@@ -221,6 +222,11 @@ 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]
+
     # The checkpoint_config that points at the parent loaded at `__enter__`
     # (or the synthetic-empty checkpoint, on first run). We capture it
     # eagerly because every `_put_checkpoint` advances `self.checkpoint_config`
@@ -677,6 +683,11 @@ 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(
+            ch
+            for ch, v in writes
+            if isinstance(self.specs.get(ch), DeltaChannel) and _get_overwrite(v)[0]
+        )
         # all tasks have finished
         self.updated_channels = apply_writes(
             self.checkpoint,
@@ -980,6 +991,11 @@ class PregelLoop:
                 manager=None,
                 updated_channels=updated_channels,
             )
+            self._delta_channels_with_overwrite.update(
+                c
+                for c, v in input_writes
+                if isinstance(self.specs.get(c), DeltaChannel) and _get_overwrite(v)[0]
+            )
             # apply input writes
             updated_channels = apply_writes(
                 self.checkpoint,
@@ -1120,6 +1136,7 @@ class PregelLoop:
         # create new checkpoint
         channels_to_snapshot = (
             delta_channels_to_snapshot(self.channels, new_counters)
+            | self._delta_channels_with_overwrite
             if do_checkpoint
             else set()
         )
@@ -1136,6 +1153,8 @@ 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)
         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
@@ -1218,7 +1237,10 @@ class PregelLoop:
         counters = dict(
             self.checkpoint_metadata.get("counters_since_delta_snapshot") or {}
         )
-        channels_to_snapshot = delta_channels_to_snapshot(self.channels, counters)
+        channels_to_snapshot = (
+            delta_channels_to_snapshot(self.channels, counters)
+            | self._delta_channels_with_overwrite
+        )
 
         pending = [
             (step, tid, ch, v)
@@ -1662,6 +1684,7 @@ class SyncPregelLoop(PregelLoop, AbstractContextManager):
         )
         self._delta_write_futs = []
         self._error_handler_write_futs = []
+        self._delta_channels_with_overwrite = set()
         self._exit_delta_writes = (
             [] if self.durability == "exit" and self.checkpointer is not None else None
         )
@@ -1919,6 +1942,7 @@ class AsyncPregelLoop(PregelLoop, AbstractAsyncContextManager):
         )
         self._delta_write_futs = []
         self._error_handler_write_futs = []
+        self._delta_channels_with_overwrite = set()
         self._exit_delta_writes = (
             [] if self.durability == "exit" and self.checkpointer is not None else None
         )

libs/langgraph/langgraph/pregel/main.py

@@ -130,7 +130,6 @@ from langgraph.pregel._algo import (
 from langgraph.pregel._call import identifier
 from langgraph.pregel._checkpoint import (
     achannels_from_checkpoint,
-    advance_delta_counters,
     channels_from_checkpoint,
     copy_checkpoint,
     create_checkpoint,
@@ -2033,40 +2032,15 @@ class Pregel(
                 checkpointer.get_next_version,
                 self.trigger_to_nodes,
             )
-            # Advance delta-channel snapshot counters for this manual super-
-            # step so the written checkpoint carries a correct
-            # `counters_since_delta_snapshot` (mirrors the main loop). Without
-            # it, the supersteps-based delta history walk can't find the seed.
-            updated_channel_names = {
-                c for task in run_tasks for c, _ in task.writes if c != PUSH
-            }
-            new_counters, channels_to_snapshot = advance_delta_counters(
-                channels,
-                saved.metadata if saved else None,
-                updated_channel_names,
-            )
-            # Keep `updated_channels` at its default (None) — matching the
-            # historical checkpoint shape here — so resume/trigger semantics
-            # (e.g. deferred nodes) are unaffected. `get_next_version` /
-            # `channels_to_snapshot` are no-ops unless a delta channel snapshots.
-            checkpoint = create_checkpoint(
-                checkpoint,
-                channels,
-                step + 1,
-                get_next_version=checkpointer.get_next_version,
-                channels_to_snapshot=channels_to_snapshot,
-            )
-            update_metadata: dict[str, Any] = {
-                "source": "update",
-                "step": step + 1,
-                "parents": saved.metadata.get("parents", {}) if saved else {},
-            }
-            if non_zero := {k: v for k, v in new_counters.items() if v != (0, 0)}:
-                update_metadata["counters_since_delta_snapshot"] = non_zero
+            checkpoint = create_checkpoint(checkpoint, channels, step + 1)
             next_config = checkpointer.put(
                 checkpoint_config,
                 checkpoint,
-                update_metadata,
+                {
+                    "source": "update",
+                    "step": step + 1,
+                    "parents": saved.metadata.get("parents", {}) if saved else {},
+                },
                 get_new_channel_versions(
                     checkpoint_previous_versions, checkpoint["channel_versions"]
                 ),
@@ -2526,41 +2500,16 @@ class Pregel(
                 checkpointer.get_next_version,
                 self.trigger_to_nodes,
             )
-            # Advance delta-channel snapshot counters for this manual super-
-            # step so the written checkpoint carries a correct
-            # `counters_since_delta_snapshot` (mirrors the main loop). Without
-            # it, the supersteps-based delta history walk can't find the seed.
-            updated_channel_names = {
-                c for task in run_tasks for c, _ in task.writes if c != PUSH
-            }
-            new_counters, channels_to_snapshot = advance_delta_counters(
-                channels,
-                saved.metadata if saved else None,
-                updated_channel_names,
-            )
-            # Keep `updated_channels` at its default (None) — matching the
-            # historical checkpoint shape here — so resume/trigger semantics
-            # (e.g. deferred nodes) are unaffected. `get_next_version` /
-            # `channels_to_snapshot` are no-ops unless a delta channel snapshots.
-            checkpoint = create_checkpoint(
-                checkpoint,
-                channels,
-                step + 1,
-                get_next_version=checkpointer.get_next_version,
-                channels_to_snapshot=channels_to_snapshot,
-            )
-            update_metadata: dict[str, Any] = {
-                "source": "update",
-                "step": step + 1,
-                "parents": saved.metadata.get("parents", {}) if saved else {},
-            }
-            if non_zero := {k: v for k, v in new_counters.items() if v != (0, 0)}:
-                update_metadata["counters_since_delta_snapshot"] = non_zero
+            checkpoint = create_checkpoint(checkpoint, channels, step + 1)
             # save checkpoint, after applying writes
             next_config = await checkpointer.aput(
                 checkpoint_config,
                 checkpoint,
-                update_metadata,
+                {
+                    "source": "update",
+                    "step": step + 1,
+                    "parents": saved.metadata.get("parents", {}) if saved else {},
+                },
                 get_new_channel_versions(
                     checkpoint_previous_versions, checkpoint["channel_versions"]
                 ),

libs/langgraph/langgraph/types.py

@@ -976,3 +976,9 @@ class Overwrite:
 
     value: Any
     """The value to write directly to the channel, bypassing any reducer."""
+
+    type: Literal["__overwrite__"] = "__overwrite__"
+    """Discriminator field. Lets the channel reducer recognise an `Overwrite`
+    even after its dataclass form is JSON-serialised and the typed instance
+    is lost (e.g. an `orjson`-encoded state update routed through the
+    LangGraph API server)."""

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.2.6"
+version = "1.2.7"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"

libs/langgraph/tests/test_channels.py

@@ -186,6 +186,50 @@ def test_delta_channel_overwrite() -> None:
     assert ch.get()[0].content == "new"
 
 
+def test_overwrite_dataclass_form_survives_json_roundtrip() -> None:
+    """`Overwrite` serialised with `orjson` collapses to a plain dict but
+    must still be recognised as an overwrite by the channel reducer.
+
+    Without the `type` discriminator the dataclass-erased shape (`{"value":
+    ...}`) is indistinguishable from a literal channel value, and downstream
+    reducers raise `MESSAGE_COERCION_FAILURE` (or similar) on read.
+    """
+    import orjson
+
+    from langgraph._internal._constants import OVERWRITE
+    from langgraph.channels.binop import _get_overwrite
+
+    ow = Overwrite(value=[HumanMessage(content="new", id="h2")])
+    erased = orjson.loads(orjson.dumps(ow, default=lambda o: o.model_dump()))
+
+    assert erased["type"] == OVERWRITE
+    is_overwrite, value = _get_overwrite(erased)
+    assert is_overwrite
+    assert isinstance(value, list)
+    assert value[0]["content"] == "new"
+
+
+def test_overwrite_sentinel_dict_still_recognised() -> None:
+    """The pre-existing `{"__overwrite__": value}` dict form continues to be
+    recognised. This is the canonical sentinel emitted by producers that do
+    not have an `Overwrite` dataclass available."""
+    from langgraph._internal._constants import OVERWRITE
+    from langgraph.channels.binop import _get_overwrite
+
+    is_overwrite, value = _get_overwrite({OVERWRITE: ["b"]})
+    assert is_overwrite
+    assert value == ["b"]
+
+
+def test_overwrite_non_matching_dict_not_recognised() -> None:
+    """Dicts that resemble the erased shape but do not carry the
+    `__overwrite__` discriminator must not be misclassified as overwrites."""
+    from langgraph.channels.binop import _get_overwrite
+
+    assert _get_overwrite({"value": ["b"]}) == (False, None)
+    assert _get_overwrite({"type": "human", "value": "hi"}) == (False, None)
+
+
 def test_delta_channel_remove_message_and_replay() -> None:
     """RemoveMessage must round-trip correctly when writes are replayed."""
     spec = DeltaChannel(_messages_delta_reducer, list)
@@ -397,6 +441,93 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     assert len(state.values["messages"]) == 4  # 2 human + 2 AI
 
 
+def test_delta_channel_overwrite_superstep_snapshots() -> None:
+    def reducer(state: list[str], writes: Sequence[list[str]]) -> list[str]:
+        result = list(state)
+        for write in writes:
+            result.extend(write)
+        return result
+
+    class State(TypedDict):
+        items: Annotated[
+            list[str], DeltaChannel(reducer, list, snapshot_frequency=1000)
+        ]
+
+    def node_a(state: State) -> dict:
+        return {"items": ["a"]}
+
+    def node_b(state: State) -> dict:
+        return {"items": Overwrite(["b"])}
+
+    def node_c(state: State) -> dict:
+        return {"items": ["c"]}
+
+    builder = StateGraph(State)
+    builder.add_node("node_a", node_a)
+    builder.add_node("node_b", node_b)
+    builder.add_node("node_c", node_c)
+    builder.add_edge(START, "node_a")
+    builder.add_edge("node_a", "node_b")
+    builder.add_edge("node_a", "node_c")
+
+    saver = InMemorySaver()
+    graph = builder.compile(checkpointer=saver)
+    config = {"configurable": {"thread_id": "overwrite-snapshot"}}
+
+    result = graph.invoke({"items": ["START"]}, config)
+    assert result == {"items": ["b"]}
+
+    saved = saver.get_tuple(config)
+    assert saved is not None
+    snapshot = saved.checkpoint["channel_values"].get("items")
+    assert isinstance(snapshot, _DeltaSnapshot)
+    assert snapshot.value == ["b"]
+    assert saved.metadata.get("counters_since_delta_snapshot", {}).get("items") is None
+
+
+def test_delta_channel_replay_after_overwrite_snapshot() -> None:
+    def reducer(state: list[str], writes: Sequence[list[str]]) -> list[str]:
+        result = list(state)
+        for write in writes:
+            result.extend(write)
+        return result
+
+    class State(TypedDict):
+        items: Annotated[
+            list[str], DeltaChannel(reducer, list, snapshot_frequency=1000)
+        ]
+
+    calls = 0
+
+    def node(state: State) -> dict:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return {"items": Overwrite(["reset"])}
+        return {"items": ["after"]}
+
+    builder = StateGraph(State)
+    builder.add_node("node", node)
+    builder.add_edge(START, "node")
+
+    saver = InMemorySaver()
+    graph = builder.compile(checkpointer=saver)
+    config = {"configurable": {"thread_id": "overwrite-replay"}}
+
+    assert graph.invoke({"items": ["before"]}, config) == {"items": ["reset"]}
+    first_saved = saver.get_tuple(config)
+    assert first_saved is not None
+    assert isinstance(
+        first_saved.checkpoint["channel_values"].get("items"), _DeltaSnapshot
+    )
+
+    assert graph.invoke({"items": []}, config) == {"items": ["reset", "after"]}
+    second_saved = saver.get_tuple(config)
+    assert second_saved is not None
+    assert "items" not in second_saved.checkpoint["channel_values"]
+    assert graph.get_state(config).values == {"items": ["reset", "after"]}
+
+
 # ---------------------------------------------------------------------------
 # DeltaChannel — dict reducer
 # ---------------------------------------------------------------------------

libs/langgraph/uv.lock

@@ -1438,7 +1438,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.6"
+version = "1.2.7"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/uv.lock

@@ -285,7 +285,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.6"
+version = "1.2.7"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/uv.lock

@@ -298,7 +298,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.6"
+version = "1.2.7"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },