事件与同步

上一篇讲的是请求与响应；但对话中逐字吐出的文本、工具进度、权限询问，都是服务端主动推送的。OpenCode 用一套事件系统承载这些推送，同一套系统还负责持久化会话状态。

事件的定义

事件定义在 packages/schema/src/event.ts：Event.define 接收类型名、可选的 durable 声明与数据 schema。一条运行时事件（Payload）包含 id、type、data，以及可选的 location（所属目录与工作区）、metadata 和 durable。关键在 durable：声明了它的事件会写进 SQLite，并按聚合字段（会话事件都是 sessionID）分配单调递增的 seq，事件上随之带出 aggregateID、seq、version；没声明的只在内存里广播一次。V1 会话事件的划分（packages/schema/src/v1/session.ts）：

V2 的 session.next.* 遵循同一原则：

Stream fragments are live-only; Text.Ended is the replayable full-value boundary.

（packages/schema/src/session-event.ts:209）

text.delta、reasoning.delta、tool.input.delta、compaction.delta 这些碎片不落盘，携带完整值的 *.ended 事件才是可回放的边界。

发布：先落盘、投影，再广播

发布持久事件时，packages/core/src/event.ts 的 EventV2.publish 在一个 immediate 模式的 SQLite 事务里完成三件事：读 EventSequenceTable 取该聚合的最新序号，新事件取下一个；依次运行通过 project() 注册的投影器；写入 EventSequenceTable 与 EventTable。事务提交后才通知内存里的监听者与 PubSub。

投影器集中在 packages/core/src/session/projector.ts：SessionV1.Event.Created 写会话表、MessageUpdated 写消息表、PartUpdated 写 part 表，V2 事件也各有投影。换句话说，数据库里的会话状态是持久事件的投影，改状态的正途是发事件。设计初衷见 packages/opencode/src/sync/README.md:31：

This system defines a basic event sourcing system for session replayability. The goal is to allow for one device to control and modify the session, and allow multiple other devices to "sync" session data.

单写者前提让排序变得简单，一个递增整数就够，不需要分布式时钟。不过 README 描述的还是早期的 SyncEvent API，这套机制如今已并入 EventV2，packages/opencode/src/server/projectors.ts 里只剩一个空的 initProjectors()。

分发：EventV2Bridge 与 GlobalBus

EventV2 是 Effect 服务，而 SSE 路由、Worker 隧道需要普通的回调接口，两者之间隔着 packages/opencode/src/event-v2-bridge.ts。它发布时自动补上当前实例的 location，同时监听所有事件，转发给进程级的 GlobalBus（一个 Node EventEmitter）：

        GlobalBus.emit("event", {
          directory: event.location?.directory ?? ctx?.directory,
          project: ctx?.project.id,
          workspace: workspaceID,
          payload: { id: event.id, type: event.type, properties: event.data },
        })
        if (event.durable === undefined) return
        GlobalBus.emit("event", {
          // ...
          payload: {
            type: "sync",
            syncEvent: {
              id: event.id,
              type: EventV2.versionedType(event.type, event.durable.version),

（packages/opencode/src/event-v2-bridge.ts:39，删减后）

两个细节：对外的旧形状用 properties 而非 data；持久事件会再发一次，包成 type: "sync"，类型带版本号（如 session.created.1），并附上 seq 与 aggregateID，这是给同步用的原始记录。

订阅：几条 SSE 路由

三条实时路由的第一条消息都是 server.connected，前两条每 10 秒发一次 server.heartbeat，/api/event 用 SSE 注释行每 15 秒保活；/event 收到本实例的 server.instance.disposed 后主动结束。监听的注册时机也有讲究：

Listener registration is eager, so events published after this point cannot be lost while the HTTP body fiber is starting or emitting server.connected.

（packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts:29）

先挂监听、事件进无界队列，再写响应体，中间不留空窗。/api/event 则用有界的丢弃队列，满了以 SubscriberOverflowError 断开慢客户端。

持久事件与断线续传

EventV2.durable({ aggregateID, after }) 是 V2 的续传原语：先订阅一个容量为 1 的 sliding PubSub 作为“脏标记”，再读历史，之后每次被唤醒就重查 seq 大于游标的行：

            const historical = yield* read
            const live = Stream.fromSubscription(wakes).pipe(
              Stream.mapEffect(() => read),
              Stream.flattenIterable,
            )
            return Stream.concat(Stream.fromIterable(historical), live)

（packages/core/src/event.ts:597）

唤醒只是提示，事件靠数据库保存，多次提交合并成一次唤醒也不会丢。specs/v2/session.md:175 对边界的表述是：

Live-only text, reasoning, and tool-input fragments remain available through EventV2 subscriptions for connected renderers; they are intentionally absent from the replayable Session stream.

“可回放”与“实时渲染”被刻意拆成两个通道：一个游标严格对应一个持久序号，碎片只给在线的渲染器（见V2 会话内核）。V1 的两条路由不带游标，断线期间的事件不会补发，客户端只能靠 REST 重建基线。

TUI 怎样消费事件

入口在 packages/tui/src/context/sdk.tsx：宿主传入了 events（默认模式下就是 Worker 的 RPC 通道）就直接订阅，否则自己调 sdk.global.event() 连 SSE，断线后从 1 秒起翻倍退避、封顶 30 秒。事件按帧攒批：

    const handleEvent = (event: GlobalEvent) => {
      queue.push(event)
      const elapsed = Date.now() - last

      if (timer) return
      // If we just flushed recently (within 16ms), batch this with future events
      // Otherwise, process immediately to avoid latency
      if (elapsed < 16) {
        timer = setTimeout(flush, 16)
        return
      }
      flush()
    }

（packages/tui/src/context/sdk.tsx:68）

flush 用 Solid 的 batch 包住一批事件，所有 store 更新只触发一次渲染。

同目录的 event.ts 滤掉 sync 事件，sync.tsx 把其余事件折叠进 Solid store：会话、消息、part 按 ID 有序存放，用二分查找定位插入点；message.part.delta 直接把 delta 拼到 part 的对应字段；--auto 模式下 permission.asked 自动回复 once；收到 server.instance.disposed 就重新 bootstrap；每个会话只保留最新 100 条消息。进入会话时 session.sync() 用 REST 拉一次基线，并记下拉取期间被事件改过的消息与 part，合并时以事件为准，避免旧快照盖掉新数据。V2 的 data.tsx 订阅同一条流，把 properties 改名为 data、补上 location 后按 session.next.* 处理。

工作区同步：事件溯源的回报

持久事件最大的收益是跨机器同步（实验特性 experimentalWorkspaces）。连接远程工作区时，packages/opencode/src/control-plane/workspace.ts 先 POST /sync/history，把本地已知的“聚合 ID 到最新序号”映射发过去，取回缺失的事件；再订阅远端的 /global/event，遇到 sync 事件就用 events.replay 在本地重放。重放会校验序号连续、内容一致，并记录聚合的属主，避免不同来源写乱同一个会话。

下一篇：SDK、协议与客户端——这些路由与事件如何被写成契约，并生成 Promise 与 Effect 两种客户端。