refactor(langgraph): call create_checkpoint directly for the fork seal

create_fork_checkpoint only forwarded to create_checkpoint, and an empty fork set bumps nothing there either, so the four update_state call sites pass the set straight through. The fork tests also had two identical helpers; keep one.
revert(langgraph): leave exit-mode resume writes as they are on main
2026-10-01 05:55:14 +02:00 · 2026-09-30 15:07:17 -04:00 · 2026-09-29 11:39:36 -04:00 · 2026-09-29 09:42:55 -04:00 · 2026-09-28 19:54:23 -04:00 · 2026-09-28 19:54:23 -04:00
25 changed files with 868 additions and 539 deletions
@@ -139,10 +139,23 @@ jobs:
            echo EOF
          } >> "$GITHUB_OUTPUT"

+  test-pypi-publish:
+    needs:
+      - build
+      - release-notes
+    permissions:
+      contents: read
+      id-token: write
+    uses: ./.github/workflows/_test_release.yml
+    with:
+      working-directory: ${{ inputs.working-directory }}
+    secrets: inherit
+
  pre-release-checks:
    needs:
      - build
      - release-notes
+      - test-pypi-publish
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
@@ -167,20 +180,31 @@ jobs:
          enable-cache: false
          working-directory: ${{ inputs.working-directory }}

-      - uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
-        with:
-          name: dist
-          path: ${{ inputs.working-directory }}/dist/
-
-      - name: Import dist package
+      - name: Import published package
        shell: bash
        working-directory: ${{ inputs.working-directory }}
        env:
          PKG_NAME: ${{ needs.build.outputs.pkg-name }}
          VERSION: ${{ needs.build.outputs.version }}
-        # Install directly from the locally-built wheel (no index resolution needed).
+        # Here we use:
+        # - The default regular PyPI index as the *primary* index, meaning
+        #   that it takes priority (https://pypi.org/simple)
+        # - The test PyPI index as an extra index, so that any dependencies that
+        #   are not found on test PyPI can be resolved and installed anyway.
+        #   (https://test.pypi.org/simple). This will include the PKG_NAME==VERSION
+        #   package because VERSION will not have been uploaded to regular PyPI yet.
+        # - attempt install again after 5 seconds if it fails because there is
+        #   sometimes a delay in availability on test pypi
        run: |
