Flue Valkey 持久化实战：用 @flue/redis 适配器把 Agent 会话状态接入 Valkey-育师

Flue Valkey 持久化实战：用 @flue/redis 适配器把 Agent 会话状态接入 Valkey

【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue

Flue 的 Node 目标项目可以通过官方@flue/redis适配器把 Agent 的规范会话流、不可变附件与已接受的提交持久化到 Valkey，配置方式是在源码根目录落一个实现自定义 runner 的db.ts。本篇以仓库中的 Valkey 数据库蓝图（blueprints/database--valkey.md）为主体，结合 packages/redis 的源码实现，完整讲解从部署选型、db.ts编写、启动迁移到验证与升级的全流程。读完后你应当能独立完成一个可运行、可验证、可升级的 Valkey 持久化集成，并理解迁移钩子、inspectServer检查与keyPrefix隔离在底层是如何执行的。

蓝图定位：这是 flue add 返回给编码代理的实现指南

在展开细节前先说明这篇文档在 Flue 仓库中的角色。blueprints/ 目录存放的是flue add与flue update两个 CLI 命令返回的 Markdown 实现指南——Blueprint 是写给 AI 编码代理的指南，不是 npm 包，也不是运行时抽象；CLI 负责拉取并输出指南，由编码代理去修改用户项目。按 blueprints/README.md 的命名规则，database--valkey.md对应的 slug 是valkey，即flue add database valkey/flue update database valkey会返回这份完整指南。

该蓝图的 JSON frontmatter 为：

{ "kind": "database", "version": 1, "website": "https://valkey.io" }

其中version: 1标识完整蓝图契约的版本（见文末 Upgrade Guide）。所有database类型的蓝图产出的都是一个位于源码根目录、默认导出PersistenceAdapter的db.ts，而不是sandboxes/下的文件。

蓝图开篇明确划定了支持边界：Valkey 实现了本适配器所用的 Redis 协议与命令面，因此@flue/redis可用于 Valkey；但该蓝图仅针对 Valkey 提供支持，不要据此推断所有自称"Redis 兼容"的服务都受支持。

前置检查一：运行时目标与部署形态

先确认项目不是 Cloudflare 目标

db.ts适配器是仅面向 Node 目标的持久化手段。Cloudflare 目标自动使用 Durable Object SQLite，并在构建期直接拒绝项目自有的db.ts。如果项目面向 Cloudflare，应停止操作并告知用户：没有可添加的内容。packages/redis/README.md 的 "Target support" 一节与蓝图表述一致：该适配器仅支持 Node.js，Cloudflare 项目使用 Durable Object SQLite。

Valkey 部署形态的硬性要求

蓝图要求使用持久化的独立 Valkey 服务器或托管的单分片（single-shard）端点，且必须满足：

maxmemory-policy noeviction（禁止逐出）；
Valkey Cluster 与纯缓存（cache-only）配置不受支持；
按恢复目标启用 AOF（显式 fsync 策略）和/或持久化快照。

这里有一个容易误解的点：noeviction只保证已确认的写入不被内存压力逐出，并不保证写入在服务端整体丢失后仍然存活——跨服务端丢失的持久性必须由 AOF/快照策略提供。

为什么不支持 Cluster：从 key 命名结构看

packages/redis/src/redis-keys.ts 中的RedisKeys类揭示了原因：Flue 的 key 采用前缀:种类:base64url 编码段的结构（如submission:<base64url(id)>、session-unsettled:<base64url(sessionKey)>），且一次业务操作往往横跨多个独立 key 的原子脚本。这类 key 分布并不是为 Cluster 哈希槽设计的 schema，跨 key 的 Lua 原子操作在分片拓扑下无法保证落在同一节点。因此蓝图与 packages/redis/README.md 均明确：仅支持独立服务器与托管单分片端点。

前置检查二：检查项目现状并安装依赖

蓝图要求的检查步骤：

阅读项目内的本地指令（AGENTS.md及同类文件）；
探测项目使用的包管理器；
按顺序选择第一个存在的源码根目录：<root>/.flue/→<root>/src/→<root>/；
检查是否已存在db.ts；若已存在，替换前必须先与用户确认；
检查项目的密钥（secret）约定。

随后用项目的包管理器安装两个依赖：@flue/redis与官方redis@^5.12.1（node-redis）客户端。需要强调：@flue/redis不捆绑任何生产客户端，项目自己拥有凭证、TLS、超时、重连行为与拓扑选择。packages/redis/package.json 中redis仅作为该包的 devDependency（仓库开发环境使用^6.1.0）出现，不会进入运行时产物；packages/redis/README.md 的示例安装命令为pnpm add @flue/redis redis。

创建 db.ts：完整的 runner 实现

