fix(langgraph): share one saver resolution between runs and state methods

`_defaults` picked the run's saver on its own, and disagreed with the state methods on `checkpointer=False`: a run ignored a lent saver while `update_state` wrote a checkpoint with it that no run reads. A `checkpointer=True` root also got `True` back from `_state_checkpointer` and failed with an AttributeError instead of the run's error. Both now go through `_resolve_checkpointer`, and the state methods require an actual saver.
fix(langgraph): resolve the saver and checkpointer=True namespace in one place
2026-10-01 22:15:11 +02:00 · 2026-09-30 12:24:47 -04:00 · 2026-09-29 10:52:52 -04:00 · 2026-09-29 10:48:45 -04:00 · 2026-09-29 10:48:45 -04:00 · 2026-09-29 10:48:45 -04:00
22 changed files with 640 additions and 595 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]]
@@ -114,26 +114,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],
@@ -246,6 +226,20 @@ def _needs_replay(spec: BaseChannel, stored: object) -> bool:
    return stored is MISSING


+def _require_saver_for_history(
+    checkpoint: Checkpoint,
+    delta_channels: list[str],
+    saver: BaseCheckpointSaver | None,
+    config: RunnableConfig | None,
+) -> None:
+    written = [k for k in delta_channels if k in checkpoint["channel_versions"]]
+    if written and (saver is None or config is None):
+        raise ValueError(
+            f"DeltaChannel {written} has history to replay but no checkpointer "
+            "or config was passed to read it"
+        )
+
+
 def channels_from_checkpoint(
    specs: Mapping[str, BaseChannel | ManagedValueSpec],
    checkpoint: Checkpoint,
@@ -276,6 +270,7 @@ def channels_from_checkpoint(
        for k, spec in channel_specs.items()
        if _needs_replay(spec, checkpoint["channel_values"].get(k, MISSING))
    ]
+    _require_saver_for_history(checkpoint, delta_channels, saver, config)
    histories: Mapping[str, Any] = {}
    if delta_channels and saver is not None and config is not None:
        histories = saver.get_delta_channel_history(
@@ -318,6 +313,7 @@ async def achannels_from_checkpoint(
        for k, spec in channel_specs.items()
        if _needs_replay(spec, checkpoint["channel_values"].get(k, MISSING))
    ]
+    _require_saver_for_history(checkpoint, delta_channels, saver, config)
    histories: Mapping[str, Any] = {}
    if delta_channels and saver is not None and config is not None:
        histories = await saver.aget_delta_channel_history(
@@ -129,7 +129,6 @@ from langgraph.pregel._algo import (
 from langgraph.pregel._call import identifier
 from langgraph.pregel._checkpoint import (
    achannels_from_checkpoint,
-    advance_delta_counters,
    channels_from_checkpoint,
    copy_checkpoint,
    create_checkpoint,
@@ -836,6 +835,41 @@ class Pregel(
        if auto_validate:
            self.validate()

+    def _resolve_checkpointer(
+        self, config: RunnableConfig
+    ) -> BaseCheckpointSaver | None:
+        """The saver runs and state methods use: none for `checkpointer=False`,
+        else the one a parent lends a subgraph through the config, else this
+        graph's own."""
+        if self.checkpointer is False:
+            return None
+        conf = config.get(CONF, {})
+        if CONFIG_KEY_CHECKPOINTER in conf:
+            checkpointer = conf[CONFIG_KEY_CHECKPOINTER]
+        elif self.checkpointer is True:
+            raise RuntimeError("checkpointer=True cannot be used for root graphs.")
+        else:
+            checkpointer = self.checkpointer
+        if isinstance(checkpointer, BaseCheckpointSaver):
+            checkpointer = self._apply_checkpointer_allowlist(checkpointer)
+        return checkpointer
+
+    def _state_checkpointer(self, config: RunnableConfig) -> BaseCheckpointSaver:
+        checkpointer = self._resolve_checkpointer(ensure_config(config))
+        if not isinstance(checkpointer, BaseCheckpointSaver):
+            raise ValueError("No checkpointer set")
+        return checkpointer
+
+    def _own_checkpoint_config(self, config: RunnableConfig) -> RunnableConfig:
+        """A `checkpointer=True` subgraph keeps one history per thread, stored
+        under its namespace with the task ids removed."""
+        if self.checkpointer is not True:
+            return config
+        ns = config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "")
+        return patch_configurable(
+            config, {CONFIG_KEY_CHECKPOINT_NS: recast_checkpoint_ns(ns)}
+        )
+
    def _apply_checkpointer_allowlist(
        self, checkpointer: BaseCheckpointSaver | None
    ) -> BaseCheckpointSaver | None:
@@ -1147,7 +1181,9 @@ class Pregel(
        self,
        config: RunnableConfig,
        saved: CheckpointTuple | None,
-        recurse: BaseCheckpointSaver | None = None,
+        *,
+        saver: BaseCheckpointSaver,
+        recurse: bool = False,
        apply_pending_writes: bool = False,
    ) -> StateSnapshot:
        if not saved:
@@ -1170,9 +1206,7 @@ class Pregel(
        channels, managed = channels_from_checkpoint(
            self.channels,
            saved.checkpoint,
-            saver=self.checkpointer
-            if isinstance(self.checkpointer, BaseCheckpointSaver)
-            else None,
+            saver=saver,
            config=saved.config,
        )
        # tasks for this checkpoint
@@ -1187,11 +1221,7 @@ class Pregel(
            stop,
            for_execution=True,
            store=self.store,
-            checkpointer=(
-                self.checkpointer
-                if isinstance(self.checkpointer, BaseCheckpointSaver)
-                else None
-            ),
+            checkpointer=saver,
            manager=None,
        )
        # get the subgraphs
@@ -1218,7 +1248,7 @@ class Pregel(
                # get the state of the subgraph
                config = {
                    CONF: {
-                        CONFIG_KEY_CHECKPOINTER: recurse,
+                        CONFIG_KEY_CHECKPOINTER: saver,
                        "thread_id": saved.config[CONF]["thread_id"],
                        CONFIG_KEY_CHECKPOINT_NS: task_ns,
                    }
@@ -1270,7 +1300,9 @@ class Pregel(
        self,
        config: RunnableConfig,
        saved: CheckpointTuple | None,
-        recurse: BaseCheckpointSaver | None = None,
+        *,
+        saver: BaseCheckpointSaver,
+        recurse: bool = False,
        apply_pending_writes: bool = False,
    ) -> StateSnapshot:
        if not saved:
@@ -1293,9 +1325,7 @@ class Pregel(
        channels, managed = await achannels_from_checkpoint(
            self.channels,
            saved.checkpoint,
-            saver=self.checkpointer
-            if isinstance(self.checkpointer, BaseCheckpointSaver)
-            else None,
+            saver=saver,
            config=saved.config,
        )
        # tasks for this checkpoint
@@ -1310,11 +1340,7 @@ class Pregel(
            stop,
            for_execution=True,
            store=self.store,
-            checkpointer=(
-                self.checkpointer
-                if isinstance(self.checkpointer, BaseCheckpointSaver)
-                else None
-            ),
+            checkpointer=saver,
            manager=None,
        )
        # get the subgraphs
@@ -1341,7 +1367,7 @@ class Pregel(
                # get the state of the subgraph
                config = {
                    CONF: {
-                        CONFIG_KEY_CHECKPOINTER: recurse,
+                        CONFIG_KEY_CHECKPOINTER: saver,
                        "thread_id": saved.config[CONF]["thread_id"],
                        CONFIG_KEY_CHECKPOINT_NS: task_ns,
                    }
@@ -1394,13 +1420,7 @@ class Pregel(
        self, config: RunnableConfig, *, subgraphs: bool = False
    ) -> StateSnapshot:
        """Get the current state of the graph."""
-        checkpointer: BaseCheckpointSaver | None = ensure_config(config)[CONF].get(
-            CONFIG_KEY_CHECKPOINTER, self.checkpointer
-        )
-        if isinstance(checkpointer, BaseCheckpointSaver):
-            checkpointer = self._apply_checkpointer_allowlist(checkpointer)
-        if not checkpointer:
-            raise ValueError("No checkpointer set")
+        checkpointer = self._state_checkpointer(config)

        if (
            checkpoint_ns := config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "")
@@ -1417,11 +1437,7 @@ class Pregel(
                raise ValueError(f"Subgraph {recast} not found")

        config = merge_configs(self.config, config) if self.config else config
-        if self.checkpointer is True:
-            ns = cast(str, config[CONF][CONFIG_KEY_CHECKPOINT_NS])
-            config = merge_configs(
-                config, {CONF: {CONFIG_KEY_CHECKPOINT_NS: recast_checkpoint_ns(ns)}}
-            )
+        config = self._own_checkpoint_config(config)
        thread_id = config[CONF][CONFIG_KEY_THREAD_ID]
        if not isinstance(thread_id, str):
            config[CONF][CONFIG_KEY_THREAD_ID] = str(thread_id)
@@ -1430,7 +1446,8 @@ class Pregel(
        return self._prepare_state_snapshot(
            config,
            saved,
-            recurse=checkpointer if subgraphs else None,
+            saver=checkpointer,
+            recurse=subgraphs,
            apply_pending_writes=CONFIG_KEY_CHECKPOINT_ID not in config[CONF],
        )

@@ -1438,13 +1455,7 @@ class Pregel(
        self, config: RunnableConfig, *, subgraphs: bool = False
    ) -> StateSnapshot:
        """Get the current state of the graph."""
-        checkpointer: BaseCheckpointSaver | None = ensure_config(config)[CONF].get(
-            CONFIG_KEY_CHECKPOINTER, self.checkpointer
-        )
-        if isinstance(checkpointer, BaseCheckpointSaver):
-            checkpointer = self._apply_checkpointer_allowlist(checkpointer)
-        if not checkpointer:
-            raise ValueError("No checkpointer set")
+        checkpointer = self._state_checkpointer(config)

        if (
            checkpoint_ns := config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "")
@@ -1461,11 +1472,7 @@ class Pregel(
                raise ValueError(f"Subgraph {recast} not found")

        config = merge_configs(self.config, config) if self.config else config
-        if self.checkpointer is True:
-            ns = cast(str, config[CONF][CONFIG_KEY_CHECKPOINT_NS])
-            config = merge_configs(
-                config, {CONF: {CONFIG_KEY_CHECKPOINT_NS: recast_checkpoint_ns(ns)}}
-            )
+        config = self._own_checkpoint_config(config)
        thread_id = config[CONF][CONFIG_KEY_THREAD_ID]
        if not isinstance(thread_id, str):
            config[CONF][CONFIG_KEY_THREAD_ID] = str(thread_id)
@@ -1474,7 +1481,8 @@ class Pregel(
        return await self._aprepare_state_snapshot(
            config,
            saved,
-            recurse=checkpointer if subgraphs else None,
+            saver=checkpointer,
+            recurse=subgraphs,
            apply_pending_writes=CONFIG_KEY_CHECKPOINT_ID not in config[CONF],
        )

@@ -1488,13 +1496,7 @@ class Pregel(
    ) -> Iterator[StateSnapshot]:
        """Get the history of the state of the graph."""
        config = ensure_config(config)
-        checkpointer: BaseCheckpointSaver | None = config[CONF].get(
-            CONFIG_KEY_CHECKPOINTER, self.checkpointer
-        )
-        if isinstance(checkpointer, BaseCheckpointSaver):
-            checkpointer = self._apply_checkpointer_allowlist(checkpointer)
-        if not checkpointer:
-            raise ValueError("No checkpointer set")
+        checkpointer = self._state_checkpointer(config)

        if (
            checkpoint_ns := config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "")
@@ -1523,12 +1525,13 @@ class Pregel(
                }
            },
        )
+        config = self._own_checkpoint_config(config)
        # eagerly consume list() to avoid holding up the db cursor
        for checkpoint_tuple in list(
            checkpointer.list(config, before=before, limit=limit, filter=filter)
        ):
            yield self._prepare_state_snapshot(
-                checkpoint_tuple.config, checkpoint_tuple
+                checkpoint_tuple.config, checkpoint_tuple, saver=checkpointer
            )

    async def aget_state_history(
@@ -1541,13 +1544,7 @@ class Pregel(
    ) -> AsyncIterator[StateSnapshot]:
        """Asynchronously get the history of the state of the graph."""
        config = ensure_config(config)
-        checkpointer: BaseCheckpointSaver | None = ensure_config(config)[CONF].get(
-            CONFIG_KEY_CHECKPOINTER, self.checkpointer
-        )
-        if isinstance(checkpointer, BaseCheckpointSaver):
-            checkpointer = self._apply_checkpointer_allowlist(checkpointer)
-        if not checkpointer:
-            raise ValueError("No checkpointer set")
+        checkpointer = self._state_checkpointer(config)

        if (
            checkpoint_ns := config[CONF].get(CONFIG_KEY_CHECKPOINT_NS, "")
@@ -1577,6 +1574,7 @@ class Pregel(
                }
            },
        )
+        config = self._own_checkpoint_config(config)
        # eagerly consume list() to avoid holding up the db cursor
        for checkpoint_tuple in [
            c
@@ -1585,7 +1583,7 @@ class Pregel(
            )
        ]:
            yield await self._aprepare_state_snapshot(
-                checkpoint_tuple.config, checkpoint_tuple
+                checkpoint_tuple.config, checkpoint_tuple, saver=checkpointer
            )

    def bulk_update_state(
@@ -1609,13 +1607,7 @@ class Pregel(
            RunnableConfig: The updated config.
        """

-        checkpointer: BaseCheckpointSaver | None = ensure_config(config)[CONF].get(
-            CONFIG_KEY_CHECKPOINTER, self.checkpointer
-        )
-        if isinstance(checkpointer, BaseCheckpointSaver):
-            checkpointer = self._apply_checkpointer_allowlist(checkpointer)
-        if not checkpointer:
-            raise ValueError("No checkpointer set")
+        checkpointer = self._state_checkpointer(config)

        if len(supersteps) == 0:
            raise ValueError("No supersteps provided")
@@ -1642,7 +1634,9 @@ class Pregel(
            input_config: RunnableConfig, updates: Sequence[StateUpdate]
        ) -> RunnableConfig:
            # get last checkpoint
-            config = ensure_config(self.config, input_config)
+            config = self._own_checkpoint_config(
+                ensure_config(self.config, input_config)
+            )
            saved = checkpointer.get_tuple(config)
            if saved is not None:
                self._migrate_checkpoint(saved.checkpoint)
@@ -1667,10 +1661,7 @@ class Pregel(
            channels, managed = channels_from_checkpoint(
                self.channels,
                checkpoint,
-                saver=self.checkpointer
-                if saved is not None
-                and isinstance(self.checkpointer, BaseCheckpointSaver)
-                else None,
+                saver=checkpointer,
                config=saved.config if saved is not None else None,
            )
            values, as_node = updates[0][:2]
@@ -1682,7 +1673,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 +1695,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 +1710,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(),
@@ -1735,13 +1725,6 @@ class Pregel(
                        "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,
@@ -1760,7 +1743,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, [])],
@@ -1783,15 +1766,6 @@ class Pregel(
                            "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,
@@ -1837,12 +1811,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"
-                        },
                    },
                    {},
                )
@@ -2098,13 +2066,7 @@ class Pregel(
            RunnableConfig: The updated config.
        """

-        checkpointer: BaseCheckpointSaver | None = ensure_config(config)[CONF].get(
-            CONFIG_KEY_CHECKPOINTER, self.checkpointer
-        )
-        if isinstance(checkpointer, BaseCheckpointSaver):
-            checkpointer = self._apply_checkpointer_allowlist(checkpointer)
-        if not checkpointer:
-            raise ValueError("No checkpointer set")
+        checkpointer = self._state_checkpointer(config)

        if len(supersteps) == 0:
            raise ValueError("No supersteps provided")
@@ -2131,7 +2093,9 @@ class Pregel(
            input_config: RunnableConfig, updates: Sequence[StateUpdate]
        ) -> RunnableConfig:
            # get last checkpoint
-            config = ensure_config(self.config, input_config)
+            config = self._own_checkpoint_config(
+                ensure_config(self.config, input_config)
+            )
            saved = await checkpointer.aget_tuple(config)
            if saved is not None:
                self._migrate_checkpoint(saved.checkpoint)
@@ -2156,10 +2120,7 @@ class Pregel(
            channels, managed = await achannels_from_checkpoint(
                self.channels,
                checkpoint,
-                saver=self.checkpointer
-                if saved is not None
-                and isinstance(self.checkpointer, BaseCheckpointSaver)
-                else None,
+                saver=checkpointer,
                config=saved.config if saved is not None else None,
            )
            values, as_node = updates[0][:2]
@@ -2169,7 +2130,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 +2152,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 +2167,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(),
@@ -2222,13 +2182,6 @@ class Pregel(
                        "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"]
@@ -2246,7 +2199,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, [])],
@@ -2269,15 +2222,6 @@ class Pregel(
                            "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,
@@ -2323,12 +2267,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"
-                        },
                    },
                    {},
                )
@@ -2623,16 +2561,7 @@ class Pregel(
            stream_modes.add(print_mode)
        else:
            stream_modes.update(print_mode)
-        if self.checkpointer is False:
-            checkpointer: BaseCheckpointSaver | None = None
-        elif CONFIG_KEY_CHECKPOINTER in config.get(CONF, {}):
-            checkpointer = config[CONF][CONFIG_KEY_CHECKPOINTER]
-        elif self.checkpointer is True:
-            raise RuntimeError("checkpointer=True cannot be used for root graphs.")
-        else:
-            checkpointer = self.checkpointer
-        if isinstance(checkpointer, BaseCheckpointSaver):
-            checkpointer = self._apply_checkpointer_allowlist(checkpointer)
+        checkpointer = self._resolve_checkpointer(config)
        if checkpointer and not config.get(CONF):
            raise ValueError(
                "Checkpointer requires one or more of the following 'configurable' "
@@ -2851,9 +2780,7 @@ class Pregel(
                    "`durability` has no effect when no checkpointer is present.",
                )
            # set up subgraph checkpointing
-            if self.checkpointer is True:
-                ns = cast(str, config[CONF][CONFIG_KEY_CHECKPOINT_NS])
-                config[CONF][CONFIG_KEY_CHECKPOINT_NS] = recast_checkpoint_ns(ns)
+            config = self._own_checkpoint_config(config)
            # set up messages stream mode
            if "messages" in stream_modes:
                ns_ = cast(str | None, config[CONF].get(CONFIG_KEY_CHECKPOINT_NS))
@@ -3278,9 +3205,7 @@ class Pregel(
                    "`durability` has no effect when no checkpointer is present.",
                )
            # set up subgraph checkpointing
-            if self.checkpointer is True:
-                ns = cast(str, config[CONF][CONFIG_KEY_CHECKPOINT_NS])
-                config[CONF][CONFIG_KEY_CHECKPOINT_NS] = recast_checkpoint_ns(ns)
+            config = self._own_checkpoint_config(config)
            # set up messages stream mode
            if "messages" in stream_modes:
                # namespace can be None in a root level graph?
@@ -0,0 +1,395 @@
+import operator
+from typing import Annotated, Any, Literal
+
+import pytest
+from langgraph.checkpoint.base import BaseCheckpointSaver
+from langgraph.checkpoint.memory import InMemorySaver
+from typing_extensions import TypedDict
+
+from langgraph._internal._constants import CONFIG_KEY_CHECKPOINTER
+from langgraph.channels.delta import DeltaChannel
+from langgraph.graph import END, START, StateGraph
+from langgraph.pregel._checkpoint import (
+    achannels_from_checkpoint,
+    channels_from_checkpoint,
+    empty_checkpoint,
+)
+
+pytestmark = pytest.mark.anyio
+
+
+def _extend(state: list | None, writes: list[Any]) -> list:
+    out = list(state or [])
+    for write in writes:
+        out.extend(write if isinstance(write, list) else [write])
+    return out
+
+
+def _state_schema(snapshot_frequency: int = 1000) -> type:
+    class State(TypedDict, total=False):
+        delta: Annotated[
+            list, DeltaChannel(_extend, snapshot_frequency=snapshot_frequency)
+        ]
+        plain: Annotated[list, operator.add]
+
+    return State
+
+
+def _both(*items: str) -> dict:
+    return {"delta": list(items), "plain": list(items)}
+
+
+def _child_builder(*, snapshot_frequency: int = 1000) -> StateGraph:
+    builder = StateGraph(_state_schema(snapshot_frequency))
+    builder.add_node("a", lambda state: _both("a1"))
+    builder.add_node("b", lambda state: _both("b1", "b2"))
+    builder.add_edge(START, "a")
+    builder.add_edge("a", "b")
+    builder.add_edge("b", END)
+    return builder
+
+
+def _wrap(
+    inner: StateGraph,
+    *,
+    checkpointer: bool | None = None,
+    interrupt_before: list[str] | None = None,
+) -> StateGraph:
+    builder = StateGraph(inner.state_schema)
+    builder.add_node(
+        "child",
+        inner.compile(checkpointer=checkpointer, interrupt_before=interrupt_before),
+    )
+    builder.add_edge(START, "child")
+    builder.add_edge("child", END)
+    return builder
+
+
+def _nested_app(
+    checkpointer: BaseCheckpointSaver,
+    *,
+    depth: int = 1,
+    snapshot_frequency: int = 1000,
+    pause_before_b: bool = False,
+    subgraph_checkpointer: bool | None = None,
+) -> Any:
+    graph = _child_builder(snapshot_frequency=snapshot_frequency)
+    for _ in range(depth):
+        graph = _wrap(
+            graph,
+            checkpointer=subgraph_checkpointer,
+            interrupt_before=["b"] if pause_before_b else None,
+        )
+    return graph.compile(checkpointer=checkpointer)
+
+
+def _scoped(config: dict, namespace: str) -> dict:
+    return {"configurable": {**config["configurable"], "checkpoint_ns": namespace}}
+
+
+def _child_namespace(app: Any, config: dict, *, depth: int = 1) -> str:
+    namespace = ""
+    for level in range(depth):
+        scoped = _scoped(config, namespace) if namespace else config
+        namespace = next(
+            (
+                task.state["configurable"]["checkpoint_ns"]
+                for snapshot in app.get_state_history(scoped)
+                for task in snapshot.tasks
+                if task.name == "child" and isinstance(task.state, dict)
+            ),
+            "",
+        )
+        assert namespace, f"no `child` subgraph task at nesting level {level}"
+    return namespace
+
+
+async def _achild_namespace(app: Any, config: dict) -> str:
+    async for snapshot in app.aget_state_history(config):
+        for task in snapshot.tasks:
+            if task.name == "child" and isinstance(task.state, dict):
+                return task.state["configurable"]["checkpoint_ns"]
+    raise AssertionError("no `child` subgraph task")
+
+
+HISTORY = [_both("a1", "b1", "b2"), _both("a1"), _both(), _both()]
+
+
+def test_subgraph_get_state(sync_checkpointer: BaseCheckpointSaver) -> None:
+    app = _nested_app(sync_checkpointer)
+    config = {"configurable": {"thread_id": "1"}}
+    app.invoke({}, config)
+
+    child = _scoped(config, _child_namespace(app, config))
+
+    assert app.get_state(config).values == _both("a1", "b1", "b2")
+    assert app.get_state(child).values == _both("a1", "b1", "b2")
+
+
+async def test_subgraph_aget_state(async_checkpointer: BaseCheckpointSaver) -> None:
+    app = _nested_app(async_checkpointer)
+    config = {"configurable": {"thread_id": "1"}}
+    await app.ainvoke({}, config)
+
+    child = _scoped(config, await _achild_namespace(app, config))
+
+    assert (await app.aget_state(child)).values == _both("a1", "b1", "b2")
+
+
+def test_subgraph_get_state_history(sync_checkpointer: BaseCheckpointSaver) -> None:
+    app = _nested_app(sync_checkpointer)
+    config = {"configurable": {"thread_id": "1"}}
+    app.invoke({}, config)
+
+    child = _scoped(config, _child_namespace(app, config))
+
+    assert [s.values for s in app.get_state_history(child)] == HISTORY
+
+
+async def test_subgraph_aget_state_history(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    app = _nested_app(async_checkpointer)
+    config = {"configurable": {"thread_id": "1"}}
+    await app.ainvoke({}, config)
+
+    child = _scoped(config, await _achild_namespace(app, config))
+
+    assert [s.values async for s in app.aget_state_history(child)] == HISTORY
+
+
+def test_doubly_nested_subgraph_get_state(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    app = _nested_app(sync_checkpointer, depth=2)
+    config = {"configurable": {"thread_id": "1"}}
+    app.invoke({}, config)
+
+    child = _scoped(config, _child_namespace(app, config, depth=2))
+
+    assert app.get_state(child).values == _both("a1", "b1", "b2")
+
+
+@pytest.mark.parametrize("persistence", ["per-invocation", "per-thread"])
+def test_interrupted_subgraph_task_state(
+    sync_checkpointer: BaseCheckpointSaver,
+    persistence: Literal["per-invocation", "per-thread"],
+) -> None:
+    app = _nested_app(
+        sync_checkpointer,
+        pause_before_b=True,
+        subgraph_checkpointer=True if persistence == "per-thread" else None,
+    )
+    config = {"configurable": {"thread_id": "1"}}
+    app.invoke({}, config)
+
+    (task,) = app.get_state(config, subgraphs=True).tasks
+
+    assert task.state.values == _both("a1")
+
+
+@pytest.mark.parametrize("persistence", ["per-invocation", "per-thread"])
+async def test_interrupted_subgraph_task_state_async(
+    async_checkpointer: BaseCheckpointSaver,
+    persistence: Literal["per-invocation", "per-thread"],
+) -> None:
+    app = _nested_app(
+        async_checkpointer,
+        pause_before_b=True,
+        subgraph_checkpointer=True if persistence == "per-thread" else None,
+    )
+    config = {"configurable": {"thread_id": "1"}}
+    await app.ainvoke({}, config)
+
+    (task,) = (await app.aget_state(config, subgraphs=True)).tasks
+
+    assert task.state.values == _both("a1")
+
+
+@pytest.mark.parametrize("persistence", ["per-invocation", "per-thread"])
+def test_interrupted_subgraph_history_from_its_task_config(
+    sync_checkpointer: BaseCheckpointSaver,
+    persistence: Literal["per-invocation", "per-thread"],
+) -> None:
+    app = _nested_app(
+        sync_checkpointer,
+        pause_before_b=True,
+        subgraph_checkpointer=True if persistence == "per-thread" else None,
+    )
+    config = {"configurable": {"thread_id": "1"}}
+    app.invoke({}, config)
+    (task,) = app.get_state(config).tasks
+
+    history = list(app.get_state_history(task.state))
+
+    assert [snapshot.values for snapshot in history][:1] == [_both("a1")]
+
+
+@pytest.mark.parametrize("persistence", ["per-invocation", "per-thread"])
+async def test_interrupted_subgraph_ahistory_from_its_task_config(
+    async_checkpointer: BaseCheckpointSaver,
+    persistence: Literal["per-invocation", "per-thread"],
+) -> None:
+    app = _nested_app(
+        async_checkpointer,
+        pause_before_b=True,
+        subgraph_checkpointer=True if persistence == "per-thread" else None,
+    )
+    config = {"configurable": {"thread_id": "1"}}
+    await app.ainvoke({}, config)
+    (task,) = (await app.aget_state(config)).tasks
+
+    history = [snapshot async for snapshot in app.aget_state_history(task.state)]
+
+    assert [snapshot.values for snapshot in history][:1] == [_both("a1")]
+
+
+@pytest.mark.parametrize("persistence", ["per-invocation", "per-thread"])
+def test_interrupted_subgraph_update_state_from_its_task_config(
+    sync_checkpointer: BaseCheckpointSaver,
+    persistence: Literal["per-invocation", "per-thread"],
+) -> None:
+    app = _nested_app(
+        sync_checkpointer,
+        pause_before_b=True,
+        subgraph_checkpointer=True if persistence == "per-thread" else None,
+    )
+    config = {"configurable": {"thread_id": "1"}}
+    app.invoke({}, config)
+    (task,) = app.get_state(config).tasks
+
+    app.update_state(task.state, _both("edit"), as_node="a")
+
+    (task,) = app.get_state(config, subgraphs=True).tasks
+    assert task.state.values == _both("a1", "edit")
+
+
+@pytest.mark.parametrize("persistence", ["per-invocation", "per-thread"])
+async def test_interrupted_subgraph_aupdate_state_from_its_task_config(
+    async_checkpointer: BaseCheckpointSaver,
+    persistence: Literal["per-invocation", "per-thread"],
+) -> None:
+    app = _nested_app(
+        async_checkpointer,
+        pause_before_b=True,
+        subgraph_checkpointer=True if persistence == "per-thread" else None,
+    )
+    config = {"configurable": {"thread_id": "1"}}
+    await app.ainvoke({}, config)
+    (task,) = (await app.aget_state(config)).tasks
+
+    await app.aupdate_state(task.state, _both("edit"), as_node="a")
+
+    (task,) = (await app.aget_state(config, subgraphs=True)).tasks
+    assert task.state.values == _both("a1", "edit")
+
+
+def test_subgraph_update_state_keeps_history(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    app = _nested_app(sync_checkpointer, snapshot_frequency=2)
+    config = {"configurable": {"thread_id": "1"}}
+    app.invoke({}, config)
+
+    child = _scoped(config, _child_namespace(app, config))
+    app.update_state(child, _both("manual"))
+
+    assert app.get_state(child).values == _both("a1", "b1", "b2", "manual")
+
+
+async def test_subgraph_aupdate_state_keeps_history(
+    async_checkpointer: BaseCheckpointSaver,
+) -> None:
+    app = _nested_app(async_checkpointer, snapshot_frequency=2)
+    config = {"configurable": {"thread_id": "1"}}
+    await app.ainvoke({}, config)
+
+    child = _scoped(config, await _achild_namespace(app, config))
+    await app.aupdate_state(child, _both("manual"))
+
+    assert (await app.aget_state(child)).values == _both("a1", "b1", "b2", "manual")
+
+
+def test_stateless_subgraph_persists_nothing(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    app = _nested_app(sync_checkpointer, subgraph_checkpointer=False)
+    config = {"configurable": {"thread_id": "1"}}
+    app.invoke({}, config)
+
+    child_tasks = [
+        task
+        for snapshot in app.get_state_history(config)
+        for task in snapshot.tasks
+        if task.name == "child" and isinstance(task.state, dict)
+    ]
+
+    assert child_tasks == []
+    assert app.get_state(config).values == _both("a1", "b1", "b2")
+
+
+def test_completed_subgraph_exposes_no_task_state(
+    sync_checkpointer: BaseCheckpointSaver,
+) -> None:
+    app = _nested_app(sync_checkpointer)
+    config = {"configurable": {"thread_id": "1"}}
+    app.invoke({}, config)
+
+    assert app.get_state(config, subgraphs=True).tasks == ()
+
+
+def _written_delta_checkpoint() -> Any:
+    checkpoint = empty_checkpoint()
+    checkpoint["channel_versions"]["delta"] = 1
+    return checkpoint
+
+
+def test_hydrating_written_delta_channel_without_saver_raises() -> None:
+    with pytest.raises(ValueError, match="no checkpointer"):
+        channels_from_checkpoint(
+            {"delta": DeltaChannel(_extend)}, _written_delta_checkpoint()
+        )
+
+
+async def test_ahydrating_written_delta_channel_without_saver_raises() -> None:
+    with pytest.raises(ValueError, match="no checkpointer"):
+        await achannels_from_checkpoint(
+            {"delta": DeltaChannel(_extend)}, _written_delta_checkpoint()
+        )
+
+
+def test_hydrating_written_delta_channel_without_config_raises() -> None:
+    with pytest.raises(ValueError, match="no checkpointer"):
+        channels_from_checkpoint(
+            {"delta": DeltaChannel(_extend)},
+            _written_delta_checkpoint(),
+            saver=InMemorySaver(),
+        )
+
+
+def test_hydrating_unwritten_delta_channel_without_saver_is_empty() -> None:
+    channels, _ = channels_from_checkpoint(
+        {"delta": DeltaChannel(_extend)}, empty_checkpoint()
+    )
+    assert channels["delta"].get() == []
+
+
+def test_root_checkpointer_true_graph_state_read_raises() -> None:
+    app = _child_builder().compile(checkpointer=True)
+    with pytest.raises(RuntimeError, match="checkpointer=True cannot be used"):
+        app.get_state({"configurable": {"thread_id": "1"}})
+
+
+def test_stateless_graph_update_state_ignores_lent_saver() -> None:
+    saver = InMemorySaver()
+    app = _child_builder().compile(checkpointer=False)
+    config = {
+        "configurable": {
+            "thread_id": "1",
+            "checkpoint_ns": "child:1",
+            CONFIG_KEY_CHECKPOINTER: saver,
+        }
+    }
+    with pytest.raises(ValueError, match="No checkpointer set"):
+        app.update_state(config, _both("x"), as_node="a")
+    assert list(saver.list(None)) == []
@@ -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
@@ -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	7d601b1b82	fix(langgraph): share one saver resolution between runs and state methods `_defaults` picked the run's saver on its own, and disagreed with the state methods on `checkpointer=False`: a run ignored a lent saver while `update_state` wrote a checkpoint with it that no run reads. A `checkpointer=True` root also got `True` back from `_state_checkpointer` and failed with an AttributeError instead of the run's error. Both now go through `_resolve_checkpointer`, and the state methods require an actual saver.	2026-09-30 12:24:47 -04:00
Elior Nataf Lackritz	314cb19da8	fix(langgraph): resolve the saver and checkpointer=True namespace in one place The six state methods each resolved the saver themselves, and only get_state recast a checkpointer=True subgraph's namespace. So get_state_history and update_state given the subgraph task's config looked under `name:<task_id>`, where the run never stores anything: the history came back empty and the update was silently lost. The state methods and the run loop now share one saver lookup and one namespace rule. The hydration guard also raises when a saver comes without a config, which skipped the replay the same way.	2026-09-29 10:52:52 -04:00
Elior Nataf Lackritz	0207ec7cff	fix(langgraph): raise when a written DeltaChannel is hydrated without a saver A missing saver used to hydrate the channel as empty, silently. Raise when a DeltaChannel has a version but no stored value and no saver was passed. Keying on the version keeps a never-written channel, which has no value either, reading as empty without a saver.	2026-09-29 10:48:45 -04:00
Elior Nataf Lackritz	a4a028a254	refactor(langgraph): make _prepare_state_snapshot recurse a bool recurse only ever carried the same saver as saver, or None. Take a flag and hand saver to the child's CONFIG_KEY_CHECKPOINTER, so the snapshot helpers hold the saver in one argument.	2026-09-29 10:48:45 -04:00
Elior Nataf Lackritz gururafiki Yuan Gao	0afdb91adc	fix(langgraph): hydrate subgraph delta channels with the resolved saver A subgraph has no saver of its own: it uses the parent's, passed through `CONFIG_KEY_CHECKPOINTER`, or holds `checkpointer=True`. Every state reader resolved that saver, then called `_prepare_state_snapshot`, which re-derived it from `self.checkpointer` and got `None`. A `DeltaChannel` stores nothing in `channel_values`, so without a saver it hydrated empty while plain channels in the same snapshot read correctly. `bulk_update_state` had the same expression, and once an update reached the channel's snapshot cadence it persisted the empty value as a `_DeltaSnapshot`, losing history on disk. Pass the caller's saver in as a required keyword argument. Fixes #8470 Co-authored-by: gururafiki <22777967+gururafiki@users.noreply.github.com> Co-authored-by: Yuan Gao <119447586+DavidGao520@users.noreply.github.com>	2026-09-29 10:48:45 -04:00