-          uv run pip install dist/*.whl
+          uv run pip install \
+            --extra-index-url https://test.pypi.org/simple/ \
+            "$PKG_NAME==$VERSION" || \
+          ( \
+            sleep 5 && \
+            uv run pip install \
+              --extra-index-url https://test.pypi.org/simple/ \
+              "$PKG_NAME==$VERSION" \
+          )

          if [[ "$PKG_NAME" == *prebuilt* ]]; then
            uv run pip install langgraph
@@ -202,7 +226,7 @@ jobs:
        run: uv sync --group test
        working-directory: ${{ inputs.working-directory }}

-      # Overwrite the local version of the package with the built version
+      # Overwrite the local version of the package with the test PyPI version.
      - name: Import published package (again)
        working-directory: ${{ inputs.working-directory }}
        shell: bash
@@ -210,25 +234,14 @@ jobs:
          PKG_NAME: ${{ needs.build.outputs.pkg-name }}
          VERSION: ${{ needs.build.outputs.version }}
        run: |
-          uv run pip install dist/*.whl
+          uv run pip install \
+            --extra-index-url https://test.pypi.org/simple/ \
+            "$PKG_NAME==$VERSION"

      - name: Run unit tests
        run: make test
        working-directory: ${{ inputs.working-directory }}

-  test-pypi-publish:
-    needs:
-      - build
-      - release-notes
-      - pre-release-checks
-    permissions:
-      contents: read
-      id-token: write
-    uses: ./.github/workflows/_test_release.yml
-    with:
-      working-directory: ${{ inputs.working-directory }}
-    secrets: inherit
-
  publish:
    needs:
      - build
@@ -943,11 +943,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.8.0"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
 ]

 [[package]]
@@ -448,12 +448,11 @@ class PostgresSaver(BasePostgresSaver):

        Two-stage query, both stages cover ALL requested channels:

-        * Stage 1 (paged): dynamic SELECT over `checkpoints` with three
-          columns per channel: its version, an `EXISTS` probe for a stored
-          blob at that version, and its inline value. 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 a page comes
-          back short.
+        * 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.

        * Stage 2 (per-channel UNION ALL): one branch per channel reading
          `checkpoint_writes` filtered to that channel's specific
@@ -172,8 +172,30 @@ class _DeltaStage2Row(TypedDict, total=False):
    version: str | None  # "b" rows only


-# Delta history is rebuilt in two queries; `_build_delta_stage1_sql` and
-# `_build_delta_stage2_sql` document their shapes.
+# Multi-channel two-stage DeltaChannel reconstruction.
+#
+# 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.
+#
+# 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.
+#
+# 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):
+#
+#   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)
+#
+# 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_stage1_sql(channels: Sequence[str], *, paged: bool) -> str:
@@ -313,8 +335,10 @@ def _build_delta_stage2_sql(
    return " UNION ALL ".join(branches)


-# Stage 1 rows are dicts keyed by the per-channel aliases
-# `_build_delta_stage1_sql` emits, so there is no static TypedDict.
+# 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]):
@@ -407,11 +431,9 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
          (a) it found a stored value for its channel — a blob or an inline
              primitive (channel becomes seeded),
          (b) it reached a real root (parent_of[cid] is None — fully
-              materialized at this point),
+              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), or
-          (d) the target's own row isn't in `parent_of` yet (the walk has
-              not started; no cursor is set, so a later page retries).
+              a later page; the cursor stays put).

        Mutates `chain_by_ch`, `seed_ver_by_ch`, `seed_inline_by_ch`,
        `walk_cursor_by_ch`, and `seeded` in place.
@@ -419,12 +441,9 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
        for i, ch in enumerate(channels):
            if ch in seeded:
                continue
-            # Pages start at the thread head, so the target may not have
-            # loaded yet; a `None` cursor would read as "target is a root".
+            # First-time entry: cursor starts at the target's parent.
            if ch not in walk_cursor_by_ch:
-                if target_id not in parent_of:
-                    continue
-                walk_cursor_by_ch[ch] = parent_of[target_id]
+                walk_cursor_by_ch[ch] = parent_of.get(target_id)
            cur_cid = walk_cursor_by_ch[ch]
            ch_chain = chain_by_ch[ch]
            hb_i = hb_by_i_by_cid[i]
@@ -1,132 +0,0 @@
-from __future__ import annotations
-
-from typing import Any
-from uuid import uuid4
-
-import pytest
-from langgraph.checkpoint.base import (
-    Checkpoint,
-    DeltaChannelHistory,
-    empty_checkpoint,
-)
-from langgraph.checkpoint.base.id import uuid6
-from langgraph.checkpoint.serde.types import _DeltaSnapshot
-
-from langgraph.checkpoint.postgres import PostgresSaver
-from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
-from langgraph.checkpoint.postgres.base import _DELTA_PAGE_SIZE
-from tests.conftest import DEFAULT_URI
-
-CHANNEL = "items"
-STEPS = 8
-SEED_STEP = 1
-SEED_VALUE = [10, 20]
-TARGET_STEP = 4
-
-# The real page size is the control; the rest leave the target off the first
-# page (three checkpoints are newer than it).
-PAGE_SIZES = [_DELTA_PAGE_SIZE, 3, 2, 1]
-
-
-def _step_args(
-    thread_id: str, step: int, parent: dict | None
-) -> tuple[dict, Checkpoint, dict[str, Any]]:
-    config: dict = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
-    if parent is not None:
-        config["configurable"]["checkpoint_id"] = parent["configurable"][
-            "checkpoint_id"
-        ]
-    checkpoint: Checkpoint = empty_checkpoint()
-    checkpoint["id"] = str(uuid6(clock_seq=step))
-    checkpoint["channel_versions"][CHANNEL] = f"v{step}"
-    if step == SEED_STEP:
-        checkpoint["channel_values"][CHANNEL] = _DeltaSnapshot(list(SEED_VALUE))
-        return config, checkpoint, {CHANNEL: f"v{step}"}
-    return config, checkpoint, {}
-
-
-async def _abuild_chain(saver: AsyncPostgresSaver) -> list[dict]:
-    thread_id = str(uuid4())
-    parent: dict | None = None
-    configs: list[dict] = []
-    for step in range(STEPS):
-        config, checkpoint, new_versions = _step_args(thread_id, step, parent)
-        parent = await saver.aput(
-            config,
-            checkpoint,
-            {"source": "loop", "step": step, "parents": {}},
-            new_versions,
-        )
-        await saver.aput_writes(parent, [(CHANNEL, f"w{step}")], str(uuid4()))
-        configs.append(parent)
-    return configs
-
-
-def _build_chain(saver: PostgresSaver) -> list[dict]:
-    thread_id = str(uuid4())
-    parent: dict | None = None
-    configs: list[dict] = []
-    for step in range(STEPS):
-        config, checkpoint, new_versions = _step_args(thread_id, step, parent)
-        parent = saver.put(
-            config,
-            checkpoint,
-            {"source": "loop", "step": step, "parents": {}},
-            new_versions,
-        )
-        saver.put_writes(parent, [(CHANNEL, f"w{step}")], str(uuid4()))
-        configs.append(parent)
-    return configs
-
-
-def _assert_history(entry: DeltaChannelHistory, page_size: int) -> None:
-    seed = entry.get("seed")
-    assert isinstance(seed, _DeltaSnapshot), (
-        f"page_size={page_size}: expected a snapshot seed, "
-        f"got {entry.get('seed', '<missing>')!r}"
-    )
-    assert seed.value == SEED_VALUE
-    assert [w[2] for w in entry["writes"]] == ["w1", "w2", "w3"], (
-        f"page_size={page_size}: got {[w[2] for w in entry['writes']]}"
-    )
-
-
-@pytest.mark.parametrize("page_size", PAGE_SIZES)
-async def test_async_target_older_than_the_first_page(
-    page_size: int, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.setattr("langgraph.checkpoint.postgres.aio._DELTA_PAGE_SIZE", page_size)
-    async with AsyncPostgresSaver.from_conn_string(DEFAULT_URI) as saver:
-        await saver.setup()
-        configs = await _abuild_chain(saver)
-        result = await saver.aget_delta_channel_history(
-            config=configs[TARGET_STEP], channels=[CHANNEL]
-        )
-        _assert_history(result[CHANNEL], page_size)
-
-
-@pytest.mark.parametrize("page_size", PAGE_SIZES)
-def test_sync_target_older_than_the_first_page(
-    page_size: int, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    monkeypatch.setattr("langgraph.checkpoint.postgres._DELTA_PAGE_SIZE", page_size)
-    with PostgresSaver.from_conn_string(DEFAULT_URI) as saver:
-        saver.setup()
-        configs = _build_chain(saver)
-        result = saver.get_delta_channel_history(
-            config=configs[TARGET_STEP], channels=[CHANNEL]
-        )
-        _assert_history(result[CHANNEL], page_size)
-
-
-async def test_root_target_has_no_history_and_still_terminates(
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    monkeypatch.setattr("langgraph.checkpoint.postgres.aio._DELTA_PAGE_SIZE", 1)
-    async with AsyncPostgresSaver.from_conn_string(DEFAULT_URI) as saver:
-        await saver.setup()
-        configs = await _abuild_chain(saver)
-        result = await saver.aget_delta_channel_history(
-            config=configs[0], channels=[CHANNEL]
-        )
-        assert result[CHANNEL] == {"writes": []}
@@ -1122,11 +1122,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.8.0"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
 ]

 [[package]]
@@ -507,12 +507,13 @@ class SqliteSaver(BaseCheckpointSaver[str]):

        Two-stage query:

-        * Stage 1 (streamed): recursive CTE over `checkpoints` following
-          `parent_checkpoint_id` from the target, returning
-          `(checkpoint_id, type, checkpoint)` per ancestor. Sqlite has no
-          JSONB, so we ship the full serialized checkpoint blob and inspect
-          `channel_values` in Python. Stops reading when every channel has
-          found its seed or the chain is exhausted.
+        * Stage 1 (paged): newest-first slice of `checkpoints` returning
+          `(checkpoint_id, parent_checkpoint_id, type, checkpoint)` per
+          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.

        * Stage 2 (per-channel UNION ALL): one branch per channel reading
          `writes` filtered to that channel's specific `chain_cids`. No
@@ -537,14 +538,12 @@ class SqliteSaver(BaseCheckpointSaver[str]):
        seeded: set[str] = set()

        with self.cursor(transaction=False) as cur:
-            cur.execute(
-                DELTA_STAGE1_SQL,
-                (thread_id, checkpoint_ns, checkpoint_id, thread_id, checkpoint_ns),
-            )
+            cur.execute(DELTA_STAGE1_SQL, (thread_id, checkpoint_ns, checkpoint_id))
            for row in cur:
-                cid, type_tag, blob = row
+                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,
@@ -26,33 +26,16 @@ from typing import Any

 from langgraph.checkpoint.base import DeltaChannelHistory, PendingWrite

-# Stage 1 streams target, then its ancestors nearest-first, by following
-# `parent_checkpoint_id` rather than id order: ids are only monotonic within
-# one process, so a range scan by id can miss a parent whose id sorts above
-# its child's. Target is the anchor row; its own writes/seed are skipped
-# (matches the `BaseCheckpointSaver` contract).
-#
-# `put` is `INSERT OR REPLACE`, so re-putting an existing id under a
-# descendant's config makes the chain a loop. `step_walk_with_row` stops on a
-# repeated id; sqlite yields recursive rows lazily, so abandoning the cursor
-# ends the recursion.
-#
-# `CROSS JOIN` pins `ancestors` as the outer loop, so each step is one primary
-# key lookup. With a plain `JOIN` and no `ANALYZE` stats, sqlite can put
-# `checkpoints` outside and scan the whole thread per step.
+# 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 row without a separate lookup;
+# the caller skips target's own writes/seed (matches the
+# `BaseCheckpointSaver` contract).
 DELTA_STAGE1_SQL = (
-    "WITH RECURSIVE ancestors(checkpoint_id, parent_checkpoint_id, type, "
-    "checkpoint) AS ("
    "SELECT checkpoint_id, parent_checkpoint_id, type, checkpoint "
    "FROM checkpoints "
-    "WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id = ? "
-    "UNION ALL "
-    "SELECT c.checkpoint_id, c.parent_checkpoint_id, c.type, c.checkpoint "
-    "FROM ancestors a CROSS JOIN checkpoints c "
-    "ON c.checkpoint_id = a.parent_checkpoint_id "
-    "WHERE c.thread_id = ? AND c.checkpoint_ns = ?"
-    ") "
-    "SELECT checkpoint_id, type, checkpoint FROM ancestors"
+    "WHERE thread_id = ? AND checkpoint_ns = ? AND checkpoint_id <= ? "
+    "ORDER BY checkpoint_id DESC"
 )


@@ -85,6 +68,7 @@ def build_delta_stage2_sql(*, chain_lens: Sequence[int]) -> str:
 def step_walk_with_row(
    *,
    cid: str,
+    parent_cid: str | None,
    type_tag: str,
    blob: bytes,
    target_id: str,
@@ -97,32 +81,36 @@ def step_walk_with_row(
 ) -> bool:
    """Process one streamed stage-1 row in the merged ancestor walk.

-    The cursor returns (cid, type, blob) rows in walk order starting at
-    target. The first row is target itself and is skipped (target's own
-    writes/seed are not part of the contract).
+    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 subsequent row we deserialize the blob, append the cid to
-    every not-yet-seeded channel's chain, and check `channel_values` for
+    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.

-    Returns True when the caller can stop iterating and close the cursor:
-    every requested channel is seeded, or the chain revisited a checkpoint.
+    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
            walk_state["active"] = {ch for ch in channels if ch not in seeded}
-            walk_state["walked"] = {cid}
        # Not target yet (or target not present): keep streaming.
        return False
    active: set[str] = walk_state["active"]
    if not active:
        return True
-    walked: set[str] = walk_state["walked"]
-    if cid in walked:
-        return True
-    walked.add(cid)
+    if cid != walk_state["cur_cid"]:
+        # Off-path row from a sibling branch — skip without deserializing.
+        return False
    for ch in active:
        chain_by_ch[ch].append(cid)
    ckpt = serde.loads_typed((type_tag, blob))
@@ -132,6 +120,7 @@ def step_walk_with_row(
        seeded.add(ch)
        active.discard(ch)
    del ckpt, channel_values
+    walk_state["cur_cid"] = parent_cid
    return not active


@@ -625,8 +625,8 @@ 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. Stage 1 streams
-        the parent chain from the target and Python-deserializes each
+        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.
        """
@@ -650,13 +650,13 @@ class AsyncSqliteSaver(BaseCheckpointSaver[str]):

        async with self.lock, self.conn.cursor() as cur:
            await cur.execute(
-                DELTA_STAGE1_SQL,
-                (thread_id, checkpoint_ns, checkpoint_id, thread_id, checkpoint_ns),
+                DELTA_STAGE1_SQL, (thread_id, checkpoint_ns, checkpoint_id)
            )
            async for row in cur:
-                cid, type_tag, blob = row
+                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,
@@ -1,104 +0,0 @@
-from __future__ import annotations
-
-from typing import Any
-
-import pytest
-from langgraph.checkpoint.base import (
-    BaseCheckpointSaver,
-    Checkpoint,
-    DeltaChannelHistory,
-    empty_checkpoint,
-)
-
-from langgraph.checkpoint.sqlite import SqliteSaver
-from langgraph.checkpoint.sqlite._delta import DELTA_STAGE1_SQL
-from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
-
-CHANNEL = "ch"
-CONFIG: dict[str, Any] = {"configurable": {"thread_id": "t", "checkpoint_ns": ""}}
-EXPECTED: DeltaChannelHistory = {
-    "writes": [("task", CHANNEL, "write-root")],
-    "seed": "seed",
-}
-
-
-def _checkpoint(checkpoint_id: str, values: dict[str, Any]) -> Checkpoint:
-    value = empty_checkpoint()
-    value["id"] = checkpoint_id
-    value["channel_values"] = values
-    return value
-
-
-PARENT_ID_ORDERS = [
-    pytest.param("z-older", "a-newer", id="parent_id_sorts_above_child"),
-    pytest.param("a-older", "z-newer", id="parent_id_sorts_below_child"),
-]
-
-
-@pytest.mark.parametrize(("root_id", "child_id"), PARENT_ID_ORDERS)
-def test_sync_walk_reaches_parent_whatever_the_id_order(
-    root_id: str, child_id: str
-) -> None:
-    with SqliteSaver.from_conn_string(":memory:") as saver:
-        root = saver.put(CONFIG, _checkpoint(root_id, {CHANNEL: "seed"}), {}, {})
-        saver.put_writes(root, [(CHANNEL, "write-root")], "task")
-        child = saver.put(root, _checkpoint(child_id, {}), {}, {})
-
-        got = saver.get_delta_channel_history(config=child, channels=[CHANNEL])
-        reference = BaseCheckpointSaver.get_delta_channel_history(
-            saver, config=child, channels=[CHANNEL]
-        )
-        assert got[CHANNEL] == EXPECTED
-        assert got[CHANNEL] == reference[CHANNEL], "fast path disagrees with base"
-
-
-@pytest.mark.parametrize(("root_id", "child_id"), PARENT_ID_ORDERS)
-async def test_async_walk_reaches_parent_whatever_the_id_order(
-    root_id: str, child_id: str
-) -> None:
-    async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:
-        root = await saver.aput(CONFIG, _checkpoint(root_id, {CHANNEL: "seed"}), {}, {})
-        await saver.aput_writes(root, [(CHANNEL, "write-root")], "task")
-        child = await saver.aput(root, _checkpoint(child_id, {}), {}, {})
-
-        got = await saver.aget_delta_channel_history(config=child, channels=[CHANNEL])
-        assert got[CHANNEL] == EXPECTED
-
-
-def test_walk_reaches_root_of_long_chain_with_descending_ids() -> None:
-    steps = 40
-    with SqliteSaver.from_conn_string(":memory:") as saver:
-        parent = saver.put(
-            CONFIG, _checkpoint(f"id-{steps:03d}", {CHANNEL: "seed"}), {}, {}
-        )
-        saver.put_writes(parent, [(CHANNEL, "write-root")], "task")
-        for step in range(steps - 1, 0, -1):
-            parent = saver.put(parent, _checkpoint(f"id-{step:03d}", {}), {}, {})
-
-        got = saver.get_delta_channel_history(config=parent, channels=[CHANNEL])
-        assert got[CHANNEL] == EXPECTED
-
-
-def test_walk_terminates_when_put_makes_the_parent_chain_cycle() -> None:
-    with SqliteSaver.from_conn_string(":memory:") as saver:
-        a = saver.put(CONFIG, _checkpoint("cid-a", {}), {}, {})
-        b = saver.put(a, _checkpoint("cid-b", {}), {}, {})
-        repoint_a_under_b = _checkpoint("cid-a", {})
-        saver.put(b, repoint_a_under_b, {}, {})
-
-        got = saver.get_delta_channel_history(config=b, channels=[CHANNEL])
-        assert got[CHANNEL] == {"writes": []}
-
-
-def test_walk_step_looks_up_the_parent_by_primary_key() -> None:
-    with SqliteSaver.from_conn_string(":memory:") as saver:
-        saver.setup()
-        plan = [
-            row[3]
-            for row in saver.conn.execute(
-                f"EXPLAIN QUERY PLAN {DELTA_STAGE1_SQL}", ("t", "", "id", "t", "")
-            )
-        ]
-    assert any(
-        step.startswith("SEARCH c ") and "checkpoint_id=?" in step for step in plan
-    ), f"recursive step should look up the parent by key, got {plan}"
@@ -1057,11 +1057,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.8.0"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
 ]

 [[package]]
@@ -1338,11 +1338,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.8.0"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
 ]

 [[package]]
@@ -732,11 +732,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.8.0"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
 ]

 [[package]]
@@ -672,11 +672,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.8.0"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
 ]

 [[package]]
@@ -1777,11 +1777,11 @@ wheels = [

 [[package]]
 name = "pyjwt"
-version = "2.15.0"
+version = "2.13.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/02/a5/5197bfd06417837ac079921c66fa6393f1dea3557272a263cebfef69e432/pyjwt-2.15.0.tar.gz", hash = "sha256:b11c5f9791d7bf51c2b39a81ed669f6b2dbbd669df2942f6c60167e9e3d1abe4", size = 120513, upload-time = "2026-09-23T16:56:00.689Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/3b/81/58d0ac84e1ef3a3843791d6954d94c0b33d526c75eeb1efbce9d0a4c4077/pyjwt-2.13.0.tar.gz", hash = "sha256:41571c89ca91598c79e8ef18a2d07367d4810fbbd6f637794879baf1b7703423", size = 107515, upload-time = "2026-05-21T19:54:36.618Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/e8/55/40e45bf052ee8ee12a4dfd785519660f8effa7b065442b91646ec6828619/pyjwt-2.15.0-py3-none-any.whl", hash = "sha256:7a3742debf6b879e912dbb9819ceec1594be812452b78c5f2e2dfc56564954f8", size = 33680, upload-time = "2026-09-23T16:55:59.241Z" },
+    { url = "https://files.pythonhosted.org/packages/a3/5e/ecf12fdb62546d64385c158514e9b2b671f7832108ef2ecd2020ce0af2d1/pyjwt-2.13.0-py3-none-any.whl", hash = "sha256:66adcc2aff09b3f1bbd95fc1e1577df8ac8723c978552fd43304c8a290ac5728", size = 31274, upload-time = "2026-05-21T19:54:35.362Z" },
 ]

 [[package]]
@@ -2243,11 +2243,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.8.0"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
 ]

 [[package]]
@@ -8,13 +8,15 @@ from typing import Any, cast
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
    BaseCheckpointSaver,
+    ChannelVersions,
    Checkpoint,
+    PendingWrite,
 )
 from langgraph.checkpoint.base.id import uuid6
 from langgraph.checkpoint.serde.types import _DeltaSnapshot

 from langgraph._internal._config import DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT
-from langgraph._internal._constants import PUSH
+from langgraph._internal._constants import INTERRUPT, PUSH
 from langgraph._internal._typing import MISSING
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.delta import DeltaChannel
@@ -89,6 +91,23 @@ def get_delta_channels_from_all_channels(
    }


+def delta_channels_with_pending_writes(
+    specs: Mapping[str, Any],
+    pending_writes: Iterable[PendingWrite] | None,
+) -> set[str]:
+    """DeltaChannels a branch starting from this checkpoint must snapshot.
+
+    A checkpoint's pending writes belong to the child that consumed them, and
+    nothing records which child that was. A new branch snapshots every delta
+    channel they touch, so its ancestor walk never replays them.
+    """
+    return {
+        ch
+        for _, ch, _ in pending_writes or ()
+        if isinstance(specs.get(ch), DeltaChannel)
+    }
+
+
 def create_metadata_for_update_state_api(
    channels: Mapping[str, BaseChannel],
    updated_channels: set[str],
@@ -114,26 +133,6 @@ def create_metadata_for_update_state_api(
    return new_counters


-def advance_delta_counters(
-    channels: Mapping[str, BaseChannel],
-    updated_channels: set[str],
-    *,
-    prev_metadata: Mapping[str, Any] | None,
-) -> dict[str, Any]:
-    """The `counters_since_delta_snapshot` entry for an update_state
-    checkpoint saved one superstep after `prev_metadata`'s, for the paths
-    that skip `create_checkpoint_plan_for_update_state_api`.
-
-    Without it, the next checkpoint restarts every delta channel's snapshot
-    cadence from zero.
-    """
-    counters = create_metadata_for_update_state_api(
-        channels, updated_channels, prev_metadata=prev_metadata
-    )
-    non_zero = {k: v for k, v in counters.items() if v != (0, 0)}
-    return {"counters_since_delta_snapshot": non_zero} if non_zero else {}
-
-
 def create_checkpoint_plan_for_update_state_api(
    channels: Mapping[str, BaseChannel],
    updated_channels: set[str],
@@ -142,6 +141,7 @@ def create_checkpoint_plan_for_update_state_api(
    parents: dict[str, Any],
    saved_metadata: Mapping[str, Any] | None,
    is_fresh_thread: bool,
+    fork_channels: set[str],
 ) -> tuple[set[str], dict[str, Any]]:
    """Return ``(channels_to_snapshot, metadata)`` for an update_state head."""
    metadata: dict[str, Any] = {
@@ -157,7 +157,9 @@ def create_checkpoint_plan_for_update_state_api(
        updated_channels,
        prev_metadata=saved_metadata,
    )
-    channels_to_snapshot = delta_channels_to_snapshot(channels, new_counters)
+    channels_to_snapshot = (
+        delta_channels_to_snapshot(channels, new_counters) | fork_channels
+    )
    for k in channels_to_snapshot:
        new_counters[k] = (0, 0)
    non_zero = {k: v for k, v in new_counters.items() if v != (0, 0)}
@@ -187,6 +189,7 @@ def create_checkpoint(
    """
    ts = datetime.now(timezone.utc).isoformat()
    channels_to_snapshot = channels_to_snapshot or set()
+    bumped: dict[str, tuple[Any, Any]] = {}
    if channels is None:
        values = checkpoint["channel_values"]
        channel_versions = checkpoint["channel_versions"]
@@ -194,30 +197,29 @@ def create_checkpoint(
        values = {}
        channel_versions = dict(checkpoint["channel_versions"])
        for k in channels:
-            if k not in channel_versions:
-                continue
            ch = channels[k]
+            if k not in channel_versions:
+                # A forced snapshot of a never-written channel still has to
+                # land to stop the ancestor walk, and `put` only stores blobs
+                # for versioned channels.
+                if k in channels_to_snapshot and get_next_version is not None:
+                    channel_versions[k] = get_next_version(None, None)
+                    bumped[k] = (None, channel_versions[k])
+                    values[k] = _DeltaSnapshot(
+                        ch.get() if ch.is_available() else ch.typ()
+                    )
+                continue
            if k in channels_to_snapshot:
-                # Callers force a full snapshot blob here: exit mode when a
-                # delta channel reaches its snapshot cadence, and update_state
-                # on a fresh thread (no ancestor to replay writes from). The
-                # manual version-bump below only applies to the exit-mode case.
-                #
-                # In exit mode, the snapshot decision is deferred to exit
-                # time (intermediate steps have do_checkpoint=False). The
-                # channel's count may have reached snapshot_frequency over
-                # several supersteps, but the LAST superstep may not have
-                # written to this channel. In that case apply_writes()
-                # (in _algo.py) didn't bump this channel's version, so
-                # saver.put() wouldn't include it in new_versions and
-                # the snapshot blob would be silently dropped. The manual
-                # bump below closes the gap. In sync/async durability this
-                # branch is effectively dead code (the step that pushes
-                # the count to freq always writes the channel).
+                # `put` only stores a blob for a channel whose version moved,
+                # so snapshotting a channel this step did not write needs a
+                # bump: exit mode reaching the cadence on a superstep that
+                # skipped the channel, and a fork's first checkpoint.
                if get_next_version is not None and (
                    updated_channels is None or k not in updated_channels
                ):
-                    channel_versions[k] = get_next_version(channel_versions[k], None)
+                    old = channel_versions[k]
+                    channel_versions[k] = get_next_version(old, None)
+                    bumped[k] = (old, channel_versions[k])
                values[k] = _DeltaSnapshot(ch.get())
            else:
                v = ch.checkpoint()
@@ -229,11 +231,30 @@ def create_checkpoint(
        id=id or str(uuid6(clock_seq=step)),
        channel_values=values,
        channel_versions=channel_versions,
-        versions_seen=checkpoint["versions_seen"],
+        versions_seen=_mark_bumps_seen(checkpoint["versions_seen"], bumped),
        updated_channels=None if updated_channels is None else sorted(updated_channels),
    )


+def _mark_bumps_seen(
+    versions_seen: dict[str, ChannelVersions],
+    bumped: Mapping[str, tuple[Any, Any]],
+) -> dict[str, ChannelVersions]:
+    """Advance whoever had seen a bumped channel's old version to the new one.
+
+    A bump that only stores a snapshot is not a write. Left unseen, it would
+    re-fire `interrupt_before` and rerun the channel's subscribers.
+    """
+    if not bumped:
+        return versions_seen
+    out: dict[str, ChannelVersions] = {}
+    for node, seen in {INTERRUPT: {}, **versions_seen}.items():
+        advanced = {k: new for k, (old, new) in bumped.items() if seen.get(k) == old}
+        if advanced or node in versions_seen:
+            out[node] = {**seen, **advanced}
+    return out
+
+
 def _needs_replay(spec: BaseChannel, stored: object) -> bool:
    """True if `spec` is a `DeltaChannel` and no value is stored at this
    checkpoint, requiring an ancestor walk to reconstruct.
@@ -102,6 +102,7 @@ from langgraph.pregel._checkpoint import (
    copy_checkpoint,
    create_checkpoint,
    delta_channels_to_snapshot,
+    delta_channels_with_pending_writes,
    empty_checkpoint,
    exit_delta_task_id,
 )
@@ -222,10 +223,13 @@ class PregelLoop:
    # under the saver's `ORDER BY task_id, idx` sorting.
    _exit_delta_writes: list[tuple[int, str, str, Any]] | None = None

-    # Delta channels that saw an Overwrite since the last checkpoint. These
-    # channels must snapshot after live update applies overwrite semantics so
-    # sparse replay starts from the same post-overwrite value.
-    _delta_channels_with_overwrite: set[str]
+    # Delta channels that must snapshot at the next checkpoint, whatever their
+    # cadence counters say:
+    # * an Overwrite arrived since the last checkpoint, so sparse replay has to
+    #   start from the post-overwrite value;
+    # * the checkpoint this run starts from has pending writes to them; see
+    #   `delta_channels_with_pending_writes`.
+    _delta_channels_forced_snapshot: set[str]

    # The checkpoint_config that points at the parent loaded at `__enter__`
    # (or the synthetic-empty checkpoint, on first run). We capture it
@@ -683,7 +687,7 @@ class PregelLoop:
    def after_tick(self) -> None:
        # finish superstep
        writes = [w for t in self.tasks.values() for w in t.writes]
-        self._delta_channels_with_overwrite.update(
+        self._delta_channels_forced_snapshot.update(
            ch
            for ch, v in writes
            if isinstance(self.specs.get(ch), DeltaChannel) and _get_overwrite(v)[0]
@@ -898,6 +902,15 @@ class PregelLoop:
            self.checkpoint_pending_writes = [
                w for w in self.checkpoint_pending_writes if w[1] != RESUME
            ]
+        # A resume that is not replaying reuses the head's pending writes
+        # instead of rerunning their tasks, so none of them can leak.
+        self._delta_channels_forced_snapshot = (
+            set()
+            if is_resuming and not self.is_replaying
+            else delta_channels_with_pending_writes(
+                self.specs, self.checkpoint_pending_writes
+            )
+        )

        # map command to writes
        if input_is_command:
@@ -991,7 +1004,7 @@ class PregelLoop:
                manager=None,
                updated_channels=updated_channels,
            )
-            self._delta_channels_with_overwrite.update(
+            self._delta_channels_forced_snapshot.update(
                c
                for c, v in input_writes
                if isinstance(self.specs.get(c), DeltaChannel) and _get_overwrite(v)[0]
@@ -1136,7 +1149,7 @@ class PregelLoop:
        # create new checkpoint
        channels_to_snapshot = (
            delta_channels_to_snapshot(self.channels, new_counters)
-            | self._delta_channels_with_overwrite
+            | self._delta_channels_forced_snapshot
            if do_checkpoint
            else set()
        )
@@ -1154,7 +1167,7 @@ class PregelLoop:
        for k in channels_to_snapshot:
            new_counters[k] = (0, 0)
        if do_checkpoint:
-            self._delta_channels_with_overwrite.difference_update(channels_to_snapshot)
+            self._delta_channels_forced_snapshot.difference_update(channels_to_snapshot)
        non_zero = {k: v for k, v in new_counters.items() if v != (0, 0)}
        if non_zero:
            self.checkpoint_metadata["counters_since_delta_snapshot"] = non_zero
@@ -1239,7 +1252,7 @@ class PregelLoop:
        )
        channels_to_snapshot = (
            delta_channels_to_snapshot(self.channels, counters)
-            | self._delta_channels_with_overwrite
+            | self._delta_channels_forced_snapshot
        )

        pending = [
@@ -1684,7 +1697,6 @@ 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
        )
@@ -1942,7 +1954,6 @@ 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
        )
@@ -129,11 +129,11 @@ 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,
    create_checkpoint_plan_for_update_state_api,
+    delta_channels_with_pending_writes,
    empty_checkpoint,
    get_updated_channels_from_tasks,
 )
@@ -1638,12 +1638,21 @@ class Pregel(
            else:
                raise ValueError(f"Subgraph {recast} not found")

+        # Taken from the first superstep's base, and cleared by the first
+        # checkpoint that carries the snapshots, which `__copy__` does not write.
+        fork_pending: set[str] | None = None
+
        def perform_superstep(
            input_config: RunnableConfig, updates: Sequence[StateUpdate]
        ) -> RunnableConfig:
+            nonlocal fork_pending
            # get last checkpoint
            config = ensure_config(self.config, input_config)
            saved = checkpointer.get_tuple(config)
+            if fork_pending is None:
+                fork_pending = delta_channels_with_pending_writes(
+                    self.channels, saved.pending_writes if saved else None
+                )
            if saved is not None:
                self._migrate_checkpoint(saved.checkpoint)
            checkpoint = (
@@ -1682,7 +1691,6 @@ class Pregel(
                        "Cannot apply multiple updates when clearing state"
                    )

-                updated_channels: set[str] = set()
                if saved is not None:
                    # tasks for this checkpoint
                    next_tasks = prepare_next_tasks(
@@ -1705,7 +1713,7 @@ class Pregel(
                        for w in saved.pending_writes or []
                        if w[0] == NULL_TASK_ID
                    ]:
-                        updated_channels |= apply_writes(
+                        apply_writes(
                            checkpoint,
                            channels,
                            [PregelTaskWrites((), INPUT, null_writes, [])],
@@ -1720,7 +1728,7 @@ class Pregel(
                            continue
                        next_tasks[tid].writes.append((k, v))
                    # clear all current tasks
-                    updated_channels |= apply_writes(
+                    apply_writes(
                        checkpoint,
                        channels,
                        next_tasks.values(),
@@ -1728,24 +1736,25 @@ class Pregel(
                        self.trigger_to_nodes,
                    )
                # save checkpoint
+                next_checkpoint = create_checkpoint(
+                    checkpoint,
+                    channels,
+                    step,
+                    get_next_version=checkpointer.get_next_version,
+                    channels_to_snapshot=fork_pending,
+                )
+                fork_pending.difference_update(next_checkpoint["channel_values"])
                next_config = checkpointer.put(
                    checkpoint_config,
-                    create_checkpoint(checkpoint, channels, step),
+                    next_checkpoint,
                    {
                        "source": "update",
                        "step": step + 1,
                        "parents": saved.metadata.get("parents", {}) if saved else {},
-                        **(
-                            advance_delta_counters(
-                                channels, updated_channels, prev_metadata=saved.metadata
-                            )
-                            if saved
-                            else {}
-                        ),
                    },
                    get_new_channel_versions(
                        checkpoint_previous_versions,
-                        checkpoint["channel_versions"],
+                        next_checkpoint["channel_versions"],
                    ),
                )
                return patch_checkpoint_map(
@@ -1760,7 +1769,7 @@ class Pregel(
                    )

                if input_writes := deque(map_input(self.input_channels, values)):
-                    updated_channels = apply_writes(
+                    apply_writes(
                        checkpoint,
                        channels,
                        [PregelTaskWrites((), INPUT, input_writes, [])],
@@ -1774,28 +1783,27 @@ class Pregel(
                        if saved and saved.metadata.get("step") is not None
                        else -1
                    )
+                    next_checkpoint = create_checkpoint(
+                        checkpoint,
+                        channels,
+                        next_step,
+                        get_next_version=checkpointer.get_next_version,
+                        channels_to_snapshot=fork_pending,
+                    )
+                    fork_pending.difference_update(next_checkpoint["channel_values"])
                    next_config = checkpointer.put(
                        checkpoint_config,
-                        create_checkpoint(checkpoint, channels, next_step),
+                        next_checkpoint,
                        {
                            "source": "input",
                            "step": next_step,
                            "parents": saved.metadata.get("parents", {})
                            if saved
                            else {},
-                            **(
-                                advance_delta_counters(
-                                    channels,
-                                    updated_channels,
-                                    prev_metadata=saved.metadata,
-                                )
-                                if saved
-                                else {}
-                            ),
                        },
                        get_new_channel_versions(
                            checkpoint_previous_versions,
-                            checkpoint["channel_versions"],
+                            next_checkpoint["channel_versions"],
                        ),
                    )

@@ -1837,12 +1845,6 @@ class Pregel(
                        "source": "fork",
                        "step": step + 1,
                        "parents": saved.metadata.get("parents", {}),
-                        # The copy has the same values and the same parent.
-                        **{
-                            k: v
-                            for k, v in saved.metadata.items()
-                            if k == "counters_since_delta_snapshot"
-                        },
                    },
                    {},
                )
@@ -2044,18 +2046,19 @@ class Pregel(
                    parents=saved.metadata.get("parents", {}) if saved else {},
                    saved_metadata=saved.metadata if saved else None,
                    is_fresh_thread=saved is None,
+                    fork_channels=fork_pending,
                )
            )
            checkpoint = create_checkpoint(
                checkpoint,
                channels,
                step + 1,
-                updated_channels=updated_channels if channels_to_snapshot else None,
                get_next_version=checkpointer.get_next_version
                if channels_to_snapshot
                else None,
                channels_to_snapshot=channels_to_snapshot,
            )
+            fork_pending.difference_update(checkpoint["channel_values"])
            next_config = checkpointer.put(
                checkpoint_config,
                checkpoint,
@@ -2127,12 +2130,21 @@ class Pregel(
            else:
                raise ValueError(f"Subgraph {recast} not found")

+        # Taken from the first superstep's base, and cleared by the first
+        # checkpoint that carries the snapshots, which `__copy__` does not write.
+        fork_pending: set[str] | None = None
+
        async def aperform_superstep(
            input_config: RunnableConfig, updates: Sequence[StateUpdate]
        ) -> RunnableConfig:
+            nonlocal fork_pending
            # get last checkpoint
            config = ensure_config(self.config, input_config)
            saved = await checkpointer.aget_tuple(config)
+            if fork_pending is None:
+                fork_pending = delta_channels_with_pending_writes(
+                    self.channels, saved.pending_writes if saved else None
+                )
            if saved is not None:
                self._migrate_checkpoint(saved.checkpoint)
            checkpoint = (
@@ -2169,7 +2181,6 @@ class Pregel(
                    raise InvalidUpdateError(
                        "Cannot apply multiple updates when clearing state"
                    )
-                updated_channels: set[str] = set()
                if saved is not None:
                    # tasks for this checkpoint
                    next_tasks = prepare_next_tasks(
@@ -2192,7 +2203,7 @@ class Pregel(
                        for w in saved.pending_writes or []
                        if w[0] == NULL_TASK_ID
                    ]:
-                        updated_channels |= apply_writes(
+                        apply_writes(
                            checkpoint,
                            channels,
                            [PregelTaskWrites((), INPUT, null_writes, [])],
@@ -2207,7 +2218,7 @@ class Pregel(
                            continue
                        next_tasks[tid].writes.append((k, v))
                    # clear all current tasks
-                    updated_channels |= apply_writes(
+                    apply_writes(
                        checkpoint,
                        channels,
                        next_tasks.values(),
@@ -2215,23 +2226,25 @@ class Pregel(
                        self.trigger_to_nodes,
                    )
                # save checkpoint
+                next_checkpoint = create_checkpoint(
+                    checkpoint,
+                    channels,
+                    step,
+                    get_next_version=checkpointer.get_next_version,
+                    channels_to_snapshot=fork_pending,
+                )
+                fork_pending.difference_update(next_checkpoint["channel_values"])
                next_config = await checkpointer.aput(
                    checkpoint_config,
-                    create_checkpoint(checkpoint, channels, step),
+                    next_checkpoint,
                    {
                        "source": "update",
                        "step": step + 1,
                        "parents": saved.metadata.get("parents", {}) if saved else {},
-                        **(
-                            advance_delta_counters(
-                                channels, updated_channels, prev_metadata=saved.metadata
-                            )
-                            if saved
-                            else {}
-                        ),
                    },
                    get_new_channel_versions(
-                        checkpoint_previous_versions, checkpoint["channel_versions"]
+                        checkpoint_previous_versions,
+                        next_checkpoint["channel_versions"],
                    ),
                )
                return patch_checkpoint_map(
@@ -2246,7 +2259,7 @@ class Pregel(
                    )

                if input_writes := deque(map_input(self.input_channels, values)):
-                    updated_channels = apply_writes(
+                    apply_writes(
                        checkpoint,
                        channels,
                        [PregelTaskWrites((), INPUT, input_writes, [])],
@@ -2260,28 +2273,27 @@ class Pregel(
                        if saved and saved.metadata.get("step") is not None
                        else -1
                    )
+                    next_checkpoint = create_checkpoint(
+                        checkpoint,
+                        channels,
+                        next_step,
+                        get_next_version=checkpointer.get_next_version,
+                        channels_to_snapshot=fork_pending,
+                    )
+                    fork_pending.difference_update(next_checkpoint["channel_values"])
                    next_config = await checkpointer.aput(
                        checkpoint_config,
-                        create_checkpoint(checkpoint, channels, next_step),
+                        next_checkpoint,
                        {
                            "source": "input",
                            "step": next_step,
                            "parents": saved.metadata.get("parents", {})
                            if saved
                            else {},
-                            **(
-                                advance_delta_counters(
-                                    channels,
-                                    updated_channels,
-                                    prev_metadata=saved.metadata,
-                                )
-                                if saved
-                                else {}
-                            ),
                        },
                        get_new_channel_versions(
                            checkpoint_previous_versions,
-                            checkpoint["channel_versions"],
+                            next_checkpoint["channel_versions"],
                        ),
                    )

@@ -2323,12 +2335,6 @@ class Pregel(
                        "source": "fork",
                        "step": step + 1,
                        "parents": saved.metadata.get("parents", {}),
-                        # The copy has the same values and the same parent.
-                        **{
-                            k: v
-                            for k, v in saved.metadata.items()
-                            if k == "counters_since_delta_snapshot"
-                        },
                    },
                    {},
                )
@@ -2527,18 +2533,19 @@ class Pregel(
                    parents=saved.metadata.get("parents", {}) if saved else {},
                    saved_metadata=saved.metadata if saved else None,
                    is_fresh_thread=saved is None,
+                    fork_channels=fork_pending,
                )
            )
            checkpoint = create_checkpoint(
                checkpoint,
                channels,
                step + 1,
-                updated_channels=updated_channels if channels_to_snapshot else None,
                get_next_version=checkpointer.get_next_version
                if channels_to_snapshot
                else None,
                channels_to_snapshot=channels_to_snapshot,
            )
+            fork_pending.difference_update(checkpoint["channel_values"])
            next_config = await checkpointer.aput(
                checkpoint_config,
                checkpoint,
@@ -85,11 +85,13 @@ class MemorySaverAssertImmutable(InMemorySaver):
                )
                == saved
            ), config["configurable"]["checkpoint_ns"]
+        next_config = super().put(config, checkpoint, metadata, new_versions)
+        # Read back, not the object handed in: a DeltaChannel a step did not
+        # write is refilled on read from the blob its inherited version points at.
        self.storage_for_copies[thread_id][checkpoint_ns][checkpoint["id"]] = (
-            self.serde.dumps_typed(checkpoint)
+            self.serde.dumps_typed(super().get(next_config))
        )
-        # call super to write checkpoint
-        return super().put(config, checkpoint, metadata, new_versions)
+        return next_config


 class MemorySaverNoPending(InMemorySaver):
@@ -0,0 +1,533 @@
+"""Forking a thread must not replay the abandoned branch into the fork.
+
+Every graph carries a `DeltaChannel` and a plain reducer channel fed the same
+values; the plain channel needs no replay, so it is the oracle.
+"""
+
+from collections.abc import Sequence
+from operator import add
+from typing import Annotated, Any
+
+import pytest
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.base import BaseCheckpointSaver
+from langgraph.checkpoint.serde.types import _DeltaSnapshot
+from typing_extensions import TypedDict
+
+from langgraph._internal._constants import INPUT
+from langgraph.channels.delta import DeltaChannel
+from langgraph.graph import END, START, StateGraph
+from langgraph.types import Command, Durability, StateSnapshot, StateUpdate, interrupt
+
+pytestmark = pytest.mark.anyio
+
+
+def _append(current: list | None, writes: Sequence[Any]) -> list:
+    out = list(current or [])
+    for write in writes:
+        out.extend(write if isinstance(write, list) else [write])
+    return out
+
+
+class _State(TypedDict):
+    log: Annotated[list, DeltaChannel(_append, snapshot_frequency=1000)]
+    plain: Annotated[list, add]
+    other: Annotated[list, add]
+
+
+def _build(checkpointer: BaseCheckpointSaver, tag: str) -> Any:
+    def node(state: _State) -> dict:
+        return {"log": [f"{tag}-out"], "plain": [f"{tag}-out"]}
+
+    builder = StateGraph(_State)
+    builder.add_node("n", node)
+    builder.set_entry_point("n")
+    builder.set_finish_point("n")
+    return builder.compile(checkpointer=checkpointer)
+
+
+def _build_without_delta_writes(checkpointer: BaseCheckpointSaver, tag: str) -> Any:
+    def node(state: _State) -> dict:
+        return {"other": [f"{tag}-other"]}
+
+    builder = StateGraph(_State)
+    builder.add_node("n", node)
+    builder.set_entry_point("n")
+    builder.set_finish_point("n")
+    return builder.compile(checkpointer=checkpointer)
+
+
+def _thread(thread_id: str) -> RunnableConfig:
+    return {"configurable": {"thread_id": thread_id}}
+
+
+def _at(config: RunnableConfig, snapshot: StateSnapshot) -> RunnableConfig:
+    return {
+        "configurable": {
+            **config["configurable"],
+            "checkpoint_ns": "",
+            "checkpoint_id": snapshot.config["configurable"]["checkpoint_id"],
+        }
+    }
+
+
+def _both(marker: str) -> dict:
+    return {"log": [marker], "plain": [marker]}
+
+
+def _snapshotted_checkpoints(
+    checkpointer: BaseCheckpointSaver, config: RunnableConfig
+) -> list[str]:
+    return [
+        tuple_.config["configurable"]["checkpoint_id"]
+        for tuple_ in checkpointer.list(config)
+        if isinstance(tuple_.checkpoint["channel_values"].get("log"), _DeltaSnapshot)
+    ]
+
+
+def _assert_fork_is_clean(state: StateSnapshot, abandoned: str) -> None:
+    assert state.values["log"] == state.values["plain"], (
+        f"delta channel diverged from the plain channel: "
+        f"{state.values['log']} != {state.values['plain']}"
+    )
+    assert abandoned not in state.values["log"], (
+        f"{abandoned!r} belongs to the branch the fork replaced, "
+        f"but was replayed into {state.values['log']}"
+    )
+
+
+def test_fork_by_invoke(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    _build(sync_checkpointer, "first").invoke(
+        _both("in-1"), config, durability=durability
+    )
+    graph = _build(sync_checkpointer, "second")
+    graph.invoke(_both("in-2"), config, durability=durability)
+    abandoned_head = graph.get_state(config)
+
+    base = next(
+        snapshot
+        for snapshot in graph.get_state_history(config)
+        if "in-2" not in snapshot.values["log"]
+    )
+    _build(sync_checkpointer, "third").invoke(
+        _both("in-3"), _at(config, base), durability=durability
+    )
+
+    state = graph.get_state(config)
+    _assert_fork_is_clean(state, "in-2")
+    assert state.values["log"] == [*base.values["log"], "in-3", "third-out"]
+
+    abandoned = graph.get_state(abandoned_head.config).values
+    assert abandoned["log"] == abandoned["plain"] == abandoned_head.values["log"]
+
+
+async def test_afork_by_invoke(
+    async_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    await _build(async_checkpointer, "first").ainvoke(
+        _both("in-1"), config, durability=durability
+    )
+    graph = _build(async_checkpointer, "second")
+    await graph.ainvoke(_both("in-2"), config, durability=durability)
+    abandoned_head = await graph.aget_state(config)
+
+    base = await anext(
+        snapshot
+        async for snapshot in graph.aget_state_history(config)
+        if "in-2" not in snapshot.values["log"]
+    )
+    await _build(async_checkpointer, "third").ainvoke(
+        _both("in-3"), _at(config, base), durability=durability
+    )
+
+    state = await graph.aget_state(config)
+    _assert_fork_is_clean(state, "in-2")
+    assert state.values["log"] == [*base.values["log"], "in-3", "third-out"]
+
+    abandoned = (await graph.aget_state(abandoned_head.config)).values
+    assert abandoned["log"] == abandoned["plain"] == abandoned_head.values["log"]
+
+
+def test_fork_off_checkpoint_before_first_input(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "first")
+    graph.invoke(_both("in-1"), config, durability=durability)
+
+    root = list(graph.get_state_history(config))[-1]
+    assert root.values["log"] == []
+
+    _build(sync_checkpointer, "third").invoke(
+        _both("in-9"), _at(config, root), durability=durability
+    )
+
+    state = graph.get_state(config)
+    _assert_fork_is_clean(state, "in-1")
+    assert state.values["log"] == ["in-9", "third-out"]
+
+
+async def test_afork_off_checkpoint_before_first_input(
+    async_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    graph = _build(async_checkpointer, "first")
+    await graph.ainvoke(_both("in-1"), config, durability=durability)
+
+    root = [snapshot async for snapshot in graph.aget_state_history(config)][-1]
+    assert root.values["log"] == []
+
+    await _build(async_checkpointer, "third").ainvoke(
+        _both("in-9"), _at(config, root), durability=durability
+    )
+
+    state = await graph.aget_state(config)
+    _assert_fork_is_clean(state, "in-1")
+    assert state.values["log"] == ["in-9", "third-out"]
+
+
+def test_fork_by_update_state(sync_checkpointer: BaseCheckpointSaver) -> None:
+    config = _thread("t")
+    _build(sync_checkpointer, "first").invoke(_both("in-1"), config)
+    graph = _build(sync_checkpointer, "second")
+    graph.invoke(_both("in-2"), config)
+
+    base = next(
+        snapshot
+        for snapshot in graph.get_state_history(config)
+        if "in-2" not in snapshot.values["log"]
+    )
+    forked = graph.update_state(_at(config, base), _both("patched"))
+
+    state = graph.get_state(forked)
+    _assert_fork_is_clean(state, "in-2")
+    assert state.values["log"] == [*base.values["log"], "patched"]
+
+
+async def test_afork_by_update_state(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config = _thread("t")
+    await _build(async_checkpointer, "first").ainvoke(_both("in-1"), config)
+    graph = _build(async_checkpointer, "second")
+    await graph.ainvoke(_both("in-2"), config)
+
+    base = await anext(
+        snapshot
+        async for snapshot in graph.aget_state_history(config)
+        if "in-2" not in snapshot.values["log"]
+    )
+    forked = await graph.aupdate_state(_at(config, base), _both("patched"))
+
+    state = await graph.aget_state(forked)
+    _assert_fork_is_clean(state, "in-2")
+    assert state.values["log"] == [*base.values["log"], "patched"]
+
+
+def test_unaddressed_run_keeps_snapshot_cadence(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "first")
+    graph.invoke(_both("in-1"), config, durability=durability)
+    graph.invoke(_both("in-2"), config, durability=durability)
+
+    assert not _snapshotted_checkpoints(sync_checkpointer, config)
+
+
+def test_fork_before_first_value_when_fork_never_writes_the_channel(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "first")
+    graph.invoke(_both("in-1"), config, durability=durability)
+
+    root = list(graph.get_state_history(config))[-1]
+    assert root.values["log"] == []
+
+    _build_without_delta_writes(sync_checkpointer, "third").invoke(
+        {"other": ["in-9"]}, _at(config, root), durability=durability
+    )
+
+    state = graph.get_state(config)
+    _assert_fork_is_clean(state, "in-1")
+    assert state.values["log"] == []
+
+
+async def test_afork_before_first_value_when_fork_never_writes_the_channel(
+    async_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    graph = _build(async_checkpointer, "first")
+    await graph.ainvoke(_both("in-1"), config, durability=durability)
+
+    root = [snapshot async for snapshot in graph.aget_state_history(config)][-1]
+    assert root.values["log"] == []
+
+    await _build_without_delta_writes(async_checkpointer, "third").ainvoke(
+        {"other": ["in-9"]}, _at(config, root), durability=durability
+    )
+
+    state = await graph.aget_state(config)
+    _assert_fork_is_clean(state, "in-1")
+    assert state.values["log"] == []
+
+
+def test_fork_before_first_value_by_bulk_update(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "first")
+    graph.invoke(_both("in-1"), config)
+
+    root = list(graph.get_state_history(config))[-1]
+    assert root.values["log"] == []
+
+    forked = graph.bulk_update_state(
+        _at(config, root),
+        [
+            [StateUpdate({"other": ["s1"]}, "n")],
+            [StateUpdate(_both("s2"), "n")],
+        ],
+    )
+
+    state = graph.get_state(forked)
+    _assert_fork_is_clean(state, "in-1")
+    assert state.values["log"] == ["s2"]
+
+
+@pytest.mark.parametrize("first_as_node", [INPUT, END, "__copy__"])
+def test_fork_by_bulk_update_whose_first_superstep_skips_the_plan(
+    sync_checkpointer: BaseCheckpointSaver, first_as_node: str
+) -> None:
+    config = _thread("t")
+    _build(sync_checkpointer, "first").invoke(_both("in-1"), config)
+    graph = _build(sync_checkpointer, "second")
+    graph.invoke(_both("in-2"), config)
+
+    base = next(
+        snapshot
+        for snapshot in graph.get_state_history(config)
+        if "in-2" not in snapshot.values["log"]
+    )
+    first = (
+        StateUpdate(_both("first-step"), first_as_node)
+        if first_as_node == INPUT
+        else StateUpdate(None, first_as_node)
+    )
+    forked = graph.bulk_update_state(
+        _at(config, base),
+        [[first], [StateUpdate(_both("second-step"), "n")]],
+    )
+
+    state = graph.get_state(forked)
+    assert state.values["log"] == state.values["plain"], (
+        f"delta channel diverged from the plain channel: "
+        f"{state.values['log']} != {state.values['plain']}"
+    )
+
+
+def test_unaddressed_bulk_update_keeps_snapshot_cadence(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "first")
+    graph.invoke(_both("in-1"), config)
+
+    graph.bulk_update_state(
+        config,
+        [[StateUpdate(_both(f"u{i}"), "n")] for i in range(4)],
+    )
+
+    assert not _snapshotted_checkpoints(sync_checkpointer, config)
+
+
+def _build_paused_before_b(checkpointer: BaseCheckpointSaver) -> Any:
+    builder = StateGraph(_State)
+    builder.add_node("a", lambda state: _both("a"))
+    builder.add_node("b", lambda state: _both("b"))
+    builder.add_edge(START, "a")
+    builder.add_edge("a", "b")
+    builder.add_edge("b", END)
+    return builder.compile(checkpointer=checkpointer, interrupt_before=["b"])
+
+
+def _build_parallel_interrupt(checkpointer: BaseCheckpointSaver) -> Any:
+    def ask(state: _State) -> dict:
+        interrupt("approve?")
+        return {"other": ["q"]}
+
+    builder = StateGraph(_State)
+    builder.add_node("p", lambda state: _both("p"))
+    builder.add_node("q", ask)
+    builder.add_edge(START, "p")
+    builder.add_edge(START, "q")
+    return builder.compile(checkpointer=checkpointer)
+
+
+def test_resume_at_interrupt_before_with_the_head_checkpoint_id_runs_the_node(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    graph = _build_paused_before_b(sync_checkpointer)
+    graph.invoke(_both("in"), config, durability=durability)
+
+    graph.invoke(None, graph.get_state(config).config, durability=durability)
+
+    state = graph.get_state(config)
+    assert state.next == (), f"resume paused again before {state.next}"
+    assert state.values["log"] == state.values["plain"] == ["in", "a", "b"]
+
+
+async def test_aresume_at_interrupt_before_with_the_head_checkpoint_id_runs_the_node(
+    async_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    graph = _build_paused_before_b(async_checkpointer)
+    await graph.ainvoke(_both("in"), config, durability=durability)
+
+    await graph.ainvoke(
+        None, (await graph.aget_state(config)).config, durability=durability
+    )
+
+    state = await graph.aget_state(config)
+    assert state.next == (), f"resume paused again before {state.next}"
+    assert state.values["log"] == state.values["plain"] == ["in", "a", "b"]
+
+
+def test_replay_from_a_paused_checkpoint_runs_the_node_once(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config = _thread("t")
+    graph = _build_paused_before_b(sync_checkpointer)
+    graph.invoke(_both("in"), config)
+    paused = graph.get_state(config).config
+    graph.invoke(None, config)
+
+    graph.invoke(None, paused)
+
+    state = graph.get_state(config)
+    assert state.next == (), f"replay paused again before {state.next}"
+    assert state.values["log"] == state.values["plain"] == ["in", "a", "b"]
+
+
+@pytest.mark.parametrize("addressed", [False, True])
+def test_new_input_on_an_interrupted_head_does_not_replay_its_pending_writes(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability, addressed: bool
+) -> None:
+    config = _thread("t")
+    graph = _build_parallel_interrupt(sync_checkpointer)
+    graph.invoke(_both("in-1"), config, durability=durability)
+    head = graph.get_state(config).config
+
+    graph.invoke(_both("in-2"), head if addressed else config, durability=durability)
+
+    state = graph.get_state(config)
+    assert state.values["log"] == state.values["plain"] == ["in-1", "in-2", "p"]
+
+
+def _build_deferred_after_interrupt(checkpointer: BaseCheckpointSaver) -> Any:
+    builder = StateGraph(_State)
+    builder.add_node("a", lambda state: _both("a"))
+    builder.add_node("b", lambda state: _both("b"), defer=True)
+    builder.add_node("c", lambda state: {})
+    builder.add_edge(START, "a")
+    builder.add_edge("a", "b")
+    builder.add_edge("a", "c")
+    return builder.compile(checkpointer=checkpointer, interrupt_after=["a"])
+
+
+@pytest.mark.parametrize(
+    "durability",
+    [
+        "sync",
+        "async",
+        pytest.param(
+            "exit",
+            marks=pytest.mark.xfail(
+                reason="exit durability stores a resumed run's loaded writes twice",
+                strict=True,
+            ),
+        ),
+    ],
+)
+def test_resume_on_an_interrupted_head_consumes_its_writes_without_a_snapshot(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    graph = _build_parallel_interrupt(sync_checkpointer)
+    graph.invoke(_both("in-1"), config, durability=durability)
+
+    graph.invoke(Command(resume="yes"), config, durability=durability)
+
+    state = graph.get_state(config)
+    assert state.next == ()
+    assert state.values["log"] == state.values["plain"] == ["in-1", "p"]
+    assert not _snapshotted_checkpoints(sync_checkpointer, config)
+
+
+def test_resume_addressed_at_an_interrupted_head_reruns_its_tasks_once(
+    sync_checkpointer: BaseCheckpointSaver, durability: Durability
+) -> None:
+    config = _thread("t")
+    graph = _build_parallel_interrupt(sync_checkpointer)
+    graph.invoke(_both("in-1"), config, durability=durability)
+
+    graph.invoke(
+        Command(resume="yes"), graph.get_state(config).config, durability=durability
+    )
+
+    state = graph.get_state(config)
+    assert state.next == ()
+    assert state.values["log"] == state.values["plain"] == ["in-1", "p"]
+
+
+def test_update_state_with_the_head_checkpoint_id_keeps_a_deferred_node(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _build_deferred_after_interrupt(sync_checkpointer)
+    config = _thread("t")
+    graph.invoke(_both("in"), config)
+
+    graph.update_state(graph.get_state(config).config, _both("u"), as_node="c")
+    graph.invoke(None, config)
+
+    state = graph.get_state(config)
+    assert state.next == (), f"deferred node never ran, still pending: {state.next}"
+    assert state.values["log"] == state.values["plain"] == ["in", "a", "u", "b"]
+
+
+async def test_aupdate_state_with_the_head_checkpoint_id_keeps_a_deferred_node(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    graph = _build_deferred_after_interrupt(async_checkpointer)
+    config = _thread("t")
+    await graph.ainvoke(_both("in"), config)
+
+    await graph.aupdate_state(
+        (await graph.aget_state(config)).config, _both("u"), as_node="c"
+    )
+    await graph.ainvoke(None, config)
+
+    state = await graph.aget_state(config)
+    assert state.next == (), f"deferred node never ran, still pending: {state.next}"
+    assert state.values["log"] == state.values["plain"] == ["in", "a", "u", "b"]
+
+
+def test_turns_addressed_at_the_head_store_no_snapshot(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    config = _thread("t")
+    graph = _build(sync_checkpointer, "turn")
+    graph.invoke(_both("in-1"), config)
+    for turn in range(2, 5):
+        graph.invoke(_both(f"in-{turn}"), graph.get_state(config).config)
+
+    assert not _snapshotted_checkpoints(sync_checkpointer, config)
+    assert (
+        graph.get_state(config).values["log"] == graph.get_state(config).values["plain"]
+    )
@@ -147,60 +147,6 @@ async def test_predicate_fires_on_supersteps_overflow() -> None:
    assert "x" not in result2


-def _delta_counters(saver: InMemorySaver, config: Any) -> dict[str, list[int]]:
-    tup = saver.get_tuple(config)
-    assert tup is not None
-    counters = tup.metadata.get("counters_since_delta_snapshot") or {}
-    return {ch: list(c) for ch, c in counters.items()}
-
-
-_UPDATE_PATHS_WITHOUT_THE_SNAPSHOT_PLAN = pytest.mark.parametrize(
-    ("values", "as_node", "supersteps"),
-    [
-        (None, END, 1),
-        (None, "__copy__", 0),
-        ({"a": []}, "__input__", 1),
-    ],
-    ids=["clear as END", "copy", "update as input"],
-)
-
-
-@_UPDATE_PATHS_WITHOUT_THE_SNAPSHOT_PLAN
-def test_update_state_path_keeps_delta_counters(
-    values: Any, as_node: str, supersteps: int
-) -> None:
-    saver = InMemorySaver()
-    graph = _build_two_channel_graph(saver)
-    config = {"configurable": {"thread_id": "counters"}}
-    graph.invoke({"a": ["seed-a"], "b": ["seed-b"]}, config)
-    before = _delta_counters(saver, config)
-    assert set(before) == {"a", "b"}, f"both channels need live counters: {before}"
-
-    updated = graph.update_state(config, values, as_node=as_node)
-
-    assert _delta_counters(saver, updated) == {
-        ch: [u, s + supersteps] for ch, (u, s) in before.items()
-    }
-
-
-@_UPDATE_PATHS_WITHOUT_THE_SNAPSHOT_PLAN
-async def test_aupdate_state_path_keeps_delta_counters(
-    values: Any, as_node: str, supersteps: int
-) -> None:
-    saver = InMemorySaver()
-    graph = _build_two_channel_graph(saver)
-    config = {"configurable": {"thread_id": "counters"}}
-    await graph.ainvoke({"a": ["seed-a"], "b": ["seed-b"]}, config)
-    before = _delta_counters(saver, config)
-    assert set(before) == {"a", "b"}, f"both channels need live counters: {before}"
-
-    updated = await graph.aupdate_state(config, values, as_node=as_node)
-
-    assert _delta_counters(saver, updated) == {
-        ch: [u, s + supersteps] for ch, (u, s) in before.items()
-    }
-
-
 async def test_counter_reset_after_supersteps_snapshot() -> None:
    """After the supersteps bound triggers a snapshot, the counters for
    that channel reset. Verify by using a bound higher than one run's
@@ -338,3 +338,29 @@ def test_state_history_chain_after_fresh_update_state_delta_channel() -> None:
    assert update_snapshot.metadata["step"] == 0
    assert update_snapshot.parent_config is None
    assert [m.content for m in update_snapshot.values["messages"]] == ["hello"]
+
+
+def test_update_state_that_snapshots_keeps_a_deferred_node_pending() -> None:
+    channel = DeltaChannel(_messages_delta_reducer, snapshot_frequency=1)
+
+    class State(TypedDict):
+        messages: Annotated[list, channel]
+
+    builder = StateGraph(State)
+    builder.add_node("a", lambda state: {"messages": [HumanMessage("a", id="a")]})
+    builder.add_node(
+        "b", lambda state: {"messages": [HumanMessage("b", id="b")]}, defer=True
+    )
+    builder.add_node("c", lambda state: {})
+    builder.add_edge(START, "a")
+    builder.add_edge("a", "b")
+    builder.add_edge("a", "c")
+    graph = builder.compile(checkpointer=InMemorySaver(), interrupt_after=["a"])
+    config = {"configurable": {"thread_id": "t"}}
+    graph.invoke({"messages": [HumanMessage("s", id="s")]}, config)
+
+    graph.update_state(config, {"messages": [HumanMessage("u", id="u")]}, as_node="c")
+    final = graph.invoke(None, config)
+
+    assert [m.content for m in final["messages"]] == ["s", "a", "u", "b"]
+    assert graph.get_state(config).next == ()
@@ -2841,11 +2841,11 @@ wheels = [

 [[package]]
 name = "pyjwt"
-version = "2.15.1"
+version = "2.13.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/43/ea/5194e52748b0da83d71e082d75496eaec6e58f419f5e184786ded517e6a9/pyjwt-2.15.1.tar.gz", hash = "sha256:4f259e80cdfb6b3fc18a7de51fd1ef9ec79652f25019bae68975ca2468a34df8", size = 121252, upload-time = "2026-09-28T18:40:42.598Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/3b/81/58d0ac84e1ef3a3843791d6954d94c0b33d526c75eeb1efbce9d0a4c4077/pyjwt-2.13.0.tar.gz", hash = "sha256:41571c89ca91598c79e8ef18a2d07367d4810fbbd6f637794879baf1b7703423", size = 107515, upload-time = "2026-05-21T19:54:36.618Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/50/ca/44de4e75f8aadc457f0634be3b542815078ded46dca30efb960edeecad6e/pyjwt-2.15.1-py3-none-any.whl", hash = "sha256:42d59d631f7768a1028a64c7ff581a9bf7519804daf91fc5b6c56e30eec5e193", size = 33860, upload-time = "2026-09-28T18:40:41.429Z" },
+    { url = "https://files.pythonhosted.org/packages/a3/5e/ecf12fdb62546d64385c158514e9b2b671f7832108ef2ecd2020ce0af2d1/pyjwt-2.13.0-py3-none-any.whl", hash = "sha256:66adcc2aff09b3f1bbd95fc1e1577df8ac8723c978552fd43304c8a290ac5728", size = 31274, upload-time = "2026-05-21T19:54:35.362Z" },
 ]

 [[package]]
@@ -3687,11 +3687,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.8.0"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
 ]

 [[package]]
@@ -1363,11 +1363,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.8.0"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
 ]

 [[package]]
@@ -1175,11 +1175,11 @@ wheels = [

 [[package]]
 name = "urllib3"
-version = "2.8.0"
+version = "2.7.0"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/e3/05/b17359e1cefb4f909b5e40b1b90a496d987258916dbbf88e842c729f510e/urllib3-2.8.0.tar.gz", hash = "sha256:63bf2ead4c879426ebf22ef2a781eeb4aa3b4ae798a0435506f8687fd5bb9b63", size = 458972, upload-time = "2026-09-15T19:29:36.253Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/53/0c/06f8b233b8fd13b9e5ee11424ef85419ba0d8ba0b3138bf360be2ff56953/urllib3-2.7.0.tar.gz", hash = "sha256:231e0ec3b63ceb14667c67be60f2f2c40a518cb38b03af60abc813da26505f4c", size = 433602, upload-time = "2026-05-07T16:13:18.596Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/92/9d/c4e665119135114480843e7ab388fa94d8480650450e6f8e26b70d323a4c/urllib3-2.8.0-py3-none-any.whl", hash = "sha256:0cf3cae568d36aa9576b28dfb35f11328f1cb974ca7647d9475ebb86c75ac6e3", size = 135717, upload-time = "2026-09-15T19:29:34.577Z" },
+    { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" },
 ]

 [[package]]
Author	SHA1	Message	Date
Elior Nataf Lackritz	858e55f232	refactor(langgraph): call create_checkpoint directly for the fork seal create_fork_checkpoint only forwarded to create_checkpoint, and an empty fork set bumps nothing there either, so the four update_state call sites pass the set straight through. The fork tests also had two identical helpers; keep one.	2026-09-30 15:07:17 -04:00
Elior Nataf Lackritz	e59ecbc23d	revert(langgraph): leave exit-mode resume writes as they are on main Skipping the loaded writes in the exit accumulator removed the duplicate replay but reordered it: the loaded write keeps its task id while the run's later writes get step-prefixed ids that sort first. That ordering problem belongs to exit mode, not to forking, and exists on main without a fork, so it gets its own fix. The resume test marks exit durability as an expected failure until then.	2026-09-29 11:39:36 -04:00
Elior Nataf Lackritz	719a4d71bc	fix(langgraph): skip the seal on a resume that reuses the head's writes A resume that is not replaying reuses the head's pending writes instead of rerunning their tasks, so none of them can leak; deciding in `_first`, where that is known, stops a plain `Command(resume=...)` from storing a snapshot. A replaying resume reruns the tasks, so it still seals. Exit mode also re-recorded the writes a resume loaded from the head: they were already stored there, and every later read replayed them twice. The exit accumulator now skips them.	2026-09-29 09:42:55 -04:00
Elior Nataf Lackritz	ddaf708cd0	fix(langgraph): snapshot only what a fork can leak, and hide the bump A fork used to snapshot every DeltaChannel whenever the caller passed a checkpoint_id. That fired on every turn a client addresses the head (storing a full copy of the channel per turn), missed new input sent to an interrupted head without an id, and its storage-only version bump read as a real write: interrupt_before fired again on resume, and a replay paused at a node it had already passed. Snapshot the delta channels the base checkpoint has pending writes for, since only those can leak into a branch that does not consume them, and advance versions_seen past any bump that only stores a snapshot, including the interrupt tracker. update_state no longer records its narrower updated_channels when it snapshots, so a deferred node listed in next still runs on resume (#9089).	2026-09-28 19:54:23 -04:00
Elior Nataf Lackritz	9d5b0f1991	refactor(langgraph): drop the deferred fork-snapshot queue Now that a forced snapshot mints a version for a never-written channel, the first checkpoint a forked run writes seals every delta channel, so _delta_channels_awaiting_fork_snapshot never outlived it. Seed _delta_channels_forced_snapshot directly at loop construction instead. Also asserts that forking by invoke leaves the abandoned branch's reads intact, and trims comments and test docstrings.	2026-09-28 19:54:23 -04:00
Elior Nataf Lackritz	9d16b52955	fix(langgraph): seal a fork whose delta channel has no value yet A DeltaChannel that was never written on the branch being forked has no value to snapshot and no entry in channel_versions, so create_checkpoint skipped it and the fork's first checkpoint recorded no boundary at all. The walk then ran past the fork into the shared base and collected the abandoned branch's writes, the same failure this branch already fixes for channels that do have a value. Two shapes leaked. A run forking off a checkpoint older than the channel's first value and never writing that channel returned ['in-1'] where the plain-channel oracle returned []. A bulk update writing the delta key only in its second superstep returned ['in-1', 's2'] against ['s2']. No new blob type is needed. _DeltaSnapshot already carries the value and is already serialized by every saver, and from_checkpoint turns MISSING into typ(), so _DeltaSnapshot(typ()) reconstructs to the same empty value the channel would have had. What was missing is a version: without one, put drops the blob as not-a-new-version, so mint a first one. Deferring the seal to a later superstep does not work. That superstep reconstructs through the still-unsealed checkpoint and would only bake the corrupted value into its own snapshot. Checked that minting a version does not fire nodes that subscribe to the channel: a raw Pregel node subscribed directly to the delta channel stays silent across the fork. Reported by the Open SWE review bot on #8548.	2026-09-28 19:54:23 -04:00
Elior Nataf Lackritz	2de0c47c1f	test(langgraph): compare read-back checkpoints in the immutability saver MemorySaverAssertImmutable recorded the checkpoint object handed to put, then compared it against one read back through get. Those two are not the same shape: channel_values are stored per (channel, version), so a channel a step did not write is refilled from the blob its inherited version still points at. Every channel except DeltaChannel writes its value into channel_values on every checkpoint, so the two agreed by accident. A DeltaChannel stores nothing except at a snapshot, so once one snapshots and a later step does not write it, the saver reports a checkpoint that changed after it was written when nothing was mutated. Reproducible on main with no fork involved: a delta channel with snapshot_frequency=1 written by the first node and left alone by the next two trips the assertion. Existing delta tests miss it only because they all use snapshot_frequency=1000. Record what the saver reads back instead. Comparing read-back against read-back still catches a checkpoint whose stored data really changed.	2026-09-28 19:54:23 -04:00
Elior Nataf Lackritz	377083220e	fix(langgraph): seal a fork on the first checkpoint it writes The as_node INPUT, END and __copy__ paths write a checkpoint and return before create_checkpoint_plan_for_update_state_api runs, so a bulk update whose first superstep took one of them left the branch unsealed. Only INPUT actually leaked: END absorbs the base's already-run task writes, so its delta and plain channels agree. Sealing on a later superstep does not help. By then that superstep has reconstructed its value by walking through the unsealed checkpoint into the shared base, so it snapshots an already-corrupted list. The fork's first checkpoint is the one that has to carry the blob, which is what create_fork_checkpoint does. That snapshot was still being dropped by put: these paths apply writes to the input channel, not the delta channel, so nothing bumped the delta channel's version and it never entered new_versions. Pass get_next_version for the manual bump, the same reason exit mode needs it, and derive new_versions from the returned checkpoint. fork_pending tracks what is still owed, mirroring _delta_channels_awaiting_fork_snapshot in _loop.py. Caught by the Open SWE review bot on #8548.	2026-09-28 19:54:23 -04:00
Elior Nataf Lackritz	9c5914861b	fix(langgraph): only fork on the first superstep of a bulk update perform_superstep returns the config of the checkpoint it just wrote and bulk_update_state feeds that back in, so from the second superstep on the incoming config always names a checkpoint whether or not the caller addressed one. Deriving the fork flag from it made every superstep after the first force-snapshot every available DeltaChannel and reset its cadence, storing the whole growing value once per superstep. Resolve the flag once from the caller's config and pass it explicitly, true only for the first superstep. The clear-tasks recursion carries it through, since the checkpoint written there has no delta snapshot and so leaves a fork unsealed. Caught by the Open SWE review bot on #8548.	2026-09-28 19:54:23 -04:00
Elior Nataf Lackritz AnnaSuSu UditDewan	24cf33f348	fix(langgraph): don't replay an abandoned branch into a DeltaChannel fork Addressing an older checkpoint creates a fork: the shared base ends up with two children and keeps the checkpoint_writes of the branch the fork abandons. Nothing records which child consumed which write, so the DeltaChannel ancestor walk collected the abandoned branch's writes too. Live execution was correct; only the reconstruction after a reload was wrong, and it was wrong on every saver. Fixed on the write side, so no saver changes are needed. A run launched against an explicitly addressed checkpoint forces every DeltaChannel to snapshot into its first checkpoint, terminating the walk inside the fork instead of at the shared base. This mirrors the existing force-snapshot for Overwrite writes, hence the rename to _delta_channels_forced_snapshot. update_state against an older checkpoint takes the same path, for the same reason is_fresh_thread already does. A channel with no value at the fork base cannot carry a snapshot blob yet, so the request stays queued until the first superstep that gives it one. Cost is one snapshot per addressed run, not per superstep. Fixes #8443 Co-Authored-By: AnnaSuSu <64579968+AnnaSuSu@users.noreply.github.com> Co-Authored-By: UditDewan <194863456+UditDewan@users.noreply.github.com>	2026-09-28 19:54:23 -04:00