在选定的源码根目录写入如下完整 runner，作为db.ts的内容。首行// flue-blueprint: database/valkey@1是蓝图生成的主文件标记，用于后续flue update时识别受蓝图管理的文件（标记规范见 blueprints/README.md）：

// flue-blueprint: database/valkey@1 import { redis } from '@flue/redis'; import { createClient } from 'redis'; const client = createClient({ url: process.env.VALKEY_URL }); await client.connect(); export default redis({ command: (command, args = []) => client.sendCommand([command, ...args.map(String)]), eval: (script, keys, args = []) => client.eval(script, { keys, arguments: args.map(String), }), pipeline: async (commands) => { const multi = client.multi(); for (const { command, args = [] } of commands) { multi.addCommand([command, ...args.map(String)]); } const results = await multi.exec(); for (const result of results) { if (result instanceof Error) throw result; } return results; }, close: () => client.close(), });

redis(runner, options?)工厂函数在 packages/redis/src/redis-adapter.ts 中定义，接收 runner 与可选的RedisOptions，返回一个PersistenceAdapter。runner 各字段的契约定义在 packages/redis/src/redis-runner.ts：

字段	类型	说明
`command(command, args)`	必填	执行归一化的 Redis 命令，返回原始响应
`eval(script, keys, args)`	必填	执行 Lua 脚本，`keys`与`args`严格分离（对应`KEYS`/`ARGV`）
`pipeline(commands)`	可选	批量暂存大型不可变代（generation）时使用；未提供时适配器会退化为逐条`command`串行执行（见 redis-adapter.ts）
`close()`	必填	客户端生命周期收尾

三个实现要点值得结合源码展开：

1.String()强转是无损的。适配器永远不会把二进制参数透传给 runner——附件字节在存储边界就已完成 base64 编码（packages/redis/src/attachment-store.ts 中encodeBytes/decodeBytes使用 base64 跨 runner 接缝，读取时还通过 digest 校验兜底任何落盘损坏），因此 runner 把每个参数经String()强转不会丢失任何信息。RedisArgument类型本身就是string | number（redis-runner.ts）。

2. pipeline 的"每命令一个归一化结果"契约。node-redis 的multi().exec()会按[error, result]元组返回，其中任何一项为Error即代表该命令失败。蓝图 runner 在返回前遍历结果、抛出第一个Error，恰好满足这一契约；适配器侧也会二次校验结果数量与错误形态（redis-adapter.ts），但 runner 仍负责把驱动特有的失败形态归一化。

3. RESP2/RESP3 双协议兼容由适配器消化。Valkey 与不同版本的 node-redis 可能返回不同 RESP 协议的响应形状（例如HGETALL在 RESP2 下是扁平的 field/value 数组、在 RESP3 下是 map）。适配器内置的hash()归一化同时接受两种形状（packages/redis/src/conversation-store.ts 有明确注释），因此 runner 无需感知协议版本差异。

凭证规则：不要硬编码或凭空发明连接串。从项目的密钥系统中读取VALKEY_URL或其既有等价项，绝不提交凭证到仓库。

启动迁移：migrate() 与格式版本机制

Flue 会发现db.ts并在服务器启动时调用适配器的migrate()钩子——没有独立的迁移命令。migrate()的完整行为可以在 packages/redis/src/redis-adapter.ts 中逐行印证，它做三件事：

检查服务器（inspectServer未禁用时，见下节）；
幂等地初始化格式版本元数据 key：读取flue:metahash 中的format_version字段。若缺失，先探测旧字段schema_version——若其值为8（Nightly 时代遗留，存储形状与 format 1 逐字节一致），则直接改标签为format_version: 1并删除旧字段，不做任何数据重写；若旧字段为其他值则拒绝。若两字段都缺失，则用SCAN（COUNT 100分页）检查前缀下是否已有数据：有数据但无版本标记的旧存储按unversioned拒绝；确认为空库才用HSETNX落版本标记。先写新字段、再删旧字段，保证中断不会留下无标记的存储。
拒绝来自更新格式的数据：已记录的format_version与当前版本不一致（尤其是更大值，意味着由更新的 Flue 写入）时抛出PersistedFormatVersionError。当前运行时格式版本为1，定义在 packages/runtime/src/format-version.ts。

inspectServer：CONFIG GET 与 INFO 双通道验证

默认启用inspectServer时，启动期会验证两项部署属性，任一无法验证即启动失败（TypeError）：

Cluster 已禁用：先CONFIG GET cluster-enabled，失败则回退到INFO cluster并匹配cluster_enabled:1/0；检测到yes/1直接报 "Redis Cluster is not supported"；
maxmemory-policy为noeviction：先CONFIG GET maxmemory-policy，失败则回退到INFO memory并匹配maxmemory_policy:行。

