Electric Agents WakeEvent 与 Wake 配置完全指南理解实体处理器的唤醒机制【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric本篇技术指南围绕 Electric Agentselectric-ax/agents-runtime中的WakeEvent与Wake两个核心类型展开详细讲解实体entity处理器为何被调用、如何被调用以及如何在ctx.spawn()、ctx.observe()、ctx.observe(db(...))中配置唤醒条件。读完本文你将掌握WakeEvent的完整字段语义、inbox与wake两大唤醒类型及其子类型目录并能正确编写根据wake.type分支处理不同触发的实体处理器。WakeEvent 是什么WakeEvent描述了一个实体处理器entity handler被调用的原因。它作为第二个参数被传入处理器函数handler(ctx: HandlerContext, wake: WakeEvent) void | Promisevoid它的类型定义Source:electric-ax/agents-runtime如下type WakeEvent { source: string type: string fromOffset: number toOffset: number eventCount: number payload?: unknown summary?: string fullRef?: string }在仓库源码 types.ts 中可以看到该类型被原样定义并导出同时在 types.ts 中还有一组面向处理器的包装类型HandlerWakeInboxHandlerWake与OtherHandlerWake它们分别携带message含type、payload、from或wakeType字段并都保留原始的raw: WakeEvent方便处理器按需读取底层唤醒事件。字段详解FieldTypeDescriptionsourcestringURL or identifier of the stream that triggered the wake.typestringWake type. Usuallyinboxorwake; fallback webhook events can usetriggerEventormessage. See catalog.fromOffsetnumberStart offset of new events in the source stream.toOffsetnumberEnd offset (exclusive) of new events.eventCountnumberNumber of new events in this wake.payloadunknownOptional payload data associated with the wake. Shape depends ontype.summarystringOptional human-readable summary of the wake reason.fullRefstringOptional full reference identifier for the wake source.fromOffset/toOffset/eventCount共同刻画了一次唤醒所覆盖的源流事件范围一次唤醒覆盖源流上从fromOffset含到toOffset不含的一段连续 offset 区间eventCount表示这段区间内包含的新事件数量。这也是 Electric Agents 实现事件合并coalescing的基础——多个在实体繁忙期间到达的外部事件会被合并成一次唤醒但 offset 区间保持连续、不丢失。Wake-type catalog两种主类型处理器通常只会看到两个wake.type取值直接的收件箱消息到达时是inbox其余大多数非消息类触发都会被扁平化为wake具体细节由wake.payload携带。低层 webhook 回退场景下也可能直接暴露triggerEvent或在没有提供触发事件时暴露message。inbox外部消息落入了实体的收件箱——来源可以是ctx.send()、CLI 的electric agents send或任何直接的/sendHTTP 调用。FieldShapewake.sourceThefromfield of the message (sender identifier), or the entity URL if absent.wake.payloadThe message payload (any JSON-serialisable value).wake.summaryThemessage_typeif the sender set one.一个典型的发送端写法来自配套文档 waking-entities.mdctx.send(/assistant/peer, { text: hello })接收端的处理器会看到wake.type inbox并从wake.payload中取出消息内容。在源码中收件箱唤醒被建模为InboxHandlerWake见 types.ts它解析出message.type、message.payload与发送者message.from。wake针对任何非消息类触发而合成synthesised的唤醒。此时wake.payload是一个WakeMessagetype WakeMessage { timestamp: string source: string timeout: boolean changes: Array{ collection: string kind: insert | update | delete key: string } finished_child?: { url: string type: string run_status: completed | failed response?: string error?: string } other_children?: Array{ url: string type: string status: spawning | running | idle | paused | stopping | stopped | killed } }源码中WakeMessage被定义为OmitWakeEntry, key见 types.ts即它来自持久化在实体流上的 wake 条目。在 entity-timeline.ts 的buildWakeMessages函数中可以看到该结构的具体组装逻辑timestamp、source、timeout、changes是必带字段finished_child与other_children仅在对应场景存在时通过展开运算符附加到 payload 上。通过 payload 区分子类型检查 payload 即可区分具体的子类型Sub-kindProducerPayload markerChild finishedctx.spawn(..., { wake: { on: runFinished, includeResponse: true } })when the child completes or failspayload.finished_childis set (withrun_statusand optionalresponse)Observed changectx.observe(..., { wake: { on: change } })orobserve(db(...))payload.changesis non-emptyShared-state changeawait ctx.observe(db(...), { wake: { on: change } })payload.changesis non-empty,payload.sourceidentifies the shared-state streamCron firedA cron schedule entry on the entitys manifestpayload.sourceidentifies the schedule;payload.changesis emptyWebhook sourcesubscribe_webhook_sourcetool orclient.subscribeToWebhookSource()payload.type webhook_source_wakeandpayload.eventscontains matching webhook eventsScheduled sendAfuture_sendschedule firesArrives asinbox(notwake) — the schedule produces a message deliveryTimeouttimeoutMson achangewake config elapsed with no changespayload.timeout true,payload.changesis empty注意几个容易混淆的点Scheduled send 不是wakefuture_send计划触发时最终产生的是一条消息投递因此到达处理器时是inbox类型。Cron fired 与 Timeout 的区分两者changes都为空但 cron 的payload.source指向具体 schedule而 timeout 的payload.timeout true。webhook source 的特殊性它通过payload.type webhook_source_wake标记并携带完整的事件行。Webhook-source 唤醒的完整载荷Webhook-source 唤醒使用的是来自HydratedWebhookSourceWake的水合hydrated载荷type HydratedWebhookSourceWake { type: webhook_source_wake source: string sourceType: webhook endpointKey: string webhookKey: string subscription: { id: string bucketKey?: string params: Recordstring, unknown filterKey?: string reason?: string } bucket: string | null changes: Array{ collection: string kind: insert | update | delete key: string } events: WebhookEventRow[] missingEventKeys?: string[] }该类型在 webhook-sources.ts 中定义由buildHydratedWebhookSourceWake函数同文件 L409 附近在唤醒事件产生时组装events数组包含与订阅条件匹配的 webhook 事件行missingEventKeys用于标记查询时缺失的事件键。运行时在 process-wake.ts 中通过异步回调按需构建该水合载荷并注入到处理器上下文见 context-factory.ts。关于这些唤醒事件如何被生产出来的完整叙事参见 Waking entities。Wake配置何时唤醒父实体Wake类型用于配置父实体应在何时被子实体、被观察实体或共享状态变化唤醒。它被用在ctx.spawn()、ctx.observe()和ctx.observe(db(...))的 options 中type Wake | runFinished | { on: runFinished; includeResponse?: boolean } | { on: change collections?: string[] ops?: (insert | update | delete)[] debounceMs?: number timeoutMs?: number }该定义与源码 types.ts 中的Wake联合类型完全一致。运行时在 process-wake.ts 的normalizeForkWake函数中把用户友好的Wake翻译成服务端forkEntity兼容的唤醒条件runFinished短字符串形式与对象形式统一归一化为condition: runFinished对象形式的includeResponse、debounceMs、timeoutMs会被逐一透传。runFinished当子实体的 agent run 完成时状态变为completed或failed唤醒父实体。默认情况下唤醒事件会把子实体拼接后的文本响应包含在finished_child.response中。{ on: runFinished, includeResponse?: boolean }runFinished的对象形式带选项。设置includeResponse: false可以从唤醒事件中省略子实体的文本响应——当子实体响应很大、父实体不需要完整文本时这可以显著减小唤醒载荷。{ on: change }当被观察流中发生变化时唤醒父实体FieldTypeDescriptiononchangeRequired discriminant.collectionsstring[]Optional filter. Only wake on changes to these collections.opsstring[]Optional operation filter:insert,update, and/ordelete.debounceMsnumberDebounce interval in milliseconds. Batches rapid changes.timeoutMsnumberMaximum time to wait before waking, even if no changes occur.两个可选项的行为值得展开debounceMs设置后短时间内连续到达的变化会被批量合并唤醒在最后一次变化后的debounceMs毫秒才触发。适合高频写入但处理器处理成本较高的场景。timeoutMs设置后即使没有任何变化发生也会在超时后强制触发一次唤醒。这是实现心跳式heartbeat处理器的关键——处理器可以周期性检查状态而不必依赖外部事件驱动。两者都未设置时每次变化都会产生一次唤醒。在处理器中读取 WakeEvent最小可用的模式是按wake.type分支示例来自 waking-entities.mdasync handler(ctx, wake) { if (wake.type inbox) { // external input - reply, dispatch, etc. ctx.useAgent({ ... }) await ctx.agent.run() return } // everything else (child finished, change, cron, webhook source, timeout) arrives as type wake. // Inspect wake.payload for the specific sub-kind. ctx.sleep() }有两个唤醒类型会直接到达处理器inbox— 一条外部消息被投递到该实体的收件箱。wake— 为其他一切场景合成的唤醒子实体完成、集合变化、cron、webhook source、timeout具体细节在wake.payload上。future_send计划投递的是一条消息因此以inbox到达。wake.payload的完整形态changes[]、finished_child、other_children、timeout参见上文 wake-type catalog。在仓库测试 process-wake.test.ts 与 runtime-dsl.test.ts 中可以找到大量按wake.type inbox/wake.type wake分支断言唤醒语义的用例是理解该模式行为边界的直接参考。唤醒的生命周期与持久性理解WakeEvent还需要了解它背后的生命周期模型详见 Waking entitiesexternal event ─► wake entry (persisted) ─► handler invocation ─► WakeEvent passed to handler外部事件消息到达、子实体状态迁移、被观察集合变化、cron 触发或订阅的 webhook source 收到匹配数据。唤醒条目被持久化到实体的流上。这是耐久性保证——唤醒在进程重启、网络抖动和崩溃后依然存活已写入的唤醒最终一定会被投递给处理器。处理器被调用运行时拾取唤醒加载实体状态并以描述本次触发原因的WakeEvent调用你的处理器。处理器运行读取ctx.events、检查wake、配置 agent、产生新事件。处理器返回或调用ctx.sleep()后实体回到空闲直到下一次唤醒。正是由于这种唤醒即调用的模型处理器是**可重入re-entrant的每次唤醒都用同一处理器函数重新执行。跨唤醒携带状态应使用ctx.db.actions/ctx.db.collections一次性写入则应通过检查已有状态保证幂等。多个在实体繁忙期间或两次 ack 之间到达的外部事件会被合并为一次唤醒运行时保证唤醒覆盖源流上一段连续的 offset 区间即wake.fromOffset..wake.toOffsetwake.eventCount告诉你本次唤醒代表多少新事件。由于是至少一次投递at-least-once**语义处理器必须能够安全地以相同输入重复执行——优先采用幂等写入集合而不是在每次唤醒上执行副作用。深入阅读WakeEvent 类型参考 — 本文对应的权威类型文档。Waking entities — 唤醒的完整叙事七种触发来源、合并与幂等、debounce 与 timeout 的实战取舍。Spawning coordinating — 在spawn与observe中使用wake。Shared state — 在observe(db(...))中使用wake监听共享状态集合。Webhook sources — 将实体订阅到外部 webhook 数据源。Signals — 可中断或通知活跃实体的生命周期控制。Writing handlers —HandlerContext与firstWake模式。【免费下载链接】electricThe agent platform built on sync.项目地址: https://gitcode.com/GitHub_Trending/el/electric创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考