Electric Agents 实体流查询模式实战基于 queryOnce 与 useLiveQuery 的 manifests、childStatus 与时间线数据访问指南【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric本文面向在 Electric Agentselectric-ax/agents-runtime中编写实体处理器、测试、示例或 CLI 工具的开发者系统讲解如何通过durable-streams/state/db提供的queryOnce与useLiveQuery直接、类型安全地读取实体流Entity Stream中的内建集合。读完本文你将掌握查找已生成子实体、列出被观察实体、读取 manifest 状态与子实体状态、在测试与处理器中读取子流最新 runs以及为 UI/聊天界面搭建实时实体时间线查询的完整实战方案。一、背景实体流与 db.collectionsElectric Agents 的每个实体Entity都对应一条实体流Entity Stream。运行时通过createEntityStreamDB实现在 entity-stream-db.ts在实体流之上构建一个带类型的 StreamDB它把实体事件流中的各类事件物化为 TanStack DB 风格的集合Collection统一挂在db.collections.*下。SKILL 文件 SKILL.md 明确指出读取这些内建集合时应直接使用类型化查询而非为显而易见的读操作添加一次性辅助函数。内建集合一览根据 collections.md实体流暴露的内建集合包括集合含义manifests子实体child、观察observe、共享状态shared state、效果effect等资源的持久状态childStatus子实体生命周期/状态的当前行wakes由服务器追加的唤醒wake行inbox入站用户/系统消息runsAgent 运行run行texts运行对应的文本消息行textDeltas增量文本分块toolCalls工具调用行steps步骤行errors错误行entityStopped服务器写入的停止行这些集合的完整 schema 定义集中在 entity-schema.ts行类型直接来自 schema因此 SKILL 特别强调不要对db.collections.*做类型转换cast以免破坏类型安全。manifests 是资源状态而非命令日志manifests集合存储的是持久资源状态不是追加式的命令日志。其常见 kind 有child、observe、shared-state、effect。schema 中还包含source、attachment、context、schedulecron与future_send、goal等更多 kind见 entity-schema.ts。因此按manifests.kind过滤就是按资源类别查询。二、查询基础queryOnce 与统一导入所有查询示例统一假定以下导入import { eq, queryOnce } from durable-streams/state/dbqueryOnce(q ...)用于一次性读取在回调里用查询构建器描述“从哪个集合、按什么条件、取什么”返回 Promiseresolve 为查询结果数组。eq是相等条件的比较符用于where过滤。对于 UI/聊天这类需要响应式更新的场景则使用useLiveQuery(...)见下文第六节。SKILL.md 中给出的原则是一次性读取优先用queryOnce(...)响应式 UI 用useLiveQuery(...)或共享的实时查询集合在运行时、测试、示例与 CLI 代码中当读取天然具备“查询形状”时优先直接查询而非原始toArray全量扫描。三、查找一个已生成的子实体在父实体中最常见的需求是判断某个子实体是否已经通过ctx.spawn生成已生成则直接ctx.send发消息未生成则先spawn再通信。import { manifestChildKey } from electric-ax/agents-runtime const child await queryOnce((q) q .from({ manifests: ctx.db.collections.manifests }) .where(({ manifests }) eq(manifests.key, manifestChildKey(worker, child-1)) ) .findOne() ) if (child?.kind child) { ctx.send(child.entity_url, payload) } else { await ctx.spawn(worker, child-1, args) }要点拆解ctx.db实体处理器上下文中的实体 DB类型为EntityStreamDB见 entity-stream-db.tscollections.manifests是类型化集合。manifestChildKey(entityType, id)稳定键构建器。源码实现在 manifest-helpers.ts其生成格式为child:${entityType}:${id}即这里查询的键是child:worker:child-1。使用键构建器而非手拼字符串可以保证与spawn写入 manifest 时使用的键完全一致。findOne()取满足条件的单行若不存在返回undefined配合child?.kind child做存在性判断。ctx.spawn(type, id, args?, opts?)签名定义在 types.ts其中opts.observe为false时可实现 fire-and-forget 的高扇出模式父实体不订阅子实体流。四、列出被观察的实体通过ctx.observe(...)注册的观察关系同样以observe类型的 manifest 行落库。要列出当前实体观察了哪些实体只需按manifests.kind过滤const observed await queryOnce((q) q .from({ manifests: ctx.db.collections.manifests }) .where(({ manifests }) eq(manifests.kind, observe)) )从 schema 看source类型的 manifest 行携带sourceType、sourceRef与config字段entity-schema.ts可用于区分观察来源的具体形态。在测试 runtime-dsl.test.ts 中可以看到相似的真实用例处理器读取自身全部 manifest 行后遍历找出kind source sourceType entity的条目动态维护“被观察子实体”集合——这印证了“直接查询 manifests 即可驱动运行时逻辑”的实践。五、读取当前 manifest 状态要一次性读取实体当前的全部 manifest 行最直接的写法是不带where的全量查询const manifestRows await queryOnce((q) q.from({ manifests: db.collections.manifests }) )得到的是一个数组。文档给出了一条重要的工程约束分组/投影等加工在调用点call site完成即可不要为了显而易见的读操作添加一次性 helper——除非该 helper 真正编码了领域语义例如下面第六节的共享时间线投影。换言之queryOnce查询本身就已经是 API过多的便捷读包装只会让调用面变乱。六、读取当前子实体状态childStatus集合保存每个子实体的生命周期状态行schema 见 entity-schema.tsstatus取值包括spawning、running、idle、paused、stopping、stopped、killed。读取全部子实体当前状态const statuses await queryOnce((q) q.from({ childStatus: db.collections.childStatus }) )返回的每一行包含entity_url、entity_type与status可按需在调用点按entity_url建立映射。七、在测试或处理器中读取子流的最新 runs当测试或父处理器持有某个子实体的句柄如ctx.spawn返回的EntityHandle时可以直接基于该句柄的 DB 读取其runs集合取数组末尾即最新一次运行const runs await queryOnce((q) q.from({ runs: childHandle.db.collections.runs }) ) const latestRun runs[runs.length - 1]这是测试中最常用的断言手段spawn 一个子实体 → 等待其运行 → 用该模式取最新 run 并校验其statusstarted/completed/failed与finish_reason。runs行还通过_timeline_order虚拟列携带时间线顺序由运行时在物化事件时写入见 entity-stream-db.ts因此数组顺序即事件时间顺序。八、实时 UI 查询共享的实体时间线视图对于需要把 runs、inbox、wakes 以及相关实体一起呈现的 UI/聊天界面不应自行拼装多个queryOnce而应复用共享的实体视图查询。该查询定义在src/entity-timeline.ts即仓库中的 entity-timeline.ts。const timelineQuery createEntityIncludesQuery(db) const { data [] } useLiveQuery(timelineQuery, [timelineQuery]) const timeline data[0]使用要点createEntityIncludesQuery(db)返回一个查询构建器函数它内部调用ensureEntityTimelineIndexes确保所需索引存在并从种子集合出发将 runs含 textDeltas 拼接的文本、toolCalls、steps、errors 与按步骤聚合的 tokens、inbox、wakes、signals、contextInserted/contextRemoved 以及实体清单投影为一张统一的EntityTimelineData类型定义与归一化逻辑都在 entity-timeline.ts。useLiveQuery让查询保持响应式数据变化时data自动更新。同文件还导出createEntityTimelineQuery(db, opts?)它是按$key合并的时间线行视图支持inboxMode: processed | all以及自定义数据源合并。文档给出的选型结论非常明确一次性读取用queryOnce(...)只有真正需要完整响应式视图模型时才使用共享的 live query避免为简单读取引入不必要的反应式计算成本。九、底层机制与补充键构建器、类型与写路径为了让上述查询模式更可依赖几个底层事实值得了解manifest 键构建器当需要稳定键时优先使用 manifest-helpers.ts 中的键构建器manifestChildKey(entityType, id)→child:${entityType}:${id}manifestSharedStateKey(id)→shared-state:${id}manifestEffectKey(functionRef, id)→effect:${functionRef}:${id}manifestObserveKey(entityUrl)在 collections.md 中列出类型安全来自 schemaEntityStreamDB将内建集合与实体自定义状态集合合并mergedCollections并为自定义集合自动生成${name}_insert/${name}_update/${name}_deleteCRUD actions见 entity-stream-db.ts。因此读取侧的行类型、写入侧的 schema 校验都来自同一定义查询时不需要也不应该做类型断言。写入走事件 helper而非手拼事件信封SKILL.md 明确警告写入实体或共享状态流时应使用类型化的事件 helper/服务端 API不要手工构造流事件信封stream event envelope。查询侧专注读取写入侧交给运行时封装避免破坏事件格式与顺序语义。十、实践原则小结直接查集合优先queryOnce(...)针对db.collections.*的类型化查询避免无关紧要的toArray全量扫描与一次性读包装。查询即 API只有承载真实语义如 manifest 键构建、类型化事件/写入 helper、产品级共享投影如实体时间线的代码才值得提炼成 helper。保留 schema 类型不要 castdb.collections.*行类型由 schema 推导。按场景选择查询 APIone-shot 读用queryOnce完整响应式视图才用createEntityIncludesQueryuseLiveQuery。测试与处理器一视同仁子实体最新运行、被观察实体集合等运行时状态都可以用同一套查询模式在测试中直接断言参考 runtime-dsl.test.ts 中的queryOnce用法。如需更完整的集合字段与 manifest 行 kind 清单可继续阅读 collections.md要深入理解查询构建器与 StreamDB 基础可先阅读所属 Skill 的入口说明 SKILL.md。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Electric Agents 实体流查询模式实战：基于 queryOnce 与 useLiveQuery 的 manifests、childStatus 与时间线数据访问指南
Electric Agents 实体流查询模式实战基于 queryOnce 与 useLiveQuery 的 manifests、childStatus 与时间线数据访问指南【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric本文面向在 Electric Agentselectric-ax/agents-runtime中编写实体处理器、测试、示例或 CLI 工具的开发者系统讲解如何通过durable-streams/state/db提供的queryOnce与useLiveQuery直接、类型安全地读取实体流Entity Stream中的内建集合。读完本文你将掌握查找已生成子实体、列出被观察实体、读取 manifest 状态与子实体状态、在测试与处理器中读取子流最新 runs以及为 UI/聊天界面搭建实时实体时间线查询的完整实战方案。一、背景实体流与 db.collectionsElectric Agents 的每个实体Entity都对应一条实体流Entity Stream。运行时通过createEntityStreamDB实现在 entity-stream-db.ts在实体流之上构建一个带类型的 StreamDB它把实体事件流中的各类事件物化为 TanStack DB 风格的集合Collection统一挂在db.collections.*下。SKILL 文件 SKILL.md 明确指出读取这些内建集合时应直接使用类型化查询而非为显而易见的读操作添加一次性辅助函数。内建集合一览根据 collections.md实体流暴露的内建集合包括集合含义manifests子实体child、观察observe、共享状态shared state、效果effect等资源的持久状态childStatus子实体生命周期/状态的当前行wakes由服务器追加的唤醒wake行inbox入站用户/系统消息runsAgent 运行run行texts运行对应的文本消息行textDeltas增量文本分块toolCalls工具调用行steps步骤行errors错误行entityStopped服务器写入的停止行这些集合的完整 schema 定义集中在 entity-schema.ts行类型直接来自 schema因此 SKILL 特别强调不要对db.collections.*做类型转换cast以免破坏类型安全。manifests 是资源状态而非命令日志manifests集合存储的是持久资源状态不是追加式的命令日志。其常见 kind 有child、observe、shared-state、effect。schema 中还包含source、attachment、context、schedulecron与future_send、goal等更多 kind见 entity-schema.ts。因此按manifests.kind过滤就是按资源类别查询。二、查询基础queryOnce 与统一导入所有查询示例统一假定以下导入import { eq, queryOnce } from durable-streams/state/dbqueryOnce(q ...)用于一次性读取在回调里用查询构建器描述“从哪个集合、按什么条件、取什么”返回 Promiseresolve 为查询结果数组。eq是相等条件的比较符用于where过滤。对于 UI/聊天这类需要响应式更新的场景则使用useLiveQuery(...)见下文第六节。SKILL.md 中给出的原则是一次性读取优先用queryOnce(...)响应式 UI 用useLiveQuery(...)或共享的实时查询集合在运行时、测试、示例与 CLI 代码中当读取天然具备“查询形状”时优先直接查询而非原始toArray全量扫描。三、查找一个已生成的子实体在父实体中最常见的需求是判断某个子实体是否已经通过ctx.spawn生成已生成则直接ctx.send发消息未生成则先spawn再通信。import { manifestChildKey } from electric-ax/agents-runtime const child await queryOnce((q) q .from({ manifests: ctx.db.collections.manifests }) .where(({ manifests }) eq(manifests.key, manifestChildKey(worker, child-1)) ) .findOne() ) if (child?.kind child) { ctx.send(child.entity_url, payload) } else { await ctx.spawn(worker, child-1, args) }要点拆解ctx.db实体处理器上下文中的实体 DB类型为EntityStreamDB见 entity-stream-db.tscollections.manifests是类型化集合。manifestChildKey(entityType, id)稳定键构建器。源码实现在 manifest-helpers.ts其生成格式为child:${entityType}:${id}即这里查询的键是child:worker:child-1。使用键构建器而非手拼字符串可以保证与spawn写入 manifest 时使用的键完全一致。findOne()取满足条件的单行若不存在返回undefined配合child?.kind child做存在性判断。ctx.spawn(type, id, args?, opts?)签名定义在 types.ts其中opts.observe为false时可实现 fire-and-forget 的高扇出模式父实体不订阅子实体流。四、列出被观察的实体通过ctx.observe(...)注册的观察关系同样以observe类型的 manifest 行落库。要列出当前实体观察了哪些实体只需按manifests.kind过滤const observed await queryOnce((q) q .from({ manifests: ctx.db.collections.manifests }) .where(({ manifests }) eq(manifests.kind, observe)) )从 schema 看source类型的 manifest 行携带sourceType、sourceRef与config字段entity-schema.ts可用于区分观察来源的具体形态。在测试 runtime-dsl.test.ts 中可以看到相似的真实用例处理器读取自身全部 manifest 行后遍历找出kind source sourceType entity的条目动态维护“被观察子实体”集合——这印证了“直接查询 manifests 即可驱动运行时逻辑”的实践。五、读取当前 manifest 状态要一次性读取实体当前的全部 manifest 行最直接的写法是不带where的全量查询const manifestRows await queryOnce((q) q.from({ manifests: db.collections.manifests }) )得到的是一个数组。文档给出了一条重要的工程约束分组/投影等加工在调用点call site完成即可不要为了显而易见的读操作添加一次性 helper——除非该 helper 真正编码了领域语义例如下面第六节的共享时间线投影。换言之queryOnce查询本身就已经是 API过多的便捷读包装只会让调用面变乱。六、读取当前子实体状态childStatus集合保存每个子实体的生命周期状态行schema 见 entity-schema.tsstatus取值包括spawning、running、idle、paused、stopping、stopped、killed。读取全部子实体当前状态const statuses await queryOnce((q) q.from({ childStatus: db.collections.childStatus }) )返回的每一行包含entity_url、entity_type与status可按需在调用点按entity_url建立映射。七、在测试或处理器中读取子流的最新 runs当测试或父处理器持有某个子实体的句柄如ctx.spawn返回的EntityHandle时可以直接基于该句柄的 DB 读取其runs集合取数组末尾即最新一次运行const runs await queryOnce((q) q.from({ runs: childHandle.db.collections.runs }) ) const latestRun runs[runs.length - 1]这是测试中最常用的断言手段spawn 一个子实体 → 等待其运行 → 用该模式取最新 run 并校验其statusstarted/completed/failed与finish_reason。runs行还通过_timeline_order虚拟列携带时间线顺序由运行时在物化事件时写入见 entity-stream-db.ts因此数组顺序即事件时间顺序。八、实时 UI 查询共享的实体时间线视图对于需要把 runs、inbox、wakes 以及相关实体一起呈现的 UI/聊天界面不应自行拼装多个queryOnce而应复用共享的实体视图查询。该查询定义在src/entity-timeline.ts即仓库中的 entity-timeline.ts。const timelineQuery createEntityIncludesQuery(db) const { data [] } useLiveQuery(timelineQuery, [timelineQuery]) const timeline data[0]使用要点createEntityIncludesQuery(db)返回一个查询构建器函数它内部调用ensureEntityTimelineIndexes确保所需索引存在并从种子集合出发将 runs含 textDeltas 拼接的文本、toolCalls、steps、errors 与按步骤聚合的 tokens、inbox、wakes、signals、contextInserted/contextRemoved 以及实体清单投影为一张统一的EntityTimelineData类型定义与归一化逻辑都在 entity-timeline.ts。useLiveQuery让查询保持响应式数据变化时data自动更新。同文件还导出createEntityTimelineQuery(db, opts?)它是按$key合并的时间线行视图支持inboxMode: processed | all以及自定义数据源合并。文档给出的选型结论非常明确一次性读取用queryOnce(...)只有真正需要完整响应式视图模型时才使用共享的 live query避免为简单读取引入不必要的反应式计算成本。九、底层机制与补充键构建器、类型与写路径为了让上述查询模式更可依赖几个底层事实值得了解manifest 键构建器当需要稳定键时优先使用 manifest-helpers.ts 中的键构建器manifestChildKey(entityType, id)→child:${entityType}:${id}manifestSharedStateKey(id)→shared-state:${id}manifestEffectKey(functionRef, id)→effect:${functionRef}:${id}manifestObserveKey(entityUrl)在 collections.md 中列出类型安全来自 schemaEntityStreamDB将内建集合与实体自定义状态集合合并mergedCollections并为自定义集合自动生成${name}_insert/${name}_update/${name}_deleteCRUD actions见 entity-stream-db.ts。因此读取侧的行类型、写入侧的 schema 校验都来自同一定义查询时不需要也不应该做类型断言。写入走事件 helper而非手拼事件信封SKILL.md 明确警告写入实体或共享状态流时应使用类型化的事件 helper/服务端 API不要手工构造流事件信封stream event envelope。查询侧专注读取写入侧交给运行时封装避免破坏事件格式与顺序语义。十、实践原则小结直接查集合优先queryOnce(...)针对db.collections.*的类型化查询避免无关紧要的toArray全量扫描与一次性读包装。查询即 API只有承载真实语义如 manifest 键构建、类型化事件/写入 helper、产品级共享投影如实体时间线的代码才值得提炼成 helper。保留 schema 类型不要 castdb.collections.*行类型由 schema 推导。按场景选择查询 APIone-shot 读用queryOnce完整响应式视图才用createEntityIncludesQueryuseLiveQuery。测试与处理器一视同仁子实体最新运行、被观察实体集合等运行时状态都可以用同一套查询模式在测试中直接断言参考 runtime-dsl.test.ts 中的queryOnce用法。如需更完整的集合字段与 manifest 行 kind 清单可继续阅读 collections.md要深入理解查询构建器与 StreamDB 基础可先阅读所属 Skill 的入口说明 SKILL.md。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考