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.

.github/workflows/release.yml

@@ -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

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

@@ -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

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

@@ -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]

libs/checkpoint-postgres/tests/test_delta_pagination.py

@@ -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": []}

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

@@ -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,

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

@@ -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
 
 

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

@@ -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,

libs/checkpoint-sqlite/tests/test_delta_parent_walk.py

@@ -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}"

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -226,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,
@@ -256,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(
@@ -298,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(

libs/langgraph/langgraph/pregel/main.py

@@ -835,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:
@@ -1146,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:
@@ -1169,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
@@ -1186,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
@@ -1217,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,
                     }
@@ -1269,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:
@@ -1292,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
@@ -1309,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
@@ -1340,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,
                     }
@@ -1393,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, "")
@@ -1416,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)
@@ -1429,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],
         )
 
@@ -1437,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, "")
@@ -1460,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)
@@ -1473,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],
         )
 
@@ -1487,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, "")
@@ -1522,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(
@@ -1540,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, "")
@@ -1576,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
@@ -1584,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(
@@ -1608,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")
@@ -1641,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)
@@ -1666,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]
@@ -1950,7 +1942,7 @@ class Pregel(
             run_tasks: list[PregelTaskWrites] = []
             run_task_ids: list[str] = []
 
-            for i, (as_node, values, provided_task_id) in enumerate(valid_updates):
+            for as_node, values, provided_task_id in valid_updates:
                 # create task to run all writers of the chosen node
                 writers = self.nodes[as_node].flat_writers
                 if not writers:
@@ -1964,7 +1956,7 @@ class Pregel(
                 task_id = provided_task_id or (
                     prepared_task_ids.popleft()
                     if prepared_task_ids
-                    else _update_task_id(checkpoint["id"], i)
+                    else str(uuid5(UUID(checkpoint["id"]), INTERRUPT))
                 )
                 run_tasks.append(task)
                 run_task_ids.append(task_id)
@@ -2074,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")
@@ -2107,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)
@@ -2132,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]
@@ -2410,7 +2395,7 @@ class Pregel(
             run_tasks: list[PregelTaskWrites] = []
             run_task_ids: list[str] = []
 
-            for i, (as_node, values, provided_task_id) in enumerate(valid_updates):
+            for as_node, values, provided_task_id in valid_updates:
                 # create task to run all writers of the chosen node
                 writers = self.nodes[as_node].flat_writers
                 if not writers:
@@ -2424,7 +2409,7 @@ class Pregel(
                 task_id = provided_task_id or (
                     prepared_task_ids.popleft()
                     if prepared_task_ids
-                    else _update_task_id(checkpoint["id"], i)
+                    else str(uuid5(UUID(checkpoint["id"]), INTERRUPT))
                 )
                 run_tasks.append(task)
                 run_task_ids.append(task_id)
@@ -2576,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' "
@@ -2804,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))
@@ -3231,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?
@@ -4172,16 +4144,6 @@ class Pregel(
         await self.cache.aclear(namespaces)
 
 
-def _update_task_id(checkpoint_id: str, i: int) -> str:
-    """Task id for the `i`th update of a superstep that has no task to reuse.
-
-    Savers keep one write per `(task_id, idx)`, so updates sharing an id lose
-    all but the first one's writes, which a `DeltaChannel` replays from. The
-    first update keeps the id a lone update has always had.
-    """
-    return str(uuid5(UUID(checkpoint_id), INTERRUPT if i == 0 else f"{INTERRUPT}:{i}"))
-
-
 def _trigger_to_nodes(nodes: dict[str, PregelNode]) -> Mapping[str, Sequence[str]]:
     """Index from a trigger to nodes that depend on it."""
     trigger_to_nodes: defaultdict[str, list[str]] = defaultdict(list)

libs/langgraph/tests/test_delta_channel_subgraph.py

@@ -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)) == []

libs/langgraph/tests/test_delta_channel_update_state.py

@@ -20,7 +20,6 @@ from typing import Annotated, Any
 
 import pytest
 from langchain_core.messages import HumanMessage
-from langgraph.checkpoint.base import BaseCheckpointSaver
 from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import TypedDict
@@ -28,17 +27,16 @@ from typing_extensions import TypedDict
 from langgraph.channels.delta import DeltaChannel
 from langgraph.graph import START, StateGraph
 from langgraph.graph.message import _messages_delta_reducer
