fix(redis): add Redis Cluster support for AgentStateStore

agentscope-dependencies-bom/pom.xml

@@ -100,6 +100,7 @@
         <a2a-transport-jsonrpc.version>0.3.3.Final</a2a-transport-jsonrpc.version>
         <jedis.version>7.4.1</jedis.version>
         <lettuce.version>6.4.2.RELEASE</lettuce.version>
+        <testcontainers.version>2.0.5</testcontainers.version>
         <xxl-job.version>3.3.2</xxl-job.version>
         <quartz.version>2.5.2</quartz.version>
         <spring.version>7.0.7</spring.version>
@@ -197,6 +198,15 @@
                 <scope>import</scope>
             </dependency>
 
+            <!-- Testcontainers -->
+            <dependency>
+                <groupId>org.testcontainers</groupId>
+                <artifactId>testcontainers-bom</artifactId>
+                <version>${testcontainers.version}</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+
             <!-- Google Guava -->
             <dependency>
                 <groupId>com.google.guava</groupId>

agentscope-extensions/agentscope-extensions-redis/pom.xml

@@ -58,5 +58,30 @@
             <groupId>io.lettuce</groupId>
             <artifactId>lettuce-core</artifactId>
         </dependency>
+
+        <!-- Test dependencies -->
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.testcontainers</groupId>
+            <artifactId>testcontainers</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/RedisAgentStateStore.java

@@ -47,15 +47,21 @@
  *   <li>Redisson - Standalone, Cluster, Sentinel, Master/Slave</li>
  * </ul>
  *
- * <p>The session state is stored in Redis with following key structure:
+ * <p>The session state is stored in Redis with following key structure (where
+ * {@code {<user>/<session>}} is a Redis Cluster hash tag so all keys of a session share one slot):
  *
  * <ul>
- *   <li>Single state: {@code {prefix}{sessionId}:{stateKey}} - Redis String containing JSON
- *   <li>List state: {@code {prefix}{sessionId}:{stateKey}:list} - Redis List containing JSON items
- *   <li>List hash: {@code {prefix}{sessionId}:{stateKey}:list:_hash} - Hash for change detection
- *   <li>AgentStateStore marker: {@code {prefix}{sessionId}:_keys} - Redis Set tracking all state keys
+ *   <li>Single state: {@code <prefix>{<user>/<session>}:<stateKey>} - Redis String containing JSON
+ *   <li>List state: {@code <prefix>{<user>/<session>}:<stateKey>:list} - Redis List containing JSON items
+ *   <li>List hash: {@code <prefix>{<user>/<session>}:<stateKey>:list:_hash} - Hash for change detection
+ *   <li>AgentStateStore marker: {@code <prefix>{<user>/<session>}:_keys} - Redis Set tracking all state keys
  * </ul>
  *
+ * <p><strong>Breaking change:</strong> the slot id is wrapped in a Redis Cluster hash tag
+ * ({@code {...}}) so all keys of one session share one Cluster slot (required by the multi-key
+ * Lua {@code EVAL}). Data written with the previous, un-tagged key layout is not readable and
+ * must be migrated.
+ *
  * <p><strong>Jedis Usage Examples:</strong></p>
  *
  * <p>Jedis Standalone (using RedisClient):
