fix(checkpoints): skip unserializable Redis history cache entries (#7014)

* fix(checkpoints): skip Redis cache entries that cannot be serialized

* style(checkpoints): format Redis serialization regression

* fix(test): assert Redis serialization payload bytes

* test(checkpoints): format Redis cache serialization regression

* chore: sync frontend agent guidance with upstream size-compliant main

* fix(checkpoints): track Redis cache serialization failures

* test(checkpoints): consolidate Redis serialization fallback regression

* style(tests): format Redis cache regression signature

---------

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

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

@@ -51,6 +51,7 @@ class CheckpointCacheStats:
     misses: int = 0
     evictions: int = 0
     entries: int = 0
+    write_serialization_failures: int = 0
 
     def as_dict(self) -> dict[str, int]:
         return {
@@ -58,6 +59,7 @@ class CheckpointCacheStats:
             "misses": self.misses,
             "evictions": self.evictions,
             "entries": self.entries,
+            "write_serialization_failures": self.write_serialization_failures,
         }
 
 

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

@@ -59,6 +59,7 @@ class RedisCheckpointHistoryCache:
         self._ttl = ttl_seconds if ttl_seconds > 0 else None
         self._hits = 0
         self._misses = 0
+        self._write_serialization_failures = 0
 
     async def aget_many(self, keys: list[str]) -> dict[str, dict[str, Any]]:
         if not keys:
@@ -100,7 +101,18 @@ class RedisCheckpointHistoryCache:
         try:
             pipe = self._client.pipeline(transaction=False)
             for key, entry in entries.items():
-                tag, data = self._serde.dumps_typed(entry)
+                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(
+                        "checkpoint history cache serialization failed; skipping (key=%s): %s",
+                        key,
+                        type(exc).__name__,
+                    )
+                    continue
                 pipe.set(key, tag.encode() + _TAG_SEPARATOR + data, ex=self._ttl)
             await pipe.execute()
         except _redis_error() as exc:
@@ -130,7 +142,11 @@ class RedisCheckpointHistoryCache:
             logger.warning("checkpoint history cache thread purge failed; residual entries expire via TTL: %s", exc)
 
     def stats(self) -> CheckpointCacheStats:
-        return CheckpointCacheStats(hits=self._hits, misses=self._misses)
+        return CheckpointCacheStats(
+            hits=self._hits,
+            misses=self._misses,
+            write_serialization_failures=self._write_serialization_failures,
+        )
 
     async def aclose(self) -> None:
         await self._client.aclose()

backend/tests/test_checkpoint_cache_redis.py

@@ -94,11 +94,23 @@ class _FailingRedis(_FakeRedis):
         raise RedisError("connection refused")
 
 
-def _make_cache(monkeypatch: pytest.MonkeyPatch, fake: _FakeRedis, ttl_seconds: int = 60, **kwargs: Any):
+def _make_cache(
+    monkeypatch: pytest.MonkeyPatch,
+    fake: _FakeRedis,
+    ttl_seconds: int = 60,
+    *,
+    serde: Any | None = None,
+    **kwargs: Any,
+):
     import deerflow.runtime.checkpoint_cache.redis as redis_mod
 
     monkeypatch.setattr(redis_mod, "_create_client", lambda *a, **k: fake)
-    return redis_mod.RedisCheckpointHistoryCache("redis://unused", serde=JsonPlusSerializer(), ttl_seconds=ttl_seconds, **kwargs)
+    return redis_mod.RedisCheckpointHistoryCache(
+        "redis://unused",
+        serde=serde if serde is not None else JsonPlusSerializer(),
+        ttl_seconds=ttl_seconds,
+        **kwargs,
+    )
 
 
 def _entry(i: int) -> dict:
@@ -161,6 +173,37 @@ async def test_redis_outage_skips_write(monkeypatch: pytest.MonkeyPatch, caplog:
     assert "write failed" in caplog.text
 
 
+@pytest.mark.anyio
+async def test_redis_serialization_failure_preserves_neighbor_writes(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> None:
+    class _SelectiveFailingSerde:
+        def dumps_typed(self, entry: dict) -> tuple[str, bytes]:
+            if entry.get("bad"):
+                raise TypeError("sensitive serializer detail")
+            return "json", b"encoded"
+
+    fake = _FakeRedis()
+    cache = _make_cache(monkeypatch, fake, serde=_SelectiveFailingSerde())
+    with caplog.at_level("WARNING"):
+        await cache.aset_many(
+            {
+                "before": {"writes": []},
+                "broken": {"writes": [], "bad": True},
+                "after": {"writes": []},
+            }
+        )
+
+    assert fake.store == {
+        "before": b"json\x00encoded",
+        "after": b"json\x00encoded",
+    }
+    assert fake.ttls == {"before": 60, "after": 60}
+    assert "broken" in caplog.text
+    assert "TypeError" in caplog.text
+    assert "sensitive serializer detail" not in caplog.text
+    assert cache.stats().write_serialization_failures == 1
+    assert cache.stats().as_dict()["write_serialization_failures"] == 1
+
+
 @pytest.mark.anyio
 async def test_zero_ttl_disables_expiry_explicitly(monkeypatch: pytest.MonkeyPatch):
     fake = _FakeRedis()