From 7cf0abe58430f018c3ea025fcaca9bdc98c05408 Mon Sep 17 00:00:00 2001 From: Zeren Wang <53075619+Vanzeren@users.noreply.github.com> Date: Thu, 24 Sep 2026 09:16:08 +0200 Subject: [PATCH] test(checkpoint): delta-history cache migration coverage + pin two open DeltaChannel defects (#5797) * test(checkpoint): cover delta-history cache migration and pin two open DeltaChannel defects - Extend the full -> delta migration contract to {cache disabled, cold, warm} with a differential raw-saver oracle; the warm state also asserts a cache-served re-read with no extra inner history walks. - Pin two still-unfixed upstream DeltaChannel defects as tripwire regressions: #8382 (parallel-superstep delta replay order diverges from live order) and #8448 (Postgres paged delta walk hydrates old checkpoints empty). Each asserts the correct contract and skips, naming the issue and pinned version, while the defect is present - so a later dependency bump that lands the fix turns it into a live gate. The contract is never asserted as buggy behaviour (see backend/tests/AGENTS.md). * test(checkpoint): scope the #8448 skip to the postgres parameter The paged stage-1 walk that langgraph#8448 poisons only exists in langgraph.checkpoint.postgres; memory and sqlite never paginate. An unscoped skip let a future regression in long-chain materialization on those backends hide behind the Postgres-only upstream issue instead of failing the InMemorySaver differential oracle. Keep the assertion live there and trip only where the defect can occur. --- .../test_cached_history_saver_integration.py | 87 +++++++++++ .../tests/test_delta_channel_checkpointers.py | 142 +++++++++++++++++- 2 files changed, 228 insertions(+), 1 deletion(-) diff --git a/backend/tests/test_cached_history_saver_integration.py b/backend/tests/test_cached_history_saver_integration.py index 942f8622a0..0c3a6ba24e 100644 --- a/backend/tests/test_cached_history_saver_integration.py +++ b/backend/tests/test_cached_history_saver_integration.py @@ -27,6 +27,7 @@ from langchain_core.messages import AIMessage, AnyMessage, HumanMessage from langgraph.channels import DeltaChannel from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph +from langgraph.graph.message import add_messages from langgraph.types import Command, interrupt from deerflow.agents.thread_state import merge_message_writes @@ -362,3 +363,89 @@ async def test_rollback_supersede_does_not_pollute() -> None: oracle_reread = await oracle_accessor.aget(oracle_by_next[next_key].config) assert _digest(reread.values) == _expected_final_digest()[:expected_len] assert _digest(reread.values) == _digest(oracle_reread.values) + + +class _FullState(TypedDict): + """Pre-delta (full) thread schema: ``messages`` is a plain list channel.""" + + messages: Annotated[list[AnyMessage], add_messages] + + +def _noop(state: dict[str, Any]) -> dict[str, Any]: + return {} + + +def _build_noop_graph(saver: Any, schema: Any) -> Any: + builder = StateGraph(schema) + builder.add_node("noop", _noop) + builder.set_entry_point("noop") + builder.set_finish_point("noop") + return builder.compile(checkpointer=saver) + + +_MIGRATION_DELTA_STEPS = 3 + + +async def _migrate_full_to_delta(saver: Any, config: dict[str, Any]) -> tuple[list, list, list]: + """Seed a thread in full (pre-delta) mode, then swap to delta on it. + + Returns ``(migrated, replayed, history)``: the digest right after the mode + swap (the plain-value seed must read back intact), the head after + ``_MIGRATION_DELTA_STEPS`` delta appends, and every checkpoint's digest. + """ + full_graph = _build_noop_graph(saver, _FullState) + await full_graph.ainvoke(_input(), config) + + delta_graph = _build_noop_graph(saver, _state_schema()) + accessor = CheckpointStateAccessor.bind(delta_graph, saver, mode="delta") + migrated = _digest((await accessor.aget(config)).values) + + for i in range(_MIGRATION_DELTA_STEPS): + await delta_graph.ainvoke({"messages": [AIMessage(content=f"delta-{i}", id=f"ai-d{i}")]}, config) + + replayed = _digest((await accessor.aget(config)).values) + history = _history_digests(await accessor.ahistory(config)) + return migrated, replayed, history + + +_CACHE_STATES = ("disabled", "cold", "warm") + + +@pytest.mark.parametrize("cache_state", _CACHE_STATES) +@pytest.mark.anyio +async def test_full_to_delta_migration_under_cache_states(cache_state: str) -> None: + """The delta-history cache must not change full -> delta migration + semantics: with the cache disabled, cold, or already warm, the migrated + plain-value seed and the replayed delta chain must stay digest-identical to + the raw saver (differential oracle). + + Regression guard for the cache composing ``writes + seed`` on its own + (``CachedHistorySaver._aresolve``): a full-mode ``messages`` blob is stored + as a plain channel value, not a ``_DeltaSnapshot``, so it is exactly the + seed path the cache reconstructs without a saver walk.""" + inner = _CountingInMemorySaver() + saver = CachedHistorySaver( + inner, + MemoryCheckpointHistoryCache(0 if cache_state == "disabled" else 128), + key_prefix=f"itest-migrate-{uuid4().hex}", + ) + config = _config() + migrated, replayed, history = await _migrate_full_to_delta(saver, config) + + oracle_inner = _CountingInMemorySaver() + oracle_config = _config() + oracle_migrated, oracle_replayed, oracle_history = await _migrate_full_to_delta(oracle_inner, oracle_config) + + assert migrated == [("human", "kickoff", "h-0")] + assert migrated == oracle_migrated + assert replayed == oracle_replayed + assert history == oracle_history + + if cache_state == "warm": + # A second identical pass over the SAME saver must be served from the + # populated cache: same digests, no extra inner history walks. + accessor = CheckpointStateAccessor.bind(_build_noop_graph(saver, _state_schema()), saver, mode="delta") + walks_before = inner.history_walks + assert _digest((await accessor.aget(config)).values) == replayed + assert _history_digests(await accessor.ahistory(config)) == history + assert inner.history_walks == walks_before, "warm re-read triggered an inner walk" diff --git a/backend/tests/test_delta_channel_checkpointers.py b/backend/tests/test_delta_channel_checkpointers.py index 0c4ebfd6b8..c7c0106f2f 100644 --- a/backend/tests/test_delta_channel_checkpointers.py +++ b/backend/tests/test_delta_channel_checkpointers.py @@ -15,6 +15,15 @@ relies on: Every contract runs against InMemorySaver, AsyncSqliteSaver, and - when ``TEST_POSTGRES_URI`` is set - AsyncPostgresSaver, because each backend implements blob/version handling slightly differently. + +Two further tests pin currently-unfixed upstream defects (langgraph #8382 +parallel-superstep replay order, #8448 Postgres paginated delta walk). They +assert the correct contract and trip - skip, naming the issue - while the +defect is present, becoming live gates once a dependency bump lands a fix. The +#8448 trip is scoped to the Postgres parameter, the only backend that pages its +stage-1 scan: on memory/sqlite it stays a live differential assertion against +``InMemorySaver``, so a regression there fails instead of blaming the +Postgres-only upstream issue. """ from __future__ import annotations @@ -31,12 +40,13 @@ from langgraph.channels import DeltaChannel from langgraph.checkpoint.memory import InMemorySaver from langgraph.checkpoint.serde.types import _DeltaSnapshot from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver -from langgraph.graph import StateGraph +from langgraph.graph import END, START, StateGraph from langgraph.graph.message import add_messages from langgraph.types import Overwrite from deerflow.agents.goal_state import GoalState from deerflow.agents.thread_state import DeltaThreadState, merge_message_writes +from deerflow.config.database_config import DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY from deerflow.runtime.checkpoint_mode import CHECKPOINT_MODE_METADATA_KEY, checkpoint_tuple_uses_delta from deerflow.runtime.checkpoint_state import CheckpointStateAccessor from deerflow.runtime.goal import write_thread_goal @@ -318,3 +328,133 @@ async def test_non_delta_writers_preserve_delta_messages_and_markers(saver_env: assert checkpoint_tuple_uses_delta(latest_tuple) assert latest_tuple.metadata.get(CHECKPOINT_MODE_METADATA_KEY) == "delta" assert latest_tuple.metadata.get("run_durations", {}).get("run-1") == 7 + + +# --------------------------------------------------------------------------- +# Known-unfixed upstream DeltaChannel defects. +# +# langgraph #8382 (parallel-superstep delta replay order) and #8448 (Postgres +# paginated delta-walk cursor) are both still open against the pinned +# langgraph-checkpoint 4.2.0 / langgraph-checkpoint-postgres 3.1.2, with fixes +# (PR #8544 / PR #8556) unmerged and no released version carrying them. Each +# test below asserts the CORRECT contract; while the defect is present it trips +# (skips, naming the upstream issue) instead of failing, so a later dependency +# bump that lands the fix flips it into a live gate. The contract is never +# asserted as buggy behaviour. +# --------------------------------------------------------------------------- + +_PARALLEL_LABELS = tuple("abcdefgh") +_PAGINATION_STEPS = 12 +# Shrink the Postgres stage-1 page so a dozen checkpoints cross it. The defect +# is decided by which page the target lands on, not the absolute count; upstream +# PR #8556 makes the same substitution in its own regression rather than writing +# 1024+ real checkpoints. +_PAGINATION_PAGE_SIZE = 3 + + +class LongChainState(TypedDict): + """Delta ``messages`` at the production snapshot cadence (long walks).""" + + messages: Annotated[ + list[AnyMessage], + DeltaChannel(merge_message_writes, snapshot_frequency=DEFAULT_CHECKPOINT_SNAPSHOT_FREQUENCY), + ] + + +def _message_digest(values: dict[str, Any]) -> list[tuple[str, str, str | None]]: + return [(m.type, m.content, m.id) for m in values["messages"]] + + +def _append_node(index: int) -> Any: + def node(state: dict[str, Any]) -> dict[str, Any]: + return {"messages": [AIMessage(id=f"l{index}", content=f"long-{index}")]} + + return node + + +def _build_linear_graph(saver: Any, steps: int) -> Any: + builder = StateGraph(LongChainState) + for i in range(steps): + builder.add_node(f"l{i}", _append_node(i)) + builder.set_entry_point("l0") + for i in range(steps - 1): + builder.add_edge(f"l{i}", f"l{i + 1}") + builder.set_finish_point(f"l{steps - 1}") + return builder.compile(checkpointer=saver) + + +def _expected_long_chain(steps: int) -> list[tuple[str, str, str]]: + return [("human", "kickoff", "h0"), *[("ai", f"long-{i}", f"l{i}") for i in range(steps)]] + + +async def _long_chain_history(saver: Any, thread_id: str, steps: int) -> list[tuple[list, tuple]]: + """Materialize every checkpoint of a linear delta chain (newest first).""" + graph = _build_linear_graph(saver, steps) + config = _config(thread_id) + await graph.ainvoke({"messages": [HumanMessage(id="h0", content="kickoff")]}, config) + accessor = CheckpointStateAccessor.bind(graph, saver, mode="delta") + return [(_message_digest(s.values), tuple(s.next)) for s in await accessor.ahistory(config)] + + +@pytest.mark.anyio +async def test_long_chain_history_survives_pagination(saver_env: _SaverEnv, monkeypatch: pytest.MonkeyPatch) -> None: + """Every checkpoint of a long delta chain must materialize its own ancestor + prefix. Postgres resolves the history through a stage-1 scan paged + newest-first; when the target checkpoint is not on the first page the + unpatched cursor is derived from a not-yet-loaded parent and parks at + ``None`` for good, so old checkpoints hydrate empty (upstream #8448).""" + if saver_env.kind == "postgres": + pg_aio = pytest.importorskip("langgraph.checkpoint.postgres.aio", reason="postgres extra not installed") + monkeypatch.setattr(pg_aio, "_DELTA_PAGE_SIZE", _PAGINATION_PAGE_SIZE) + + got = await _long_chain_history(saver_env.saver, _thread_id(), _PAGINATION_STEPS) + oracle = await _long_chain_history(InMemorySaver(), _thread_id(), _PAGINATION_STEPS) + + # Reference sanity: newest-first, the head carries the whole chain, and + # only the pre-input root checkpoint is empty. + assert oracle[0] == (_expected_long_chain(_PAGINATION_STEPS), ()) + assert all(digest for digest, next_key in oracle if next_key != ("__start__",)) + + if saver_env.kind == "postgres" and got != oracle: + pytest.skip("upstream langgraph#8448 unfixed in langgraph-checkpoint-postgres 3.1.2: the paged delta walk poisons the channel cursor for a target past the first stage-1 page, hydrating old checkpoints empty") + assert got == oracle + + +def _fanout_node(label: str) -> Any: + def node(state: dict[str, Any]) -> dict[str, Any]: + return {"messages": [AIMessage(id=f"fan-{label}", content=label)]} + + return node + + +def _build_fanout_graph(saver: Any) -> Any: + builder = StateGraph(LongChainState) + for label in _PARALLEL_LABELS: + builder.add_node(label, _fanout_node(label)) + builder.add_edge(START, label) + builder.add_edge(label, END) + return builder.compile(checkpointer=saver) + + +@pytest.mark.anyio +async def test_parallel_superstep_replay_matches_live_write_order(saver_env: _SaverEnv) -> None: + """When several parallel tasks write the same delta channel in one + superstep, the live ``apply_writes`` order is path-sorted, but the delta + history replay sorts by the hashed ``(task_id, idx)`` - an unrelated order + (upstream #8382). ``get_state`` on the same thread must report the order the + run actually produced.""" + graph = _build_fanout_graph(saver_env.saver) + config = _config(_thread_id()) + live = await graph.ainvoke({"messages": []}, config) + live_order = [m.content for m in live["messages"]] + + # Live superstep writes are deterministic and path-sorted, so the fan-out + # lands in label order regardless of task completion order. + assert live_order == list(_PARALLEL_LABELS) + + replayed = await graph.aget_state(config) + replayed_order = [m.content for m in replayed.values["messages"]] + + if replayed_order != live_order: + pytest.skip("upstream langgraph#8382 unfixed in langgraph-checkpoint 4.2.0: delta replay orders writes by hashed (task_id, idx) instead of the path order apply_writes used live, so get_state/resume reorders same-superstep writes") + assert replayed_order == live_order