@@ -416,15 +422,19 @@ public void delete(String userId, String sessionId) {
     public Set<String> listSessionIds(String userId) {
         String userSegment = normalizeUser(userId);
         try {
-            // Pattern: {prefix}{userSegment}/{sessionId}:_keys
-            String pattern = keyPrefix + userSegment + "/*" + KEYS_SUFFIX;
+            // Keys have the form: {prefix}{{userSegment}/{sessionId}}:_keys.
+            // Escape glob metacharacters in userSegment so a userId containing
+            // '*', '?', '[', ']' or '\' cannot widen or skew the SCAN MATCH pattern.
+            String pattern = keyPrefix + "{" + escapeGlob(userSegment) + "/*}" + KEYS_SUFFIX;
             Set<String> keysKeys = client.findKeysByPattern(pattern);
             Set<String> sessionIds = new HashSet<>();
-            String userPrefix = keyPrefix + userSegment + "/";
+            String openTag = keyPrefix + "{" + userSegment + "/";
+            String closeTag = "}" + KEYS_SUFFIX;
             for (String keysKey : keysKeys) {
-                String withoutPrefix = keysKey.substring(userPrefix.length());
+                // Strip the prefix and the closing tag to recover the sessionId.
+                String afterPrefix = keysKey.substring(openTag.length());
                 String sessionId =
-                        withoutPrefix.substring(0, withoutPrefix.length() - KEYS_SUFFIX.length());
+                        afterPrefix.substring(0, afterPrefix.length() - closeTag.length());
                 sessionIds.add(sessionId);
             }
             return sessionIds;
@@ -440,12 +450,43 @@ private static String normalizeUser(String userId) {
         return userId == null || userId.isBlank() ? ANON_USER : userId;
     }
 
-    /** Combine {@code (userId, sessionId)} into a single Redis slot identifier. */
+    /** Escape Redis glob metacharacters so {@code userSegment} matches literally in SCAN MATCH. */
+    private static String escapeGlob(String s) {
+        StringBuilder sb = new StringBuilder(s.length());
+        for (int i = 0; i < s.length(); i++) {
+            char c = s.charAt(i);
+            if (c == '*' || c == '?' || c == '[' || c == ']' || c == '\\') {
+                sb.append('\\');
+            }
+            sb.append(c);
+        }
+        return sb.toString();
+    }
+
+    /** Reject characters that would prematurely terminate a Redis Cluster hash tag. */
+    private static void rejectBraces(String value, String name) {
+        if (value.indexOf('{') >= 0 || value.indexOf('}') >= 0) {
+            throw new IllegalArgumentException(
+                    name + " must not contain '{' or '}' (reserved for Redis Cluster hash tags)");
+        }
+    }
+
+    /**
+     * Combine {@code (userId, sessionId)} into a single Redis slot identifier.
+     *
+     * <p>The result is wrapped in a Redis Cluster hash tag {@code {...}} so that all keys derived
+     * from this slot (payload, version, keys-set, list, list-hash) hash to the same slot. This is
+     * required by the multi-key {@code SAVE_SCRIPT} Lua eval in cluster mode. {@code userId}
+     * and {@code sessionId} must not contain {@code { } }, otherwise the tag is truncated early.
+     */
     private static String slotId(String userId, String sessionId) {
         if (sessionId == null || sessionId.isBlank()) {
             throw new IllegalArgumentException("sessionId must not be blank");
         }
-        return normalizeUser(userId) + "/" + sessionId;
+        rejectBraces(sessionId, "sessionId");
+        String user = normalizeUser(userId);
+        rejectBraces(user, "userId");
+        return "{" + user + "/" + sessionId + "}";
     }
 
     @Override
@@ -464,7 +505,11 @@ public Mono<Integer> clearAllSessions() {
                             try {
                                 Set<String> keys = client.findKeysByPattern(keyPrefix + "*");
                                 if (!keys.isEmpty()) {
-                                    client.deleteKeys(keys.toArray(new String[0]));
+                                    // Delete keys one by one to avoid CROSSSLOT errors
+                                    // in Redis Cluster mode where keys span multiple slots.
+                                    for (String key : keys) {
+                                        client.deleteKeys(key);
+                                    }
                                 }
                                 return keys.size();
                             } catch (Exception e) {

agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/jedis/JedisClientAdapter.java

@@ -23,8 +23,6 @@
 import redis.clients.jedis.RedisClusterClient;
 import redis.clients.jedis.RedisSentinelClient;
 import redis.clients.jedis.UnifiedJedis;
-import redis.clients.jedis.params.ScanParams;
-import redis.clients.jedis.resps.ScanResult;
 
 /**
  * Adapter for Jedis Redis client.
@@ -153,21 +151,17 @@ public boolean keyExists(String key) {
 
     @Override
     public Set<String> findKeysByPattern(String pattern) {
-        Set<String> matchingKeys = new HashSet<>();
-        String cursor = ScanParams.SCAN_POINTER_START;
-        ScanParams scanParams = new ScanParams().match(pattern);
-        do {
-            ScanResult<String> scanResult = unifiedJedis.scan(cursor, scanParams);
-            if (scanResult != null) {
-                matchingKeys.addAll(scanResult.getResult());
-                cursor = scanResult.getCursor();
-            } else {
-                break;
-            }
-        } while (!cursor.equals(ScanParams.SCAN_POINTER_START));
-        return matchingKeys;
+        // scanIteration transparently walks every master node in cluster mode (and the single
+        // node in standalone mode), so keys living on other shards are not silently missed
+        // (affects RedisAgentStateStore.listSessionIds / clearAllSessions).
+        Set<String> keys = new HashSet<>();
+        unifiedJedis.scanIteration(SCAN_BATCH_SIZE, pattern).collect(keys);
+        return keys;
     }
 
+    /** COUNT hint passed to each SCAN call inside {@link #findKeysByPattern}. */
+    private static final int SCAN_BATCH_SIZE = 100;
+
     @Override
     public long evalScript(String script, List<String> keys, List<String> args) {
         Object result = unifiedJedis.eval(script, keys, args);

agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/state/lettuce/LettuceClientAdapter.java

@@ -26,6 +26,8 @@
 import io.lettuce.core.cluster.RedisClusterClient;
 import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
 import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
+import io.lettuce.core.cluster.models.partitions.Partitions;
+import io.lettuce.core.cluster.models.partitions.RedisClusterNode;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -107,6 +109,14 @@ public class LettuceClientAdapter implements RedisClientAdapter {
      */
     private final RedisAdvancedClusterCommands<String, String> clusterCommands;
 
+    /**
+     * The cluster connection backing {@link #clusterCommands}. Kept so that
+     * {@link #findKeysByPattern} can iterate over every master node (Lettuce's
+     * {@code clusterCommands.scan(...)} only scans the default node and would
+     * otherwise miss keys living on other shards). Null in standalone/sentinel mode.
+     */
+    private final StatefulRedisClusterConnection<String, String> clusterConnection;
+
     /**
      * Closeable resource handler for cleaning up connections and clients.
      * Uses a strategy pattern to handle different cleanup logic for
@@ -117,9 +127,11 @@ public class LettuceClientAdapter implements RedisClientAdapter {
     private LettuceClientAdapter(
             RedisCommands<String, String> commands,
             RedisAdvancedClusterCommands<String, String> clusterCommands,
+            StatefulRedisClusterConnection<String, String> clusterConnection,
             AutoCloseable closeable) {
         this.commands = commands;
         this.clusterCommands = clusterCommands;
+        this.clusterConnection = clusterConnection;
         this.closeable = closeable;
     }
 
@@ -136,7 +148,7 @@ public static LettuceClientAdapter of(RedisClient redisClient) {
         }
         StatefulRedisConnection<String, String> connection = redisClient.connect();
         return new LettuceClientAdapter(
-                connection.sync(), null, new StandaloneCloser(connection, redisClient));
+                connection.sync(), null, null, new StandaloneCloser(connection, redisClient));
     }
 
     /**
@@ -152,7 +164,10 @@ public static LettuceClientAdapter of(RedisClusterClient redisClusterClient) {
         }
         StatefulRedisClusterConnection<String, String> connection = redisClusterClient.connect();
         return new LettuceClientAdapter(
-                null, connection.sync(), new ClusterCloser(connection, redisClusterClient));
+                null,
+                connection.sync(),
+                connection,
+                new ClusterCloser(connection, redisClusterClient));
     }
 
     @Override
@@ -249,9 +264,22 @@ public boolean keyExists(String key) {
     public Set<String> findKeysByPattern(String pattern) {
         if (commands != null) {
             return scanKeys(pattern, commands::scan);
-        } else {
-            return scanKeys(pattern, clusterCommands::scan);
         }
+        // Cluster mode: SCAN only touches a single node by default, so iterate over
+        // every master node and aggregate the results; otherwise keys living on other
+        // shards would be silently missed (affects listSessionIds / clearAllSessions).
+        Set<String> keys = new HashSet<>();
+        Partitions partitions = clusterConnection.getPartitions();
+        for (RedisClusterNode node : partitions) {
+            if (!node.is(RedisClusterNode.NodeFlag.MASTER)) {
+                continue;
+            }
+            // nodeConn is owned by the cluster connection; do not close it here.
+            StatefulRedisConnection<String, String> nodeConn =
+                    clusterConnection.getConnection(node.getNodeId());
+            keys.addAll(scanKeys(pattern, nodeConn.sync()::scan));
+        }
+        return keys;
     }
 
     /**

agentscope-extensions/agentscope-extensions-redis/src/main/java/io/agentscope/extensions/redis/store/RedisStore.java

@@ -33,15 +33,24 @@
  * <p>For each item with namespace {@code [a, b, c]} and key {@code k}, two Redis keys are used:
  *
  * <ul>
- *   <li><b>Item hash</b> {@code <prefix>item:<ns>\0<k>} — a Redis hash with fields {@code value}
+ *   <li><b>Item hash</b> {@code <prefix>item:{<ns>}\0<k>} — a Redis hash with fields {@code value}
  *       (JSON-encoded {@code Map<String,Object>}) and {@code version} (a stringified long).
- *   <li><b>Namespace index</b> {@code <prefix>idx:<ns>} — a sorted set (all scores {@code 0})
+ *   <li><b>Namespace index</b> {@code <prefix>idx:{<ns>}} — a sorted set (all scores {@code 0})
  *       holding every {@code k} written under that exact namespace, enabling lexicographic
  *       {@link #search} via {@code ZRANGEBYLEX} without scanning the keyspace.
  * </ul>
  *
  * <p>{@code <ns>} is the namespace components joined with {@code "\0"}.
  *
+ * <p>The {@code {<ns>}} wrapper is a Redis Cluster hash tag: it forces the item hash and the
+ * namespace index into the same slot, which is required by the multi-key {@code EVAL} scripts
+ * below in cluster mode. An empty namespace is mapped to {@code {_root_}} because Redis ignores
+ * an empty tag. Namespace segments must not contain {@code { } }.
+ *
+ * <p><strong>Breaking change:</strong> the namespace is wrapped in a Redis Cluster hash tag
+ * ({@code {<ns>}}); data written with the previous, un-tagged key layout is not readable and
+ * must be migrated.
+ *
  * <h2>Concurrency</h2>
  *
  * <p>{@link #put} and {@link #putIfVersion} both run as a single Lua {@code EVAL}, making the
@@ -231,12 +240,33 @@ private Map<String, Object> deserialize(String json) {
         }
     }
 
+    /**
+     * Tag content used when the namespace is empty. Redis ignores an empty {@code {}}
+     * tag (treating the whole key as the hash input), which would split itemKey and
+     * indexKey across slots; a non-empty placeholder keeps them together.
+     */
+    private static final String EMPTY_NAMESPACE_TAG = "_root_";
+
+    /** Build the Redis Cluster hash tag for a namespace, mapping empty to a placeholder. */
+    private String hashTag(List<String> namespace) {
+        String ns = namespacePath(namespace);
+        return "{" + (ns.isEmpty() ? EMPTY_NAMESPACE_TAG : ns) + "}";
+    }
+
+    /** Reject characters that would prematurely terminate a Redis Cluster hash tag. */
+    private static void rejectBraces(String value, String name) {
+        if (value.indexOf('{') >= 0 || value.indexOf('}') >= 0) {
+            throw new IllegalArgumentException(
+                    name + " must not contain '{' or '}' (reserved for Redis Cluster hash tags)");
+        }
+    }
+
     private String itemKey(List<String> namespace, String key) {
-        return keyPrefix + "item:" + namespacePath(namespace) + NS_SEPARATOR + key;
+        return keyPrefix + "item:" + hashTag(namespace) + NS_SEPARATOR + key;
     }
 
     private String indexKey(List<String> namespace) {
-        return keyPrefix + "idx:" + namespacePath(namespace);
+        return keyPrefix + "idx:" + hashTag(namespace);
     }
 
     private static String namespacePath(List<String> namespace) {
@@ -247,6 +277,7 @@ private static String namespacePath(List<String> namespace) {
             if (segment == null) {
                 throw new IllegalArgumentException("namespace segment must not be null");
             }
+            rejectBraces(segment, "namespace segment");
             if (i > 0) {
                 sb.append(NS_SEPARATOR);
             }