fix(checkpoints): confine Redis history purge to literal namespaces (#6709)

* fix(checkpoints): keep Redis history purge within literal namespace

* test(checkpoints): cover glob metacharacters in thread IDs

* test(checkpoints): cover Redis purge with non-UTF-8 client encoding

* fix(checkpoints): honor Redis client encoding in purge fence

* test(checkpoints): simulate Redis client encoders in purge regressions

* fix(checkpoints): restore Redis encoder fence assignment

* test(checkpoints): give Redis fake an encoder

README.md

@@ -498,7 +498,9 @@ 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.
+different cache settings. Redis thread cleanup treats a configured `key_prefix`
+as literal text, including glob metacharacters, so deletion stays within that
+namespace and thread.
 
 > [!TIP]
 > On Linux, if Docker-based commands fail with `permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker.sock`, add your user to the `docker` group and re-login before retrying. See [CONTRIBUTING.md](CONTRIBUTING.md#linux-docker-daemon-permission-denied) for the full fix.

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

@@ -1,3 +1,10 @@
+### 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.
+
 ### Stream Bridge Heartbeats
 
 Memory/Redis bridges keep startup-only `stream_bridge.heartbeat_interval_seconds`; explicit `subscribe(..., heartbeat_interval=...)` overrides it. Provider contexts drain owned cache/bridge/checkpointer/Store teardown across cancellation. `close_agent_stream()` shields close to completion: preserve counts, defer host cancellation, balance repeats, keep active errors, log attached close failures at callers, map close cancellation to failure, never time out.

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

@@ -112,10 +112,16 @@ class RedisCheckpointHistoryCache:
         TTL-bounded residual retention; the source-of-truth delete already
         happened, so this never raises."""
         stem = thread_key_stem(key_prefix, thread_id)
+        # The configured namespace is a literal key prefix, not a Redis glob.
+        pattern = "".join("\\" + char if char in "\\*?[]" else char for char in stem) + "*"
+        # redis-py encodes SCAN results with the configured connection encoder.
+        # A fixed UTF-8 fence silently skips owned keys for other encodings.
+        encoded_stem = self._client.connection_pool.get_encoder().encode(stem)
         try:
             cursor = 0
             while True:
-                cursor, keys = await self._client.scan(cursor=cursor, match=stem + "*", count=500)
+                cursor, keys = await self._client.scan(cursor=cursor, match=pattern, count=500)
+                keys = [key for key in keys if key.startswith(encoded_stem if isinstance(key, bytes) else stem)]
                 if keys:
                     await self._client.unlink(*keys)
                 if cursor == 0:

backend/tests/test_checkpoint_cache_redis.py

@@ -13,6 +13,16 @@ from deerflow.runtime.checkpoint_cache.provider import (
 )
 
 
+class _FakeEncoder:
+    def encode(self, value: str) -> bytes:
+        return value.encode()
+
+
+class _FakeConnectionPool:
+    def get_encoder(self) -> _FakeEncoder:
+        return _FakeEncoder()
+
+
 class _FakeRedis:
     """Minimal async redis stand-in: mget / set / pipeline / scan / unlink."""
 
@@ -20,6 +30,7 @@ class _FakeRedis:
         self.store: dict[str, bytes] = {}
         self.ttls: dict[str, int | None] = {}
         self.unlinked: list[tuple[str, ...]] = []
+        self.connection_pool = _FakeConnectionPool()
 
     async def mget(self, keys: list[str]) -> list[bytes | None]:
         return [self.store.get(k) for k in keys]

backend/tests/test_checkpoint_cache_redis_literal_purge.py

@@ -0,0 +1,78 @@
+"""Thread deletion treats configured cache namespaces as literal Redis keys."""
+
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+from deerflow.runtime.checkpoint_cache.base import make_history_key
+from deerflow.runtime.checkpoint_cache.redis import RedisCheckpointHistoryCache
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("prefix", "thread_id", "sibling_thread_id", "pattern"),
+    [
+        ("tenant*", "t1", "t10", r"tenant\*:t1:*"),
+        ("tenant?", "t1", "t10", r"tenant\?:t1:*"),
+        ("tenant[ab]", "t1", "t10", r"tenant\[ab\]:t1:*"),
+        (r"tenant\name", "t1", "t10", r"tenant\\name:t1:*"),
+        ("plain", "t1", "t10", "plain:t1:*"),
+        ("plain", "t1*", "t10", r"plain:t1\*:*"),
+    ],
+)
+async def test_purge_escapes_namespace_and_filters_results(prefix, thread_id, sibling_thread_id, pattern):
+    # Exercise the actual purge method without needing a running Redis server.
+    # The expected SCAN pattern is a wire-level Redis glob contract; returned
+    # keys also include a foreign namespace to pin the final deletion fence.
+    cache = RedisCheckpointHistoryCache.__new__(RedisCheckpointHistoryCache)
+    client = AsyncMock()
+    cache._client = client
+    client.connection_pool = Mock()
+    client.connection_pool.get_encoder.return_value.encode.side_effect = lambda value: value.encode("utf-8")
+    owned = make_history_key(prefix, thread_id, "", "c1", "messages").encode()
+    foreign = make_history_key("tenant-other", thread_id, "", "c1", "messages").encode()
+    sibling = make_history_key(prefix, sibling_thread_id, "", "c1", "messages").encode()
+
+    async def scan(*, cursor, match, count):
+        assert match == pattern
+        assert count == 500
+        # Mixed byte/string results and multiple pages exercise redis clients
+        # with either response mode and preserve the scan cursor protocol.
+        if cursor == 0:
+            return 7, [owned, foreign, sibling]
+        assert cursor == 7
+        return 0, [owned.decode()]
+
+    client.scan.side_effect = scan
+    await cache.adelete_thread(prefix, thread_id)
+    assert client.unlink.await_args_list[0].args == (owned,)
+    assert client.unlink.await_args_list[1].args == (owned.decode(),)
+
+
+@pytest.mark.anyio
+async def test_purge_does_not_unlink_empty_filtered_page():
+    cache = RedisCheckpointHistoryCache.__new__(RedisCheckpointHistoryCache)
+    cache._client = AsyncMock()
+    cache._client.connection_pool = Mock()
+    cache._client.connection_pool.get_encoder.return_value.encode.side_effect = lambda value: value.encode("utf-8")
+    cache._client.scan.return_value = (0, [b"other:t1:history"])
+    await cache.adelete_thread("owned", "t1")
+    cache._client.unlink.assert_not_awaited()
+
+
+@pytest.mark.anyio
+async def test_purge_uses_client_encoding_for_byte_key_fence():
+    """A non-UTF-8 Redis client must still unlink its own non-ASCII keys."""
+    cache = RedisCheckpointHistoryCache.__new__(RedisCheckpointHistoryCache)
+    client = AsyncMock()
+    cache._client = client
+    client.connection_pool = Mock()
+    client.connection_pool.get_encoder.return_value.encode.side_effect = lambda value: value.encode("latin-1")
+    owned = make_history_key("café", "t1", "", "c1", "messages").encode("latin-1")
+    sibling = make_history_key("café", "t10", "", "c1", "messages").encode("latin-1")
+    client.scan.return_value = (0, [owned, sibling])
+
+    await cache.adelete_thread("café", "t1")
+
+    client.scan.assert_awaited_once_with(cursor=0, match="café:t1:*", count=500)
+    client.unlink.assert_awaited_once_with(owned)