实现见 redis-adapter.ts 的 inspectServer()。

inspectServer: false的适用条件：仅当托管单分片服务商同时拒绝CONFIG与INFO两条检查命令时才使用，且前提是你已独立核实过"Cluster 关闭 + noeviction"这两项要求。关闭检查等于把部署正确性责任完全移交给运维侧，验证步骤必须相应补偿（见下文第 3 条）。

keyPrefix：Flue 命名空间隔离

通过redis(runner, { keyPrefix: '...' })的第二个参数隔离 Flue 的 key，默认前缀为flue（redis-keys.ts 中RedisKeys构造器会去掉尾部冒号并拒绝空前缀）。建议为每个应用或租户使用稳定且唯一的前缀，或者直接使用独立的 Valkey database。两个关键行为：

前缀变更只是让 Flue 指向一个独立的新空命名空间，不会迁移已有数据；
生产清理走的是维护中的 set/sorted set，从不扫描整个 keyspace（packages/redis/README.md "Storage model"）。

What gets stored：存储模型与边界

蓝图明确列出适配器存储的内容边界：

存储的：

规范（canonical）追加-only 会话流——它是唯一的事实记录（sole transcript），读取时从流头开始完整回放；回放加速与持久化日志压缩是推迟项，当前不做；
不可变外部附件——以 base64 文本写入 hash 并附带 digest，读取时校验（attachment-store.ts 的PUT脚本采用EXISTS先检 +HSET+SADD的首写者胜语义，同 id 重复写入时校验 ref 与字节一致性，不一致抛AttachmentConflictError）；
已接受的直发（direct）与调度（dispatched）提交及恢复日志（recovery journals）——提交生命周期状态（queued/running/joining/joined/terminalizing/settled 等）以 hash 存权威元数据、以 sorted set 维护顺序，准入（admission）、认领（claim）、租约（lease）等状态迁移由 Lua 脚本原子化（redis-adapter.ts 中admitSubmissionScript、claimSubmissionScript、lifecycleScript等脚本族）。

不存储的：sandbox 文件、外部 API 副作用、凭证、以及任何应用业务数据。会话在实例生命周期内只追加，没有按会话删除的能力；整实例级别的流/附件删除方法是低层原语，不属于公开编排接口。

从 packages/redis/README.md 的 "Storage model" 一节还可以补充一个实现事实：Redis Lua 在后续命令失败时不会回滚已执行的命令，因此脚本会先校验 key 类型、尽量在权威状态之前预留索引容量，并在正常操作中以权威 hash 为准修复索引；noeviction大幅限制了部分迁移的风险，但多命令脚本执行中内存耗尽仍需运维级恢复——为最大暂存代（staging generation）及其索引保留足够的内存余量。

验证清单（完整五步）

蓝图给出的验证步骤应逐条执行，且第 5 条是红线：

对配置好的 Node 目标做类型检查与构建（vite build），确认db.ts被 Flue 发现；
将VALKEY_URL指向一个**一次性（throwaway）**的持久化独立或托管单分片 Valkey 部署，且已配置noeviction；
启动服务器并确认迁移成功。若你禁用了inspectServer，必须独立核验 Cluster 关闭与逐出策略为noeviction；
创建状态、重启 Flue 服务器、确认状态重新加载。另外单独测试所选的 AOF/快照恢复——进程重启只能证明"进程内可重读"，不能证明跨 Valkey 服务端丢失的持久性；
不要使用生产数据库做验证。

更新既有集成与 Upgrade Guide

当对一个已有集成执行更新（flue update database valkey返回同一份完整指南）时，蓝图要求的操作是：检查现有实现并与当前完整蓝图逐段比对，应用所有相关变更并保留自定义部分，然后在主标记文件（即db.ts）中补上或更新标记// flue-blueprint: database/valkey@N；当标记缺失时，这一比对是强制要求。这一"无标记时的更新契约"规范同样定义在 blueprints/README.md。

Version 1 — 2026-06-14

Initial version.（首个版本，无历史差异可展示。）

小结与适用前提

适用前提：Flue 项目的Node 目标；Valkey 为独立部署或托管单分片；noeviction+ 按恢复目标配置的 AOF/快照；
交付物只有一个文件：源码根目录的db.ts，默认导出redis(runner, options?)的结果，首行带蓝图标记；
运行时行为（迁移、格式版本守卫、服务器检查、原子提交生命周期）全部由@flue/redis在 packages/redis/src/redis-adapter.ts 中实现，db.ts只负责把官方redis客户端桥接成 runner；
深入阅读路径：packages/redis/README.md（部署要求与存储模型）、packages/redis/src/redis-runner.ts（runner 契约）、packages/runtime/src/format-version.ts（格式版本规则）、blueprints/README.md（蓝图机制）。

【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考