From 81f4d7707f3338326194094a4b3c006a24925e72 Mon Sep 17 00:00:00 2001 From: RongJie G <111257566+CorgiBoyG@users.noreply.github.com> Date: Tue, 13 Oct 2026 08:34:05 +0800 Subject: [PATCH] fix(runtime): preserve canonical delta-history order in cache (#6774) * fix(runtime): preserve canonical delta-history order in cache CachedHistorySaver composed a parent checkpoint's channel writes from CheckpointTuple.pending_writes in storage order, but the saver's authoritative get_delta_channel_history orders writes by (task_id, idx). On the in-memory backend pending_writes follows task-arrival order, so a fan-out super-step (several parallel tasks writing one channel in a single checkpoint) was materialized in a different order with the cache enabled than with the bare saver, breaking the cache's transparency invariant. Sort a checkpoint's channel writes by task_id in _channel_writes, restoring the canonical (task_id, idx) order while preserving each task's internal idx sequence. The sort is a no-op on sqlite and postgres, whose pending_writes already arrive ORDER BY task_id, idx. Add async and sync regression tests over a real InMemorySaver seeded with a fan-out checkpoint whose writes arrive out of task_id order; both diverge from the inner saver's native walk without the fix and match with it. Signed-off-by: CorgiBoyG <111257566+CorgiBoyG@users.noreply.github.com> * docs(runtime): clarify cached saver ordering contract --------- Signed-off-by: CorgiBoyG <111257566+CorgiBoyG@users.noreply.github.com> Co-authored-by: Willem Jiang --- .../runtime/checkpointer/cached_saver.py | 27 +++++- backend/tests/test_cached_history_saver.py | 84 ++++++++++++++++++- 2 files changed, 107 insertions(+), 4 deletions(-) diff --git a/backend/packages/harness/deerflow/runtime/checkpointer/cached_saver.py b/backend/packages/harness/deerflow/runtime/checkpointer/cached_saver.py index 76c98d84d0..3408c55913 100644 --- a/backend/packages/harness/deerflow/runtime/checkpointer/cached_saver.py +++ b/backend/packages/harness/deerflow/runtime/checkpointer/cached_saver.py @@ -1,4 +1,9 @@ -"""Read-through delta-history cache wrapper for any BaseCheckpointSaver. +"""Read-through delta-history cache wrapper for a canonical BaseCheckpointSaver. + +The inner saver must return delta history in ascending ``(task_id, idx)`` +order per checkpoint, with each task's ``pending_writes`` in ``idx`` order. +The in-repo memory, sqlite, and postgres backends satisfy this contract; +the base saver's default history walk alone does not guarantee it. Correctness argument (spec §3): a checkpoint's delta history is a pure function of its sealed ancestor chain — the LangGraph contract excludes the @@ -46,8 +51,24 @@ def _checkpoint_ref(tup: CheckpointTuple) -> tuple[str, str, str]: def _channel_writes(tup: CheckpointTuple, channel: str) -> list[PendingWrite]: - """Writes for one channel, oldest→newest (tuple storage order).""" - return [w for w in (tup.pending_writes or []) if w[1] == channel] + """Writes for one channel in the inner saver's canonical delta order. + + The wrapper assumes the inner saver's ``get_delta_channel_history`` uses + ascending ``(task_id, idx)`` order per checkpoint; the base implementation + instead iterates ``pending_writes`` in storage order. For supported savers, + ``CheckpointTuple.pending_writes`` carries each task's writes in ``idx`` + order, but across tasks it follows + *storage/completion* order, which the InMemorySaver backend leaves in task + arrival order rather than ``task_id`` order. Composing from that raw order + makes the cached history disagree with the inner saver on fan-out + supersteps (several parallel tasks writing one channel). A stable sort by + ``task_id`` restores the canonical ``(task_id, idx)`` order while preserving + each task's internal ``idx`` sequence; it is a no-op on backends that + already return writes ``ORDER BY task_id, idx`` (sqlite, postgres). + """ + writes = [w for w in (tup.pending_writes or []) if w[1] == channel] + writes.sort(key=lambda w: w[0]) + return writes class CachedHistorySaver(BaseCheckpointSaver): diff --git a/backend/tests/test_cached_history_saver.py b/backend/tests/test_cached_history_saver.py index e7a47b5fbb..ad107053d0 100644 --- a/backend/tests/test_cached_history_saver.py +++ b/backend/tests/test_cached_history_saver.py @@ -1,9 +1,11 @@ """CachedHistorySaver composition vs. the saver's own full walk.""" from typing import Any +from uuid import uuid4 import pytest -from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointTuple +from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointTuple, empty_checkpoint +from langgraph.checkpoint.memory import InMemorySaver from deerflow.runtime.checkpoint_cache.memory import MemoryCheckpointHistoryCache from deerflow.runtime.checkpointer.cached_saver import CachedHistorySaver @@ -396,3 +398,83 @@ async def test_stats_expose_composition_counters(): assert stats["full_walks"] == 0 assert stats["compose_hits"] == 3 # c1-level + c2-level (cold) + c3 assert stats["hits"] >= 1 + + +# ---------------------------------------------------------------------- +# Fan-out write order: the compose path must agree with the inner saver's +# canonical (task_id, idx) order, not storage/arrival order. +# +# These use a REAL InMemorySaver (not _DictSaver): the base-class delta walk +# returns writes in stored order, but InMemorySaver's native override sorts a +# checkpoint's writes by (task_id, idx). CheckpointTuple.pending_writes, which +# the cache composes from, keeps arrival order. A fan-out super-step attaches +# several writes for one channel to a single checkpoint; when they arrive out +# of task_id order (parallel tasks finishing in any order), composing from the +# raw pending_writes diverges from the authoritative walk unless the cache +# re-establishes the (task_id, idx) order. _DictSaver cannot catch this (its +# walk is the unsorted base implementation), so the regression lives here. +# ---------------------------------------------------------------------- + +_FANOUT_CHANNEL = "messages" + + +def _seed_fanout_thread(saver: InMemorySaver) -> dict: + """Build root -> snapshot(c1, two out-of-order fan-out writes) -> c2 target. + + c1 carries ``channel_values[messages]`` (a snapshot seed) AND two writes for + ``messages`` deliberately inserted in DESCENDING task_id order (``task-b`` + before ``task-a``). Materializing c2 composes THROUGH c1, exercising + ``_channel_writes`` on c1's ``pending_writes``. Returns c2's config. + """ + thread_id = f"fanout-{uuid4().hex}" + base = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} + + cfg0 = saver.put(base, empty_checkpoint(), {}, {}) + + c1 = empty_checkpoint() + c1["channel_values"] = {_FANOUT_CHANNEL: ["seed"]} + c1["channel_versions"] = {_FANOUT_CHANNEL: "1"} + cfg1 = saver.put(cfg0, c1, {}, {_FANOUT_CHANNEL: "1"}) + # Arrival order is the reverse of task_id order on purpose. + saver.put_writes(cfg1, [(_FANOUT_CHANNEL, ["fan-b"])], "task-b") + saver.put_writes(cfg1, [(_FANOUT_CHANNEL, ["fan-a"])], "task-a") + + return saver.put(cfg1, empty_checkpoint(), {}, {}) + + +def _write_task_ids(history: dict[str, Any]) -> list[str]: + return [w[0] for w in history[_FANOUT_CHANNEL]["writes"]] + + +@pytest.mark.anyio +async def test_fanout_compose_matches_inner_canonical_order(): + """Composed fan-out history must equal the inner saver's native walk, + ordered by task_id, regardless of the writes' arrival order.""" + oracle = InMemorySaver() + oracle_cfg = _seed_fanout_thread(oracle) + expected = await oracle.aget_delta_channel_history(config=oracle_cfg, channels=[_FANOUT_CHANNEL]) + assert _write_task_ids(expected) == ["task-a", "task-b"] # canonical order + + inner = InMemorySaver() + cfg = _seed_fanout_thread(inner) + saver = _wrap(inner, MemoryCheckpointHistoryCache(max_entries=16)) + composed = await saver.aget_delta_channel_history(config=cfg, channels=[_FANOUT_CHANNEL]) + + assert _write_task_ids(composed) == _write_task_ids(expected) + assert composed[_FANOUT_CHANNEL]["writes"] == expected[_FANOUT_CHANNEL]["writes"] + + +def test_sync_fanout_compose_matches_inner_canonical_order(): + """Sync twin: the sync compose path must preserve the same canonical order.""" + oracle = InMemorySaver() + oracle_cfg = _seed_fanout_thread(oracle) + expected = oracle.get_delta_channel_history(config=oracle_cfg, channels=[_FANOUT_CHANNEL]) + assert _write_task_ids(expected) == ["task-a", "task-b"] + + inner = InMemorySaver() + cfg = _seed_fanout_thread(inner) + saver = _wrap(inner, MemoryCheckpointHistoryCache(max_entries=16)) + composed = saver.get_delta_channel_history(config=cfg, channels=[_FANOUT_CHANNEL]) + + assert _write_task_ids(composed) == _write_task_ids(expected) + assert composed[_FANOUT_CHANNEL]["writes"] == expected[_FANOUT_CHANNEL]["writes"]