fix(checkpoints): isolate cached history payloads from state readers (#7259)

* fix(checkpoints): isolate cached history payloads from state readers

Copy nested snapshot seeds and pending-write payloads at both cache boundaries.
Keep user copy callbacks outside the registry lock so readers cannot poison
later checkpoint materialization or deadlock by re-entering the cache.

Cover sync/async state accessor reads, nested message payloads, disabled-cache
controls and callback lock ownership.

* docs(checkpoints): document synchronous history copy costs

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -564,7 +564,8 @@ caches materialized checkpoint histories: `type` is `memory` (default) or
 Gateway/async-only; the sync TUI/embedded path supports `memory` only. The
 cache is performance-only — results are identical with it disabled — so it is
 never frozen and workers sharing one checkpoint database may safely run
-different cache settings. Redis thread cleanup treats a configured `key_prefix`
+different cache settings. Returned state snapshots have independent nested values,
+so editing one cannot change later checkpoint reads. Redis thread cleanup treats a configured `key_prefix`
 as literal text, including glob metacharacters, so deletion stays within that
 namespace and thread.
 

backend/packages/harness/deerflow/runtime/AGENTS.md

@@ -4,6 +4,19 @@ The memory history cache is shared by sync saver wrappers. Its threading lock
 serializes lookup/LRU promotion, population/eviction, thread purge, close, and
 statistics snapshots. Keep cache critical sections free of awaits and saver I/O;
 immutable histories do not make the mutable LRU registry thread-safe.
+Memory entries own deep copies of write payloads and snapshot seeds on ingest
+and read. Clone payloads outside the registry lock: user-defined copy callbacks
+can re-enter the cache. Preserve sync/async state isolation through
+`CheckpointStateAccessor`; tests live in `tests/test_checkpoint_cache_payload_isolation.py`.
+
+Copies run synchronously on the caller's thread, including `aget_many` and
+`aset_many` on the event loop. Each ingest or cache hit copies the whole
+entry, so CPU and allocation costs grow with payload size. A warm scan of
+N checkpoints can repeatedly copy growing snapshot seeds, reaching O(N²)
+total copied data. The LRU entry cap bounds entry count, not payload bytes
+or scan cost. Profile long-thread history reads before offloading copies
+to worker threads; preserve payload isolation and copy callbacks outside
+the registry lock.
 
 ### Redis Checkpoint Cache Purge
 

backend/packages/harness/deerflow/runtime/checkpoint_cache/memory.py

@@ -4,17 +4,15 @@ from __future__ import annotations
 
 import threading
 from collections import OrderedDict
+from copy import deepcopy
 from typing import Any
 
 from deerflow.runtime.checkpoint_cache.base import CheckpointCacheStats, thread_key_stem
 
 
 def _copy_entry(entry: dict[str, Any]) -> dict[str, Any]:
-    """Copy-on-read/write: fresh writes list; seed shared (never mutated in place)."""
-    copied: dict[str, Any] = {"writes": list(entry["writes"])}
-    if "seed" in entry:
-        copied["seed"] = entry["seed"]
-    return copied
+    """Own nested write payloads and snapshot seeds on both sides of the cache."""
+    return deepcopy(entry)
 
 
 class MemoryCheckpointHistoryCache:
@@ -42,15 +40,18 @@ class MemoryCheckpointHistoryCache:
                     continue
                 self._data.move_to_end(key)
                 self._hits += 1
-                found[key] = _copy_entry(entry)
-            return found
+                found[key] = entry
+        # Entries are privately owned and never changed after publication. Copy
+        # outside the registry lock: user payloads can implement __deepcopy__.
+        return {key: _copy_entry(entry) for key, entry in found.items()}
 
     def set_many(self, entries: dict[str, dict[str, Any]]) -> None:
+        if not self.enabled:
+            return
+        copied = {key: _copy_entry(entry) for key, entry in entries.items()}
         with self._lock:
-            if not self.enabled:
-                return
-            for key, entry in entries.items():
-                self._data[key] = _copy_entry(entry)
+            for key, entry in copied.items():
+                self._data[key] = entry
                 self._data.move_to_end(key)
                 while len(self._data) > self._max_entries:
                     self._data.popitem(last=False)

backend/tests/test_checkpoint_cache_payload_isolation.py

@@ -0,0 +1,116 @@
+"""Returned checkpoint snapshots must never mutate cached history payloads."""
+
+from copy import deepcopy
+from typing import Annotated, TypedDict
+
+import pytest
+from langgraph.channels import DeltaChannel
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.graph import StateGraph
+
+from deerflow.runtime.checkpoint_cache.memory import MemoryCheckpointHistoryCache
+from deerflow.runtime.checkpoint_state import CheckpointStateAccessor
+from deerflow.runtime.checkpointer.cached_saver import CachedHistorySaver
+
+
+def _fold(state, writes):
+    result = list(state)
+    for batch in writes:
+        result.extend(batch)
+    return result
+
+
+def _stack(frequency, enabled):
+    class State(TypedDict):
+        items: Annotated[list[dict], DeltaChannel(_fold, snapshot_frequency=frequency)]
+
+    inner = InMemorySaver()
+    cache = MemoryCheckpointHistoryCache(max_entries=32 if enabled else 0)
+    saver = CachedHistorySaver(inner, cache, key_prefix="payload-isolation")
+    builder = StateGraph(State)
+    for step in range(4):
+        builder.add_node(str(step), lambda state: {"items": [{"nested": {"value": "saved"}}]})
+        if step:
+            builder.add_edge(str(step - 1), str(step))
+    builder.set_entry_point("0")
+    builder.set_finish_point("3")
+    graph = builder.compile(checkpointer=saver)
+    return graph, CheckpointStateAccessor.bind(graph, saver, mode="delta"), cache
+
+
+def _mutate(snapshot):
+    original = deepcopy(snapshot.values)
+    snapshot.values["items"][0]["nested"]["value"] = "tampered"
+    snapshot.values["items"].append({"nested": {"value": "injected"}})
+    return original
+
+
+@pytest.mark.parametrize("frequency", [2, 100], ids=["snapshot-seed", "pending-writes"])
+@pytest.mark.parametrize("enabled", [True, False], ids=["cached", "uncached-control"])
+def test_sync_graph_snapshots_are_isolated(frequency, enabled):
+    graph, accessor, cache = _stack(frequency, enabled)
+    config = {"configurable": {"thread_id": "sync-payload-isolation"}}
+    graph.invoke({"items": [{"nested": {"value": "input"}}]}, config)
+    history = accessor.history(config)
+    checked = 0
+    for checkpoint in history:
+        snapshot = accessor.get(checkpoint.config)
+        if not snapshot.values.get("items"):
+            continue
+        original = _mutate(snapshot)
+        assert accessor.get(checkpoint.config).values == original
+        checked += 1
+    assert checked >= 4
+    assert cache.stats().hits > 0 if enabled else cache.stats().entries == 0
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("frequency", [2, 100], ids=["snapshot-seed", "pending-writes"])
+@pytest.mark.parametrize("enabled", [True, False], ids=["cached", "uncached-control"])
+async def test_async_graph_snapshots_are_isolated(frequency, enabled):
+    graph, accessor, cache = _stack(frequency, enabled)
+    config = {"configurable": {"thread_id": "async-payload-isolation"}}
+    await graph.ainvoke({"items": [{"nested": {"value": "input"}}]}, config)
+    history = await accessor.ahistory(config)
+    checked = 0
+    for checkpoint in history:
+        snapshot = await accessor.aget(checkpoint.config)
+        if not snapshot.values.get("items"):
+            continue
+        original = _mutate(snapshot)
+        assert (await accessor.aget(checkpoint.config)).values == original
+        checked += 1
+    assert checked >= 4
+    assert cache.stats().hits > 0 if enabled else cache.stats().entries == 0
+
+
+@pytest.mark.parametrize("direction", ["ingest", "read"])
+def test_cache_copies_nested_write_payloads_and_snapshot_seeds(direction):
+    from langchain_core.messages import HumanMessage
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+    cache = MemoryCheckpointHistoryCache()
+    seed = _DeltaSnapshot([HumanMessage(content="seed", id="seed", additional_kwargs={"nested": ["original"]})])
+    entry = {"seed": seed, "writes": [("task", "messages", [HumanMessage(content="write", id="write", additional_kwargs={"nested": ["original"]})])]}
+    original = deepcopy(entry)
+    cache.set_many({"key": entry})
+    exposed = entry if direction == "ingest" else cache.get_many(["key"])["key"]
+    exposed["seed"].value[0].additional_kwargs["nested"].append("corrupted")
+    exposed["writes"][0][2][0].content = "corrupted"
+    assert cache.get_many(["key"])["key"] == original
+
+
+def test_payload_copying_runs_outside_registry_lock():
+    cache = MemoryCheckpointHistoryCache()
+    copies = []
+
+    class Probe:
+        def __deepcopy__(self, memo):
+            assert cache._lock.acquire(blocking=False), "copy callback ran under the cache lock"
+            cache._lock.release()
+            copies.append(True)
+            return Probe()
+
+    cache.set_many({"key": {"writes": [("task", "channel", Probe())]}})
+    cache.get_many(["key"])
+    assert len(copies) == 2