简体中文 | English

matterloop-integration-redis

这个包为 MatterLoop 提供持久化 Loop 检查点、任务队列、运行记录和事件流。它不把 Redis 变成“万能后端”：长期记忆和 Worker 仍由应用选择并装配。

pip install matterloop-integration-redis

五个适配器，五份职责

RedisCheckpointStore 保存精确恢复所需的计划、步骤游标、人工反馈、待验证执行结果和 revision。 RedisRunRepository 保存给控制面查询的状态与结果；两者数据用途不同，不能相互替代。

装配

from matterloop_integration_redis import (
    RedisCheckpointStore,
    RedisConfig,
    RedisContextStore,
    RedisEventPublisher,
    RedisQueueBackend,
    RedisRunRepository,
)
from matterloop_runtime import ContextRetentionPolicy

config = RedisConfig(prefix="matterloop:{prod}", lease_seconds=300)
checkpoints = RedisCheckpointStore(client=redis_client, config=config, codec=None)
contexts = RedisContextStore(
    redis_client,
    ContextRetentionPolicy(86_400, 604_800, 604_800),
    config,
)
queue = RedisQueueBackend(client=redis_client, config=config, codec=None)
runs = RedisRunRepository(client=redis_client, config=config, codec=None)
events = RedisEventPublisher(
    client=redis_client,
    config=config,
    checkpoint_codec=None,
)

redis_client 由应用创建并配置 TLS、认证、超时和连接池。五个适配器可以共享它，但都不会关闭它。

RedisConfig(prefix, lease_seconds, event_max_length) 只保存非敏感行为配置：默认值分别为 "matterloop"、60.0 秒和 10_000 条近似事件。连接 URL、用户名和密码不应进入配置对象。

检查点字段与 CAS 语义

save() 在一段 Lua 中读取当前值、校验 schema 与 run_id、比较 revision，再写入新快照。成功 返回 expected_revision + 1，但不会原地修改调用方的 context.revision；控制器应采用返回值。 版本不一致抛 CheckpointConflictError，损坏 JSON、非法 UTF-8、未知 checkpoint schema 或异常 Redis 返回值抛 RedisPayloadError。load() 对不存在的 Key 返回 None。

Redis 命令超时具有“结果未知”语义：Lua 可能已经提交。调用方不能直接重放副作用，应先重新 load() 并核对 revision。检查点没有 TTL、列举或删除 API，也不与 RunRecord 或事件 Stream 形成跨 Key 事务；数据保留、清理和 Outbox 由部署方负责。

队列不是“恰好一次”

Worker 调用 lease() 取得任务，成功后 acknowledge()，可重试失败则 release()。租约过期的任务 只会在下一次 lease() 时被回收；当前接口没有续租、最大尝试或死信队列。执行时间可能超过租约 时，要么提高租约并预留时钟漂移，要么把任务拆小。

网络超时发生时，Lua 可能已经提交。外部写操作必须带幂等键，最终运行状态必须通过 RunRepository.compare_and_set() 提交。cancel() 只能阻止等待中的任务或标记租用中的任务， 不会中断已经运行的 Python 代码。

Key 布局

所有 Key 都位于 <prefix> 下：

<prefix>:queue:pending          List
<prefix>:queue:delayed          Sorted Set
<prefix>:queue:jobs             Hash
<prefix>:queue:leases           Hash
<prefix>:queue:lease-expiry     Sorted Set
<prefix>:checkpoints:<run_id>   String (Core 当前 checkpoint 结构，revision CAS)
<prefix>:runs:index             Sorted Set
<prefix>:runs:<run_id>          String (versioned JSON)
<prefix>:events:<run_id>        Stream

检查点、运行记录与事件都没有 TTL 或删除 API。归档、删除、容量告警和数据保留属于部署方职责。

Redis Cluster

队列脚本、仓储 CAS 和批量读取会访问多个 Key，因此它们必须落在同一个 hash slot；检查点 CAS 只访问自身的一个 Key。默认前缀在 Cluster 中可能触发 CROSSSLOT，应使用固定 hash tag，例如 matterloop:{prod}。需要横向分片时，请按租户配置多个前缀/客户端，或实现不同后端。

事件与敏感数据

RedisCheckpointStore 和 RedisEventPublisher 都可能保存目标、模型输出、metadata、工具结果和 人工反馈。本包不会自动调用 Redactor。生产环境应设置 Redis ACL、TLS、静态加密、保留期限、 租户隔离和 payload 大小限制。

读取事件使用排他 Stream 游标 after。Stream 被裁剪后，旧游标不会报告缺口，只会从仍存在的 下一条开始；需要不可丢失审计时，不要依赖近似裁剪的 Redis Stream 作为唯一记录。

协议与错误

AsyncRedisClient 是最小结构协议，需要 eval、get、mget、zrevrange、xadd、xrange 和 aclose。适配器只等待这些异步方法，不读取连接配置。RedisPayloadCodec 与 LoopCheckpointCodec 使用严格、版本化 JSON；未知版本或损坏内容抛 RedisPayloadError。 网络、ACL、超时和 Cluster 错误由底层客户端原样传播，本包不自动重试。

构造入口为 RedisCheckpointStore(client, config, codec)、 RedisQueueBackend(client, config, codec)、RedisRunRepository(client, config, codec) 和 RedisEventPublisher(client, config, checkpoint_codec)。完整队列控制面和关闭顺序见 企业集成指南。