From 3c53ce02216f9bfcb3cbafd2e2172959abe6ff17 Mon Sep 17 00:00:00 2001 From: Elior Nataf Lackritz Date: Wed, 5 Aug 2026 23:05:51 -0400 Subject: [PATCH] fix(checkpoint-postgres): derive the delta walk cursor once the target loads `get_delta_channel_history` pages the `checkpoints` table newest-first from the head of the thread, so the target checkpoint's own row does not land until paging reaches back to it. `_try_advance_walks` read the walk cursor out of the partial `parent_of` map on the first page regardless, and `parent_of.get(target_id)` returns `None` there for a target that is simply not loaded yet, which is the same value that means "the target is a root". The cursor is derived under a `ch not in walk_cursor_by_ch` guard, so it is never recomputed. A target older than one page (1024 checkpoints) kept that `None` for the rest of the run: the chain stayed empty, the channel never seeded, and the history came back with no seed and no writes. Downstream `channels_from_checkpoint` hydrates that as an empty channel, so `get_state`, `get_state_history` and `update_state` against an older checkpoint returned an empty `DeltaChannel` on a thread with real accumulated history, silently and without an error. Start the walk only once `target_id` is actually in `parent_of`, so "not loaded yet" and "is a root" stop sharing a representation. Both savers share this method, so both are covered. Fixes #8448 Co-authored-by: Navneet-Scaler <147032454+Navneet-Scaler@users.noreply.github.com> --- .../langgraph/checkpoint/postgres/base.py | 7 +- .../tests/test_delta_pagination.py | 132 ++++++++++++++++++ 2 files changed, 137 insertions(+), 2 deletions(-) create mode 100644 libs/checkpoint-postgres/tests/test_delta_pagination.py diff --git a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py index d58d2cc38..607c6d58d 100644 --- a/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py +++ b/libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py @@ -441,9 +441,12 @@ class BasePostgresSaver(BaseCheckpointSaver[str]): for i, ch in enumerate(channels): if ch in seeded: continue - # First-time entry: cursor starts at the target's parent. + # Pages start at the thread head, so the target may not have + # loaded yet; a `None` cursor would read as "target is a root". if ch not in walk_cursor_by_ch: - walk_cursor_by_ch[ch] = parent_of.get(target_id) + if target_id not in parent_of: + continue + walk_cursor_by_ch[ch] = parent_of[target_id] cur_cid = walk_cursor_by_ch[ch] ch_chain = chain_by_ch[ch] hb_i = hb_by_i_by_cid[i] diff --git a/libs/checkpoint-postgres/tests/test_delta_pagination.py b/libs/checkpoint-postgres/tests/test_delta_pagination.py new file mode 100644 index 000000000..6f6cfeaff --- /dev/null +++ b/libs/checkpoint-postgres/tests/test_delta_pagination.py @@ -0,0 +1,132 @@ +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', '')!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": []}