# 事件与同步

> EventV2 的定义、持久化与投影，EventV2Bridge 与 GlobalBus 的进程级广播，几条 SSE 路由的区别，以及 TUI 如何把事件流折叠成本地状态。

- 作者：David（道雾轩）
- 专栏：OpenCode 源码解读（https://daiw.org/manual/opencode.md）
- 最后更新：2026-09-15
- 原文：https://daiw.org/manual/opencode/events-sync
- 转载与引用：请注明出处并附原文链接（https://daiw.org/about/copyright）

[上一篇](https://daiw.org/manual/opencode/server)讲的是请求与响应；但对话中逐字吐出的文本、工具进度、权限询问，都是服务端主动推送的。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`）：

| 持久（版本 1） | 只推实时 |
| --- | --- |
| `session.created`、`session.updated`、`session.deleted` | `message.part.delta` |
| `message.updated`、`message.removed` | `session.diff`、`session.error` |
| `message.part.updated`、`message.part.removed` | `session.status`、`session.idle`（已废弃） |

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`）：

```ts
        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 路由

| 路由 | 数据源 | 形状 | 典型消费者 |
| --- | --- | --- | --- |
| `/global/event` | `GlobalBus` 全量，含 `sync` | `directory`、`workspace`、`payload` 等 | TUI、ACP、工作区同步 |
| `/event` | `EventV2Bridge`，按实例目录与工作区过滤 | `id`、`type`、`properties` | `opencode run`、SDK 脚本 |
| `/api/event` | `EventV2.allBounded`，容量 256 | V2 原生形状 | 新客户端 |
| `/api/session/:sessionID/event` | `EventV2.durable` | 只含持久会话事件 | 需要续传的消费者 |

三条实时路由的第一条消息都是 `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` 大于游标的行：

```ts
            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 会话内核](https://daiw.org/manual/opencode/session-v2)）。V1 的两条路由不带游标，断线期间的事件不会补发，客户端只能靠 REST 重建基线。

## TUI 怎样消费事件

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

```ts
    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、协议与客户端](https://daiw.org/manual/opencode/sdk-and-protocol)——这些路由与事件如何被写成契约，并生成 Promise 与 Effect 两种客户端。