feat: add Redis Cluster session and memory services

[Wenvii]

Summary

• add RedisClusterSessionService and RedisClusterMemoryService backed by redis-py's native cluster clients
• add cluster-aware storage handling for cross-node key scans and decoded hash responses
• keep the existing standalone Redis services and reuse their behavior through a storage factory

Why

The existing Redis services use a standalone Redis client and cannot follow Redis Cluster redirects. These services allow session and memory persistence to work with a Redis Cluster deployment.

Validation

• pytest tests/storage/test_redis_cluster_storage.py tests/sessions/test_redis_cluster_session_service.py tests/memory/test_redis_cluster_memory_service.py
• pytest tests/sessions/test_redis_session_service.py tests/memory/test_redis_memory_service.py
• local six-node Redis Cluster integration test with cross-session memory recall

[codecov]

Codecov Report

❌ Patch coverage is 73.33333% with 44 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (main@5ca3bf2). Learn more about missing BASE report.

Files with missing lines	Patch %	Lines
trpc_agent_sdk/storage/_redis_cluster.py	68.57143%	44 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main        #223   +/-   ##
==========================================
  Coverage        ?   87.85080%           
==========================================
  Files           ?         482           
  Lines           ?       45147           
  Branches        ?           0           
==========================================
  Hits            ?       39662           
  Misses          ?        5485           
  Partials        ?           0           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[helloopenworld]

AI Code Review

我已经从 diff 和相关的父类代码（_redis.py、_redis_session_service.py、_redis_memory_service.py）中获取了足够的上下文，可以得出结论。

发现的问题

🚨 Critical

（未发现必须修复的阻塞性问题。）

⚠️ Warning

• trpc_agent_sdk/storage/_redis_cluster.py:71-79：同步集群路径存在稳定性与覆盖双重风险

  • is_async=False 是服务的默认值，但 create_redis_engine 为 async def，其内部通过 SyncRedisCluster.from_url(...) 构造同步集群客户端。SyncRedisCluster 在构造时会执行阻塞式的 CLUSTER SLOTS/CLUSTER NODES 拓扑发现，这会在事件循环中阻塞所有协程（异步 AsyncRedisCluster 通常延迟发现，两者行为不一致）。建议显式标注同步模式不被支持，或在线程池中执行同步发现；并补充 is_async=False 路径的测试（当前 4 个 storage 测试、2 个 service 测试全部只走 is_async=True）。

• trpc_agent_sdk/storage/_redis_cluster.py:71-77：create_redis_engine 缺少并发保护，首个请求并发触发时会创建多个集群客户端

  • if self._redis_client: return 的检查与赋值非原子，两个并发首次调用会各创建一个 RedisCluster 客户端，多余者不会被 close() 释放。集群客户端持有所有已发现节点的连接池，泄漏代价远高于父类的单连接池。建议用 asyncio.Lock 包裹构造逻辑。

• trpc_agent_sdk/storage/_redis_cluster.py:102-106：RedisClusterAsyncContextManager.__aexit__ 不关闭客户端，资源释放完全依赖显式 close()

  • 设计上共享客户端按 storage 生命周期常驻，但 context manager 退出时仅置空本地引用，不触发任何释放。若上层（RedisClusterSessionService/RedisClusterMemoryService）未调用 close()（例如异常退出路径），集群连接池将泄漏。建议在文档/析构中明确，或在 service 层保证 close() 必被调用。

💡 Suggestion

• trpc_agent_sdk/storage/_redis_cluster.py:138-149：execute_command 对 keys/hgetall 做了特化，但未覆盖 scan 本身被外部直接调用的情形；建议将 _scan_keys 的去重逻辑与 _decode_text 在注释中关联，便于后续维护时理解为何仅 keys 走 SCAN。

总结

整体实现思路清晰：通过 _create_storage 工厂注入集群存储，并必要地重写了 delete/query/execute_command/close 等方法以规避父类基于 isinstance(conn, AsyncRedis) 与连接池的假设，SCAN 去重与 bytes 规范化处理正确。无 Critical 阻塞问题；主要风险集中在同步模式未被测试且存在事件循环阻塞隐患，以及共享集群客户端的并发构造与释放缺口。

测试建议

• 补充 is_async=False 下 RedisClusterStorage 的 create_redis_engine/scan_iter/close 路径测试，验证同步集群客户端的构造与命令路由。
• 增加并发首次调用 create_redis_engine 的测试，确认共享客户端只被创建一次。

[github-actions]

CLA Assistant Lite bot All contributors have signed the CLA ✍️ ✅

[Wenvii]

I have read the CLA Document and I hereby sign the CLA

[helloopenworld]

AI Code Review

内部没有使用动态方法。因此，__getattr__ 的局限性是潜在的，仅在外部调用者使用 storage.<cmd>() 时才会显现。值得标记为警告，但严重程度较低。

让我敲定最终结论。我还要注意一下关于 scan_iter 覆盖范围的测试质量——测试伪造的 scan_iter 忽略了 match 参数并返回硬编码的键，因此它并没有真正验证扫描是否遵守了模式匹配过滤（它只验证了 match 是否被转发了，而不是被使用了）。这是一个测试有效性差距，但伪造对象确实断言了 scan_calls 包含该模式。去重逻辑已被覆盖。可以接受，但测试命名为“scans_all_cluster_keys”，却使用了一个无法验证跨节点扫描的伪造对象。存在轻微的测试有效性差距。

让我写下审查报告。## 发现的问题

⚠️ Warning

• trpc_agent_sdk/storage/_redis_cluster.py:399（继承自 RedisStorage.__getattr__，_redis.py:244）：动态命令方法在 Cluster 存储上会静默失效
  • RedisClusterStorage 继承了父类的 __getattr__，该方法以 self._redis_pool 作为前置条件，但 Cluster 存储从不创建 _redis_pool（其值恒为 None）。任何通过 storage.<cmd>(...) 形式调用的动态命令都会因 if not self._redis_pool: return 直接返回 None 而不执行，也不报错。建议在子类中重写 __getattr__ 或基于 self._redis_client 走 create_db_session() 路径，避免对外暴露失效的动态接口。

💡 Suggestion

• tests/storage/test_redis_cluster_storage.py:89-98：_FakeClusterClient.scan_iter 忽略 match/count 语义，硬编码返回固定键
  • 该 fake 只是断言 match 被转发（scan_calls），并未验证扫描结果是否真的按 pattern 过滤或跨节点去重，测试名 test_query_scans_all_cluster_keys_and_deduplicates 的“跨节点扫描”部分实际未被覆盖。可让 fake 依据 match 做简单 glob 过滤，使去重与扫描语义同时被验证。

总结

整体风险较低：新增的 Cluster 存储与服务子类在主调用路径（execute_command/query/delete/close）上正确覆盖了 KEYS→SCAN 改写、字节解码与共享客户端生命周期，无 Critical 问题。主要遗留点是继承自父类的 __getattr__ 在 Cluster 存储上会静默失效，建议修复以避免外部调用方踩坑。

测试建议

• 补充一个 decode_responses=False 下 query 对 list/set/zset 类型返回值解码一致性的用例（当前仅覆盖 string/hash）。
• 补充 execute_command 对非 keys 的单 key 命令（如 hset/get）在 Cluster 客户端上正常路由的用例，避免后续回归。