fix(checkpoints): validate cached delta history before replay (#7548)

* fix(checkpoints): reject malformed Redis history write records

Validate pending-write tuples before optional Redis history entries reach
delta materialization. Treat the whole malformed entry as a miss so the
authoritative saver rebuilds it without dropping or inventing state.

Cover actual checkpoint reads, whole-entry replacement, valid tuple/list
records, neighbor preservation and warm-cache rereads.

* fix(checkpoints): bind cached history writes to their channel

* fix(checkpoints): keep cache warning failures best-effort

Guard warning handlers across decode, ingestion, serialization, Redis outages and purge. Preserve authoritative rebuilds and valid neighboring writes, with native graph regression coverage.

* docs(gateway): keep guidance below the hard size budget

---------

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

README.md

@@ -571,7 +571,12 @@ never frozen and workers sharing one checkpoint database may safely run
 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.
+namespace and thread. Redis rejects malformed write records on ingestion; malformed
+or wrong-channel cached histories are rebuilt in full from authoritative checkpoints.
+Seeds and write values remain opaque to the cache: these checks do not validate
+schema-specific state contents.
+Checkpoint reads, valid neighboring writes and optional-cache cleanup continue
+even if a warning handler fails.
 
 Gateway checkpoint cleanup runs when the next run wins durable same-thread
 admission, before its first checkpoint read. It checks only the root head and

backend/app/gateway/AGENTS.md

@@ -10,13 +10,13 @@ environment keys; reject arbitrary commands and manifest-policy bypasses. Keep
 credentials out of tool schemas and preserve admin checks, masked atomic edits,
 and MCP cache reloads.
 
-Memory shutdown resolves hot-reloaded config/backend, flushes, then closes as
-one `await_drained` operation. Resolve config inside the best-effort handler and
-off-loop so malformed edits cannot abort teardown. Cancellation waits for owned
-workers; `close()` must be quick or internally bounded because the host sets no
-timeout. Budget it, the flush timeout, and other hooks in the pod grace period.
-Likewise drain extension `stop_services()` across cancellation before later
-resources unwind; each service `stop()` has a 30-second `asyncio.timeout`.
+Memory shutdown resolves hot-reloaded config/backend, flushes and closes in one
+`await_drained` operation. Resolve config off-loop inside the best-effort handler
+so malformed edits cannot abort teardown. Cancellation waits for owned workers.
+The host sets no close timeout: `close()` must be quick or internally bounded;
+budget it, flush timeout and other hooks within pod grace. Drain extension
+`stop_services()` across cancellation before later resources unwind; each
+service `stop()` has a 30-second `asyncio.timeout`.
 
 Routers with persistent writes drain through
 `app.gateway.persistent_writes.run_drained_write`.
@@ -43,13 +43,12 @@ locations and keeps `merge_run_context_overrides` clear. Admission and
 `/api/features` use `conversation_references_enabled()` as the sole configured
 predicate.
 
-FastAPI listens on port 8001. `GET /health` is liveness; `GET /health/ready`
-concurrently probes the ORM engine, the checkpointer/Store and the stream
-bridge backend (`StreamBridge.ping()`) under one deadline and per-kind gates,
-from startup snapshots. Unreachable or unresolved gating backends return 503;
-`memory` reports `not_configured`; the provisioner `/health` verdict is
-report-only. See `health.py`. `GATEWAY_ENABLE_DOCS=false` disables `/docs`,
-`/redoc`, and `/openapi.json`.
+FastAPI listens on 8001. `GET /health` is liveness; `GET /health/ready` concurrently
+probes startup snapshots of ORM, checkpointer/Store and stream bridge
+(`StreamBridge.ping()`) under one deadline and per-kind gates. Unreachable or
+unresolved gating backends return 503; `memory` reports `not_configured`;
+provisioner `/health` is report-only. See `health.py`. `GATEWAY_ENABLE_DOCS=false`
+disables `/docs`, `/redoc` and `/openapi.json`.
 
 `build_run_config()` resolves the default LangGraph super-step budget from the
 hot-reloaded top-level `recursion_limit` setting. A valid request-level value
@@ -58,7 +57,12 @@ takes precedence; invalid values fall back to that default, and
 
 Durable MCP notifications use internal Agent runs. Keep their trusted delivery instruction outside the user-input boundary, and frame serialized remote events as untrusted before model invocation. Strict thread existence/ownership admission dead-letters events whose task outlives its deleted chat instead of recreating the thread. `start_run` marks run-manager admission conflicts with `BusyThreadConflict`; the notification launcher retries only that typed 409 and treats other 409 responses as permanent.
 
-CORS is same-origin by default when requests enter through nginx on port 2026. Split-origin or port-forwarded browser clients must opt in with `GATEWAY_CORS_ORIGINS` (exact origins); Gateway `CORSMiddleware` and `CSRFMiddleware` both read that variable so browser CORS and auth-origin checks stay aligned. Those clients also need `CORS_EXPOSED_HEADERS` (`csrf_middleware.py`): run-creating routes return the run's id in `Content-Location`, which is not CORS-safelisted, so JS cannot read it unless it is exposed — and the LangGraph SDK resolves run metadata from that header alone, so withholding it breaks `useStream`'s `onCreated` and thread-gated actions.
+CORS defaults to same-origin behind nginx on 2026. Split-origin and port-forwarded
+browser clients must set exact `GATEWAY_CORS_ORIGINS`, shared by `CORSMiddleware`
+and `CSRFMiddleware`. Expose `Content-Location` through `CORS_EXPOSED_HEADERS`
+(`csrf_middleware.py`): run-creating routes put the run ID there, and the LangGraph
+SDK uses only that header for run metadata. Without exposure, JS cannot read it,
+breaking `useStream.onCreated` and thread-gated actions.
 
 Browser sessions belong to `app.gateway.auth.session_cookie`. Login's `remember_me`
 flag never stores passwords. `SessionCookiePolicy` persists `HttpOnly access_token`

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

@@ -1,29 +1,17 @@
-### Checkpoint Cache Concurrency
+### Checkpoint History Cache
 
-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`.
+Lock registry/LRU/stats/lifecycle without awaits, saver I/O or copy callbacks.
+Deep-copy writes/seeds on caller-thread ingest/read (async too): O(payload)
+CPU/allocation per entry, O(N²) across N growing seeds. LRU bounds entries, not
+bytes/scan cost; profile before offload. Isolation:
+`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
-
-Thread purge escapes Redis glob metacharacters in the complete literal key stem
-before SCAN, and checks returned byte/string keys against that stem before UNLINK.
-Do not interpret operator `key_prefix` text as a pattern or widen thread deletion
-to sibling namespaces. Keep the cursor loop and outage/TTL fallback unchanged.
+Redis read/write records require tuple/list triples and string task/channel IDs;
+skip bad ingests, rebuild bad reads. Bind sync/async target/ancestor hits to the
+requested channel; seeds/values stay opaque. Warnings are best-effort: handler
+errors preserve cache fallbacks and valid neighbors, including Redis outages/purge.
+Escape literal stems for SCAN; recheck byte/string keys before UNLINK. Preserve
+namespace/thread bounds, cursor iteration and outage/TTL fallback.
 
 ### Stream Bridge Heartbeats
 

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

@@ -23,6 +23,13 @@ from typing import Any, Protocol
 CACHE_FORMAT_VERSION = 1
 
 
+def valid_history_entry(entry: Any, *, channel: str | None = None) -> bool:
+    """Check replay records; seed and write values belong to the channel schema."""
+    if not isinstance(entry, dict) or not isinstance(entry.get("writes"), list):
+        return False
+    return all(isinstance(write, (tuple, list)) and len(write) == 3 and isinstance(write[0], str) and isinstance(write[1], str) and (channel is None or write[1] == channel) for write in entry["writes"])
+
+
 def make_history_key(
     key_prefix: str,
     thread_id: str,

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

@@ -12,9 +12,10 @@ The redis import is lazy (module is importable without the optional
 from __future__ import annotations
 
 import logging
+from contextlib import suppress
 from typing import Any
 
-from deerflow.runtime.checkpoint_cache.base import CheckpointCacheStats, thread_key_stem
+from deerflow.runtime.checkpoint_cache.base import CheckpointCacheStats, thread_key_stem, valid_history_entry
 
 logger = logging.getLogger(__name__)
 
@@ -23,6 +24,12 @@ REDIS_INSTALL = "redis is required for the redis checkpoint cache backend. Insta
 _TAG_SEPARATOR = b"\x00"
 
 
+def _warn(message: str, *args: Any) -> None:
+    """Keep optional-cache diagnostics from changing fallback behavior."""
+    with suppress(Exception):
+        logger.warning(message, *args)
+
+
 def _create_client(redis_url: str, *, max_connections: int | None) -> Any:
     try:
         import redis.asyncio as redis_async
@@ -68,7 +75,7 @@ class RedisCheckpointHistoryCache:
             raws = await self._client.mget(keys)
         except _redis_error() as exc:
             # Performance-only bypass: a redis outage costs hits, never availability.
-            logger.warning("checkpoint history cache mget failed; treating as all-miss: %s", exc)
+            _warn("checkpoint history cache mget failed; treating as all-miss: %s", exc)
             self._misses += len(keys)
             return {}
         found: dict[str, dict[str, Any]] = {}
@@ -79,12 +86,15 @@ class RedisCheckpointHistoryCache:
             try:
                 tag, payload = raw.split(_TAG_SEPARATOR, 1)
                 entry = self._serde.loads_typed((tag.decode(), payload))
-                if not isinstance(entry, dict) or not isinstance(entry.get("writes"), list):
-                    raise ValueError("invalid checkpoint history entry shape")
+                # Delta materialization unpacks each pending write as a triple.
+                # A three-character string also unpacks, silently inventing a
+                # channel value, so validate the record rather than just len().
+                if not valid_history_entry(entry):
+                    raise ValueError("invalid checkpoint history entry or write record")
             except Exception as exc:
                 # Cached history is optional. A malformed entry must not prevent
                 # rebuilding history from authoritative checkpoints.
-                logger.warning(
+                _warn(
                     "checkpoint history cache decode failed; treating as miss (key=%s): %s",
                     key,
                     type(exc).__name__,
@@ -101,13 +111,16 @@ class RedisCheckpointHistoryCache:
         try:
             pipe = self._client.pipeline(transaction=False)
             for key, entry in entries.items():
+                if not valid_history_entry(entry):
+                    _warn("checkpoint history cache invalid entry; skipping (key=%s)", key)
+                    continue
                 try:
                     tag, data = self._serde.dumps_typed(entry)
                 except Exception as exc:
                     self._write_serialization_failures += 1
                     # Cache writes are optional; one unserializable entry must not
                     # abort authoritative history reads or discard other entries.
-                    logger.warning(
+                    _warn(
                         "checkpoint history cache serialization failed; skipping (key=%s): %s",
                         key,
                         type(exc).__name__,
@@ -117,7 +130,7 @@ class RedisCheckpointHistoryCache:
             await pipe.execute()
         except _redis_error() as exc:
             # Writes are optional; the next read simply recomputes the history.
-            logger.warning("checkpoint history cache write failed; skipping: %s", exc)
+            _warn("checkpoint history cache write failed; skipping: %s", exc)
 
     async def adelete_thread(self, key_prefix: str, thread_id: str) -> None:
         """SCAN+UNLINK every entry of one thread. Failure degrades to
@@ -139,7 +152,7 @@ class RedisCheckpointHistoryCache:
                 if cursor == 0:
                     break
         except _redis_error() as exc:
-            logger.warning("checkpoint history cache thread purge failed; residual entries expire via TTL: %s", exc)
+            _warn("checkpoint history cache thread purge failed; residual entries expire via TTL: %s", exc)
 
     def stats(self) -> CheckpointCacheStats:
         return CheckpointCacheStats(

backend/packages/harness/deerflow/runtime/checkpointer/cached_saver.py

@@ -30,7 +30,7 @@ from typing import Any
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import BaseCheckpointSaver, CheckpointTuple, PendingWrite
 
-from deerflow.runtime.checkpoint_cache.base import make_history_key
+from deerflow.runtime.checkpoint_cache.base import make_history_key, valid_history_entry
 
 logger = logging.getLogger(__name__)
 
@@ -132,7 +132,7 @@ class CachedHistorySaver(BaseCheckpointSaver):
         missing: list[str] = []
         for ch in channels:
             entry = hits.get(keys[ch])
-            if entry is None:
+            if not valid_history_entry(entry, channel=ch):
                 missing.append(ch)
             else:
                 found[ch] = entry
@@ -178,7 +178,7 @@ class CachedHistorySaver(BaseCheckpointSaver):
         key = self._key(parent, channel)
         hits = await self._cache.aget_many([key])
         parent_history = hits.get(key)
-        if parent_history is None:
+        if not valid_history_entry(parent_history, channel=channel):
             if depth > 0:
                 parent_history = await self._aresolve(parent, channel, depth - 1)
             else:
@@ -222,7 +222,7 @@ class CachedHistorySaver(BaseCheckpointSaver):
         missing: list[str] = []
         for ch in channels:
             entry = hits.get(keys[ch])
-            if entry is None:
+            if not valid_history_entry(entry, channel=ch):
                 missing.append(ch)
             else:
                 found[ch] = entry
@@ -253,7 +253,7 @@ class CachedHistorySaver(BaseCheckpointSaver):
 
         key = self._key(parent, channel)
         parent_history = self._cache.get_many([key]).get(key)
-        if parent_history is None:
+        if not valid_history_entry(parent_history, channel=channel):
             if depth > 0:
                 parent_history = self._resolve_sync(parent, channel, depth - 1)
             else:

backend/tests/test_redis_history_record_validation.py

@@ -0,0 +1,295 @@
+"""Malformed cache write records must rebuild from authoritative checkpoints."""
+
+from types import SimpleNamespace
+from typing import Annotated, TypedDict
+from unittest.mock import AsyncMock
+
+import pytest
+from langgraph.channels import DeltaChannel
+from langgraph.checkpoint.memory import InMemorySaver
+from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
+from langgraph.graph import StateGraph
+
+from deerflow.runtime.checkpoint_cache import redis as redis_cache
+from deerflow.runtime.checkpoint_cache.memory import MemoryCheckpointHistoryCache
+from deerflow.runtime.checkpoint_state import CheckpointStateAccessor
+from deerflow.runtime.checkpointer.cached_saver import CachedHistorySaver
+
+
+class RedisTransport:
+    """Only the Redis transport is replaced; serialization and graph reads are real."""
+
+    def __init__(self):
+        self.values = {}
+
+    async def mget(self, keys):
+        return [self.values.get(key) for key in keys]
+
+    def pipeline(self, transaction=False):
+        return self
+
+    def set(self, key, value, ex=None):
+        self.values[key] = value
+        return self
+
+    async def execute(self):
+        return []
+
+    async def aclose(self):
+        pass
+
+
+def fold(state, writes):
+    result = list(state)
+    for batch in writes:
+        result.extend(batch)
+    return result
+
+
+class State(TypedDict):
+    items: Annotated[list[str], DeltaChannel(fold, snapshot_frequency=100)]
+
+
+BAD_WRITES = [None, 17, "abc", {"task": "x"}, (), ("task", "items"), ("task", "items", [], "extra"), (17, "items", []), ("task", 17, [])]
+
+
+def fail_warning(*args, **kwargs):
+    raise RuntimeError("handler failed")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("bad_write", BAD_WRITES, ids=["null", "number", "string", "mapping", "empty", "short", "long", "task-type", "channel-type"])
+@pytest.mark.parametrize("warning_handler_fails", [False, True], ids=["logging", "raising-handler"])
+async def test_bad_write_rebuilds_snapshot_and_replaces_corrupt_cache(monkeypatch, bad_write, warning_handler_fails):
+    if warning_handler_fails:
+        monkeypatch.setattr(redis_cache.logger, "warning", fail_warning)
+    transport = RedisTransport()
+    monkeypatch.setattr(redis_cache, "_create_client", lambda *args, **kwargs: transport)
+    serde = JsonPlusSerializer()
+    cache = redis_cache.RedisCheckpointHistoryCache("redis://unused", serde=serde, ttl_seconds=60)
+    inner = InMemorySaver()
+    saver = CachedHistorySaver(inner, cache, key_prefix="record-validation")
+    builder = StateGraph(State)
+    builder.add_node("one", lambda state: {"items": ["saved"]})
+    builder.set_entry_point("one")
+    builder.set_finish_point("one")
+    graph = builder.compile(checkpointer=saver)
+    accessor = CheckpointStateAccessor.bind(graph, saver, mode="delta")
+    config = {"configurable": {"thread_id": "invalid-record"}}
+    try:
+        await graph.ainvoke({"items": ["input"]}, config)
+        expected = (await accessor.aget(config)).values
+        assert expected == {"items": ["input", "saved"]}
+        target = await inner.aget_tuple(config)
+        key = saver._key(target, "items")
+        # Keep a plausible prefix too: accepting only part of a corrupt entry
+        # would inject this uncommitted value into an otherwise successful read.
+        tag, payload = serde.dumps_typed({"writes": [("task", "items", ["injected"]), bad_write]})
+        corrupted = tag.encode() + b"\x00" + payload
+        transport.values[key] = corrupted
+        misses = cache.stats().misses
+
+        assert (await accessor.aget(config)).values == expected
+        assert cache.stats().misses > misses
+        assert transport.values[key] != corrupted
+        # The repaired cache is warm: reads stay correct without a new walk.
+        misses = cache.stats().misses
+        assert (await accessor.aget(config)).values == expected
+        assert cache.stats().misses == misses
+    finally:
+        await cache.aclose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("record", [("task", "items", ["saved"]), ["task", "items", ["saved"]]], ids=["tuple", "list"])
+@pytest.mark.parametrize("include_corrupt", [False, True], ids=["valid-control", "corrupt-neighbor"])
+async def test_valid_record_shapes_and_neighbor_entries_are_preserved(monkeypatch, record, include_corrupt):
+    transport = RedisTransport()
+    monkeypatch.setattr(redis_cache, "_create_client", lambda *args, **kwargs: transport)
+    serde = JsonPlusSerializer()
+    cache = redis_cache.RedisCheckpointHistoryCache("redis://unused", serde=serde, ttl_seconds=60)
+    try:
+        await cache.aset_many({"good": {"writes": [record]}, "empty": {"writes": [], "seed": None}})
+        if include_corrupt:
+            await cache.aset_many({"bad": {"writes": ["abc"]}})
+        found = await cache.aget_many(["good", "bad", "empty"] if include_corrupt else ["good", "empty"])
+        assert set(found) == {"good", "empty"}
+        assert list(found["good"]["writes"][0]) == list(record)
+        assert found["empty"] == {"writes": [], "seed": None}
+        assert cache.stats().hits == 2
+        assert cache.stats().misses == int(include_corrupt)
+    finally:
+        await cache.aclose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("entry", [{}, None, [], {"writes": None}, {"writes": ()}] + [{"writes": [bad]} for bad in BAD_WRITES])
+async def test_invalid_ingest_skips_only_bad_entry(monkeypatch, entry):
+    transport = RedisTransport()
+    monkeypatch.setattr(redis_cache, "_create_client", lambda *args, **kwargs: transport)
+    cache = redis_cache.RedisCheckpointHistoryCache("redis://unused", serde=JsonPlusSerializer(), ttl_seconds=60)
+    try:
+        await cache.aset_many({"before": {"writes": []}, "bad": entry, "after": {"writes": [("task", "items", ["saved"])]}})
+        assert set(transport.values) == {"before", "after"}
+        assert set(await cache.aget_many(["before", "after"])) == {"before", "after"}
+        assert cache.stats().write_serialization_failures == 0
+    finally:
+        await cache.aclose()
+
+
+@pytest.mark.asyncio
+async def test_invalid_ingest_logging_cannot_discard_valid_neighbor(monkeypatch):
+    transport = RedisTransport()
+    monkeypatch.setattr(redis_cache, "_create_client", lambda *args, **kwargs: transport)
+
+    monkeypatch.setattr(redis_cache.logger, "warning", fail_warning)
+    cache = redis_cache.RedisCheckpointHistoryCache("redis://unused", serde=JsonPlusSerializer(), ttl_seconds=60)
+    try:
+        await cache.aset_many({"bad": {"writes": [None]}, "good": {"writes": []}})
+        assert set(transport.values) == {"good"}
+    finally:
+        await cache.aclose()
+
+
+@pytest.mark.asyncio
+async def test_serialization_logging_cannot_discard_valid_neighbors(monkeypatch):
+    transport = RedisTransport()
+    monkeypatch.setattr(redis_cache, "_create_client", lambda *args, **kwargs: transport)
+    monkeypatch.setattr(redis_cache.logger, "warning", fail_warning)
+    execute = AsyncMock(wraps=transport.execute)
+    monkeypatch.setattr(transport, "execute", execute)
+    cache = redis_cache.RedisCheckpointHistoryCache("redis://unused", serde=JsonPlusSerializer(), ttl_seconds=60)
+    try:
+        await cache.aset_many({"before": {"writes": []}, "bad": {"writes": [], "seed": object()}, "after": {"writes": [("task", "items", ["saved"])]}})
+
+        assert set(transport.values) == {"before", "after"}
+        assert set(await cache.aget_many(["before", "after"])) == {"before", "after"}
+        assert cache.stats().write_serialization_failures == 1
+        execute.assert_awaited_once()
+    finally:
+        await cache.aclose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["read", "write", "purge"])
+async def test_warning_handler_failure_preserves_redis_error_fallback(monkeypatch, operation):
+    from redis.exceptions import RedisError
+
+    transport = RedisTransport()
+    monkeypatch.setattr(redis_cache, "_create_client", lambda *args, **kwargs: transport)
+    monkeypatch.setattr(redis_cache.logger, "warning", fail_warning)
+    cache = redis_cache.RedisCheckpointHistoryCache("redis://unused", serde=JsonPlusSerializer(), ttl_seconds=60)
+    error = RedisError("Redis unavailable")
+    try:
+        if operation == "read":
+            monkeypatch.setattr(transport, "mget", AsyncMock(side_effect=error))
+            assert await cache.aget_many(["one", "two"]) == {}
+            assert cache.stats().misses == 2
+        elif operation == "write":
+            monkeypatch.setattr(transport, "execute", AsyncMock(side_effect=error))
+            await cache.aset_many({"good": {"writes": []}})
+        else:
+            transport.connection_pool = SimpleNamespace(get_encoder=lambda: SimpleNamespace(encode=lambda key: key.encode()))
+            transport.scan = AsyncMock(side_effect=error)
+            await cache.adelete_thread("logging", "thread")
+    finally:
+        await cache.aclose()
+
+
+def build_history_graph(saver, steps=4):
+    builder = StateGraph(State)
+    for index in range(steps):
+        builder.add_node(str(index), lambda state, index=index: {"items": [f"saved{index}"]})
+        if index:
+            builder.add_edge(str(index - 1), str(index))
+    builder.set_entry_point("0")
+    builder.set_finish_point(str(steps - 1))
+    return builder.compile(checkpointer=saver)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("async_read", [False, True], ids=["sync", "async"])
+@pytest.mark.parametrize("location", ["target", "ancestor"])
+@pytest.mark.parametrize("record_type", [tuple, list], ids=["tuple", "list"])
+async def test_foreign_channel_rebuilds_complete_history(async_read, location, record_type):
+    cache = MemoryCheckpointHistoryCache()
+    inner = InMemorySaver()
+    saver = CachedHistorySaver(inner, cache, key_prefix="channel-validation")
+    graph = build_history_graph(saver)
+    accessor = CheckpointStateAccessor.bind(graph, saver, mode="delta")
+    authority = CheckpointStateAccessor.bind(build_history_graph(inner), inner, mode="delta")
+    config = {"configurable": {"thread_id": "foreign-channel"}}
+    try:
+        # The memory backend supports sync reads even in a running loop. Keep
+        # both accessor entry points covered without a transport-only assertion.
+        graph.invoke({"items": ["input"]}, config)
+        target = inner.get_tuple(config)
+        poisoned = target
+        if location == "ancestor":
+            selected = None
+            for candidate in inner.list(config):
+                if candidate.parent_config is None:
+                    continue
+                parent = inner.get_tuple(candidate.parent_config)
+                if parent is not None and "items" not in (parent.checkpoint.get("channel_values") or {}):
+                    selected = candidate, parent
+                    break
+            assert selected is not None, "fixture must reach recursive non-snapshot composition"
+            target, poisoned = selected
+        expected = authority.get(target.config).values
+        assert expected["items"]
+        before = inner.get_delta_channel_history(config=target.config, channels=["items"])
+        cache.delete_thread("channel-validation", "foreign-channel")
+        key = saver._key(poisoned, "items")
+        bad = {"writes": [("task", "items", ["INJECTED"]), record_type(("task", "foreign-channel", ["FOREIGN"]))]}
+        cache.set_many({key: bad})
+        if location == "ancestor":
+            assert cache.get_many([saver._key(target, "items")]) == {}
+
+        observed = await accessor.aget(target.config) if async_read else accessor.get(target.config)
+        assert observed.values == expected
+        assert cache.get_many([key])[key] != bad
+        assert inner.get_delta_channel_history(config=target.config, channels=["items"]) == before
+        misses = cache.stats().misses
+        observed = await accessor.aget(target.config) if async_read else accessor.get(target.config)
+        assert observed.values == expected
+        assert cache.stats().misses == misses
+    finally:
+        await cache.aclose()
+
+
+def fold_text(state, writes):
+    return state + "".join(writes)
+
+
+class StringState(TypedDict):
+    items: Annotated[str, DeltaChannel(fold_text, snapshot_frequency=100)]
+
+
+@pytest.mark.asyncio
+async def test_plain_string_seed_remains_a_warm_valid_state(monkeypatch):
+    transport = RedisTransport()
+    monkeypatch.setattr(redis_cache, "_create_client", lambda *args, **kwargs: transport)
+    cache = redis_cache.RedisCheckpointHistoryCache("redis://unused", serde=JsonPlusSerializer(), ttl_seconds=60)
+    inner = InMemorySaver()
+    saver = CachedHistorySaver(inner, cache, key_prefix="string-seed")
+    builder = StateGraph(StringState)
+    builder.add_node("one", lambda state: {"items": "saved"})
+    builder.set_entry_point("one")
+    builder.set_finish_point("one")
+    graph = builder.compile(checkpointer=saver)
+    accessor = CheckpointStateAccessor.bind(graph, saver, mode="delta")
+    config = {"configurable": {"thread_id": "string-seed"}}
+    try:
+        await graph.ainvoke({"items": "input"}, config)
+        expected = (await accessor.aget(config)).values
+        assert expected == {"items": "inputsaved"}
+        target = await inner.aget_tuple(config)
+        key = saver._key(target, "items")
+        await cache.aset_many({key: {"writes": [], "seed": "inputsaved"}})
+        misses = cache.stats().misses
+        for _ in range(2):
+            assert (await accessor.aget(config)).values == expected
+        assert cache.stats().misses == misses
+    finally:
+        await cache.aclose()