OpenRig事件总线与SSE推送实战RigEvent如何零丢失送达前端完整指南【免费下载链接】openrigBuild your own network of agents from Claude Code, Codex and Pi: persistent teams with roles, shared context and owned work.项目地址: https://gitcode.com/GitHub_Trending/op/openrigOpenRig 是一个用 Claude Code、Codex 等运行时搭建持久化 Agent 团队的开源框架。它的核心难题之一是让前端 UI 实时感知每个 Agent 的状态变化——答案是一套事件总线 SSEServer-Sent Events推送机制所有状态变化先被固化为RigEvent事件写入 SQLite 事件日志再由 Hono 路由通过 SSE 通道先回放、后直播地送达浏览器前端用原生EventSource订阅并自动刷新界面。本文将用最少代码、最直观的方式讲清这条数据链路 图中 TUI 拓扑视图里每个节点的百分比与状态灯之所以能实时跳动靠的正是这条事件推送链路。一、事件总线 EventBus所有变化的唯一出口OpenRig 的事件总线实现位于 event-bus.ts。它的设计哲学只有一句话先持久化后通知。emit()任何业务动作创建 rig、启动节点、恢复会话……都会调用它。事件先插入 SQLite 的events表拿到自增序号seq后才分发给内存中的订阅者。persistWithinTransaction()允许调用方把事件落库和业务写库放进同一个事务保证原子性——不会出现节点建好了但事件丢了。subscribe()/replaySince()前者订阅实时事件流后者按序号回放历史事件二者是 SSE 通道的核心原料。它还内置了一个巧妙的容错设计落库的事件若因损坏无法解析所谓poison row总线不会崩溃而是发出一个event.delivery_poisoned元事件记录问题其余事件继续流动 ️二、RigEvent 联合类型事件世界的类型字典打开 types.ts你会看到 OpenRig 最完整的事件说明书——RigEvent是一个 TypeScript 联合类型discriminated union每个分支以type字段为判别符。大致可分为几族事件族代表事件含义️ 拓扑生命周期rig.created、node.added、node.removedrig/节点增删 会话与启动node.launched、seat.fresh_launched、session.status_changedAgent 启动与状态迁移 恢复与交接restore.started、restore.completed、seat.handover_completed快照恢复、占座交接 协同原语stream.emitted、queue.created、inbox.absorbed消息流、任务队列、收件箱⏰ 守护与流程watchdog.evaluation_fired、workflow.step_closed看门狗与工作流推进新增事件只需在联合类型上追加一个分支编译器就会自动要求所有消费点处理新类型——这是 OpenRig 能放心扩张到 80 种事件的基础。事件落库后会被包装为PersistedEventtypes.ts#L316即RigEvent加上seq和createdAt两个字段seq正是后续断线续传的钥匙 三、SSE 通道 /api/events先回放、再直播的零缺口推送前端的总入口是 events.ts 中的GET /api/events。它的处理流程堪称教科书读Last-Event-ID头SSE 客户端断线重连时会自动带上最后收到的 seq服务器据此确定回放起点缺失则从 0 开始。先订阅、后回放在查询历史事件之前先挂上内存订阅回放期间新到的事件先存入缓冲区——这一步彻底消灭了回放与直播之间漏掉事件的缝隙。回放历史调用replaySince()全局流或replayAll()按seq升序逐条writeSSE每条都带上id即 seq。去重排空缓冲区以maxReplayedSeq水位线跳过已发送的事件保证顺序单调且无重复。切入直播模式新事件直接写流直到客户端断开自动清理订阅。支持?rigIdxxx查询参数可只订阅单个 rig 的事件省略则为全局事件流。此外协同消息流还有专用 SSE 通道 stream.ts#L143-L195GET /api/stream/sse只转发stream.emitted事件——先补发最近 50 条历史消息再进入实时模式逻辑与总通道同构。四、前端订阅一个单例 Hub 自动失效刷新浏览器端的核心是 topology-events.ts 里的TopologyEventHub用原生EventSource连接/api/events全应用只有一条 SSE 连接单例模式维护最多 100 条的回放缓存晚挂载的组件也能补看到最近事件不会漏更新暴露连接状态connected/reconnectingUI 可以据此显示实时连接中的指示灯。拿到事件后真正干活的是 useGlobalEvents.ts 这个挂在 AppShell 上的全局 Hook。它把事件翻译成 TanStack Query 的缓存失效指令例如node.startup_*→ 刷新该 rig 的节点列表rig.created/rig.archived/restore.completed→ 刷新 rig 摘要与ps视图proof.*→ 刷新评审相关查询。所有失效请求会先进入一个Set去重再延迟 150ms 批量执行合并抖动避免事件风暴打爆网络请求。这就是为什么你在 guide-step-01 演示界面 左侧 Explorer 里看到的节点状态始终活的——不是轮询而是事件驱动 五、事件如何变成界面更新三层保障总结整套机制可以浓缩为三层保障这也是它可靠的原因保障机制出处不丢事件先写 SQLite 再通知断线靠Last-Event-ID回放补齐event-bus.ts不乱seq单调递增 缓冲区按水位线去重events.ts#L27-L67不断EventSource原生自动重连 前端 100 条回放缓存topology-events.ts#L24-L97事件还能二次加工view-event-bridge.ts 订阅队列/收件箱等状态变更事件映射为view.changed元事件再唤醒/api/views/:name/sse的订阅者——事件驱动事件层层放大却始终可追溯。六、动手验证亲眼看到 RigEvent 送达想亲自感受这条链路三步即可运行git clone https://gitcode.com/GitHub_Trending/op/openrig并按 README.md 启动 daemon 与 UI打开浏览器开发者工具的Network → 筛选eventsource找到/api/events连接在 guide-step-02 所示的拓扑页 里操作任意节点启动/停止SSE 面板会立刻滚动出对应的RigEventJSON同时界面无需刷新自动更新 ✅小结OpenRig 用**SQLite 事件日志 内存总线 SSE 回放直播三板斧把 80 种RigEvent联合类型事件可靠送达前端前端则用单例 EventSource Hub 防抖缓存失效**把事件转化为无感知的界面更新。这套模式不依赖任何重型消息中间件却同时做到了不丢、不乱、不断——正是它值得每个需要实时后端的项目借鉴的地方。【免费下载链接】openrigBuild your own network of agents from Claude Code, Codex and Pi: persistent teams with roles, shared context and owned work.项目地址: https://gitcode.com/GitHub_Trending/op/openrig创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考