-from langgraph.types import StateSnapshot, StateUpdate
+from langgraph.types import StateUpdate
 
 pytestmark = pytest.mark.anyio
 
 
 def _build_graph(
-    checkpointer: BaseCheckpointSaver,
+    checkpointer: InMemorySaver,
     *,
     two_nodes: bool = False,
     snapshot_frequency: int = 1000,
-    interrupt_before: list[str] | None = None,
 ) -> Any:
     """Compile a minimal DeltaChannel-backed `messages` graph.
 
@@ -65,7 +63,7 @@ def _build_graph(
         builder.set_finish_point("assistant")
     else:
         builder.set_finish_point("model")
-    return builder.compile(checkpointer=checkpointer, interrupt_before=interrupt_before)
+    return builder.compile(checkpointer=checkpointer)
 
 
 # ---------------------------------------------------------------------------
@@ -275,6 +273,10 @@ def test_bulk_update_state_multi_task_per_superstep_delta_channel() -> None:
     that each call `put_writes`. Guards the regression where moving
     `put_writes` outside the per-task loop would persist only the last
     task's writes.
+
+    Explicit `task_id`s are required to disambiguate writes belonging to
+    different `StateUpdate`s targeting the same node — otherwise both share
+    the deterministic interrupt-derived id and collide in the saver.
     """
 
     saver = InMemorySaver()
@@ -308,92 +310,6 @@ def test_bulk_update_state_multi_task_per_superstep_delta_channel() -> None:
     assert sorted(ids) == ["m1", "m2"]
 
 
-def _update(content: str, as_node: str) -> StateUpdate:
-    return StateUpdate(
-        values={"messages": [HumanMessage(content=content, id=content)]},
-        as_node=as_node,
-    )
-
-
-def _contents(state: StateSnapshot) -> list[str]:
-    return [m.content for m in state.values["messages"]]
-
-
-def test_bulk_update_state_keeps_every_update_without_task_ids(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    graph = _build_graph(sync_checkpointer, two_nodes=True)
-    config = {"configurable": {"thread_id": "bulk-no-task-ids"}}
-    graph.invoke({"messages": [HumanMessage(content="hi", id="hi")]}, config)
-
-    graph.bulk_update_state(
-        config,
-        [
-            [
-                _update("first", "model"),
-                _update("second", "model"),
-                _update("third", "assistant"),
-            ]
-        ],
-    )
-
-    contents = _contents(graph.get_state(config))
-    assert sorted(contents) == ["first", "hi", "second", "third"], (
-        f"every update's writes must persist; got {contents}"
-    )
-
-
-async def test_abulk_update_state_keeps_every_update_without_task_ids(
-    async_checkpointer: BaseCheckpointSaver,
-) -> None:
-    graph = _build_graph(async_checkpointer, two_nodes=True)
-    config = {"configurable": {"thread_id": "bulk-no-task-ids"}}
-    await graph.ainvoke({"messages": [HumanMessage(content="hi", id="hi")]}, config)
-
-    await graph.abulk_update_state(
-        config,
-        [
-            [
-                _update("first", "model"),
-                _update("second", "model"),
-                _update("third", "assistant"),
-            ]
-        ],
-    )
-
-    contents = _contents(await graph.aget_state(config))
-    assert sorted(contents) == ["first", "hi", "second", "third"], (
-        f"every update's writes must persist; got {contents}"
-    )
-
-
-def test_bulk_update_state_keeps_every_update_next_to_a_pending_task(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    graph = _build_graph(
-        sync_checkpointer, two_nodes=True, interrupt_before=["assistant"]
-    )
-    config = {"configurable": {"thread_id": "bulk-pending-task"}}
-    graph.invoke({"messages": [HumanMessage(content="hi", id="hi")]}, config)
-    assert graph.get_state(config).next == ("assistant",)
-
-    graph.bulk_update_state(
-        config,
-        [
-            [
-                _update("first", "assistant"),
-                _update("second", "model"),
-                _update("third", "model"),
-            ]
-        ],
-    )
-
-    contents = _contents(graph.get_state(config))
-    assert sorted(contents) == ["first", "hi", "second", "third"], (
-        f"every update's writes must persist; got {contents}"
-    )
-
-
 # ---------------------------------------------------------------------------
 # Public-API observation of fresh-thread checkpoint shape
 # ---------------------------------------------------------------------------