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 <willem.jiang@gmail.com>

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):

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