05-Event Sourcing：事件系统

05-Event Sourcing 事件系统

一、事件存储的三张核心表

// packages/core/src/event/sql.tsexport const EventSequenceTable = sqliteTable("event_sequence", {aggregate_id: text().notNull().primaryKey(), // ← 聚合根 ID（如 sessionID）seq: integer().notNull(), // ← 当前已提交的最大序列号owner_id: text(), // ← 聚合根的持有者（用于多节点互斥）})export const EventTable = sqliteTable("event", {id: text().$type<EventV2.ID>().primaryKey(),// ← 事件 UUIDaggregate_id: text().notNull().references(() => EventSequenceTable.aggregate_id, { onDelete: "cascade" }),seq: integer().notNull(),// ← 聚合内单调递增序列号type: text().notNull(),// ← 版本化类型（如 "session.next.step.ended.2"）data: text({ mode: "json" }).notNull(),// ← JSON 序列化的事件 payload}, (table) => [uniqueIndex("event_aggregate_seq_idx").on(table.aggregate_id, table.seq),index("event_aggregate_type_seq_idx").on(table.aggregate_id, table.type, table.seq),])

两张表构成经典的 Event Sourcing 存储模型：

• EventSequenceTable：每个聚合根一行，记录当前 seq 高水位和 owner_id。是"序列号分配器"。
• EventTable：追加写入的事件日志，(aggregate_id, seq) 唯一索引保证同一聚合内不会有重复序列号。

二、Q1：EventV2.publish(definition, data) 的事务内做了哪些事？

2.1 调用链总览

publish(definition, data, options?)└─ publishEvent(definition, event, options?.commit) └─ commitDurableEvent(definition, event, undefined, commit) // durable 事件└─ db.transaction { ... }// 单事务原子操作 └─ notify(event, isolateListeners=true)// 事务后通知

2.2 事务内的六步操作

commitDurableEvent 的核心是 db.transaction 块（event.ts:237-363），所有操作在同一个 SQLite 事务中执行：

// packages/core/src/event.ts:237-363 (简化）yield* Effect.uninterruptible(Effect.gen(function* () {const committed = yield* db.transaction(() =>Effect.gen(function* () {// ── 步骤 1：读取当前聚合的序列号高水位 ──const row = yield* db.select({ seq, ownerID }).from(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).get().pipe(Effect.orDie)const latest = row?.seq ?? -1// ── 步骤 2：幂等性检查（仅 replay 路径）──if (input && input.seq <= latest) {// 查已存储的同 seq 事件，比较 id + type + data 是否深相等const stored = yield* db.select().from(EventTable).where(and(eq(aggregate_id, aggregateID), eq(seq, input.seq))).get().pipe(Effect.orDie)if (stored?.id === event.id &&stored.type === versionedType(definition.type, durable.version) &&isDeepStrictEqual(stored.data, encoded)) {return// ← 幂等成功：事件已存在且内容一致，直接返回}yield* Effect.die(new InvalidDurableEventError({ ... "Replay diverged" }))}// ── 步骤 3：序列号分配 ──const seq = input?.seq ?? latest + 1// publish 分配 latest+1；replay 用指定 seq// ── 步骤 4：Event ID 唯一性检查 ──const stored = yield* db.select({ aggregateID, seq }).from(EventTable).where(eq(EventTable.id, event.id)).get().pipe(Effect.orDie)if (stored) yield* Effect.die(new InvalidDurableEventError({ ... "already exists" }))// ── 步骤 5：执行 Projector（在事务内！）──const committed = { ...event, durable: { aggregateID, seq, version: durable.version } }for (const projector of list) {yield* projector(committed)// ← 投影到 session/session_message/part 等读模型表}// ── 步骤 5b：执行用户提供的 commit 回调（在事务内！）──if (commit) yield* commit(seq) // ← 额外的本地操作投影// ── 步骤 6：持久化事件 + 更新序列号 ──yield* db.insert(EventSequenceTable).values([{ aggregate_id: aggregateID, seq, owner_id: input?.ownerID }]).onConflictDoUpdate({ target: aggregate_id, set: { seq, ... } }).run().pipe(Effect.orDie)yield* db.insert(EventTable).values([{ id: event.id, aggregate_id: aggregateID, seq, type: versionedType(definition.type, durable.version), data: encoded }]).run().pipe(Effect.orDie)return { aggregateID, seq }}), { behavior: "immediate" }).pipe(Effect.orDie)// ── 事务后：唤醒 durable stream 订阅者 ──if (committed) {yield* Effect.forEach(pubsub.durable.get(committed.aggregateID) ?? [],(wake) => PubSub.publish(wake, undefined), { discard: true })}return committed}),)

2.3 关键设计：Projector 在事务内执行

这是最关键的设计决策——Projector 与事件写入在同一个 SQLite 事务中。这意味着：

1. 原子性：如果 Projector 失败（如写入 session_message 表违反约束），整个事务回滚——事件不会写入 EventTable，序列号不会递增。不存在"事件已存储但投影未完成"的半完成状态。
2. 一致性：读模型（session/session_message/part）与事件日志在任何时刻都保持一致。查询读模型不需要"等待投影追上"。
3. 崩溃恢复：如果事务提交前崩溃，所有变更都丢失——重启后重新 publish 即可。如果事务提交后崩溃，事件和投影都已持久化——重放不需要重新执行投影。

对比其他 Event Sourcing 系统（如 Kafka + CQRS）：通常 Projector 是异步的，读模型有最终一致性延迟。opencode 选择同步投影是因为它运行在单进程 SQLite 上，事务开销极低，且 Agent 会话对一致性要求极高（不能出现"工具执行了但消息没显示"的状态）。

2.4 事务后的通知

事务提交后，notify(event, isolateListeners=true) 执行：

// packages/core/src/event.ts:406-417function notify(event: Payload, isolateListeners: boolean) {return Effect.gen(function* () {yield* Effect.forEach(listeners, (listener) =>isolateListeners ? observe(event, listener) : listener(event), // ← 隔离失败{ discard: true })const typed = pubsub.typed.get(event.type)if (typed) yield* PubSub.publish(typed, event) // ← 类型订阅yield* PubSub.publish(pubsub.all, event)// ← 全局订阅})}

isolateListeners=true 时用 observe 包裹——单个 listener 失败不会影响其他 listener，只记录日志。这在事务提交后很重要：通知失败不应回滚已提交的事务。

三、Q2：durable 事件的 aggregateID + seq 如何保证顺序和幂等？

3.1 聚合根与序列号的语义

每个 durable 事件的 definition 指定 durable.aggregate 字段名（对 Session 事件统一为 "sessionID"）：

// packages/schema/src/session-event.ts:38-43const options = {durable: {aggregate: "sessionID",// ← 从 data.sessionID 取值作为 aggregateIDversion: 1,},} as const

commitDurableEvent 从事件数据中提取这个字段：

// packages/core/src/event.ts:218-219const aggregateID = (event.data as Record<string, unknown>)[durable.aggregate]

因此 aggregateID = sessionID，每个 Session 是一个独立的聚合根。seq 在每个聚合内从 0 开始单调递增。

3.2 顺序保证：latest + 1 分配

publish 路径（非 replay）的序列号分配：

// packages/core/src/event.ts:294const seq = input?.seq ?? latest + 1// publish 时 input 为 undefined，分配 latest+1

latest 从 EventSequenceTable 读取，如果聚合不存在则 -1，所以第一个事件的 seq = 0。

并发安全：由于序列号分配在 db.transaction 内，SQLite 的 immediate 事务行为会在事务开始时获取写锁——两个并发的 publish 对同一聚合会串行化，第二个事务读到第一个事务提交后的 latest，不会冲突。

3.3 幂等保证：三重检查

replay 路径（input.seq 指定）的幂等检查分三层：

第一层：序列号已存在时的内容深比较

// packages/core/src/event.ts:262-289if (input && input.seq <= latest) {const stored = yield* db.select().from(EventTable).where(and(eq(aggregate_id, aggregateID), eq(seq, input.seq))).get()if (stored?.id === event.id &&// ① ID 相同stored.type === versionedType(definition.type, ...) &&// ② 类型+版本相同isDeepStrictEqual(stored.data, encoded)) {// ③ 数据深相等// 幂等成功——可以补充 owner_id 后直接返回if (input.ownerID && row?.ownerID == null) {yield* db.update(EventSequenceTable).set({ owner_id: input.ownerID })...}return// ← 不重复写入，不重新投影}yield* Effect.die(new InvalidDurableEventError({ ... "Replay diverged" }))}

如果 seq 已存在但内容不一致——抛出 diverged 错误。这是 replay 的核心分歧检测机制（详见 Q3）。

第二层：Event ID 全局唯一检查

// packages/core/src/event.ts:303-315const stored = yield* db.select({ aggregateID, seq }).from(EventTable).where(eq(EventTable.id, event.id)).get()if (stored)yield* Effect.die(new InvalidDurableEventError({... `Event ${event.id} already exists at aggregate ${stored.aggregateID} sequence ${stored.seq}`}))

即使 seq 是新的，如果 event.id 已存在（可能在不同聚合中），也拒绝写入。这防止了事件被重复提交到不同聚合。

第三层：序列号连续性检查

// packages/core/src/event.ts:295-301if (input && seq !== latest + 1) {yield* Effect.die(new InvalidDurableEventError({... `Sequence mismatch for aggregate ${aggregateID}: expected ${latest + 1}, got ${seq}`}))}

replay 时如果 seq 不是 latest + 1（跳号），直接报错。这保证了事件序列没有空洞。

3.4 owner_id 与多节点互斥

EventSequenceTable.owner_id 用于多节点场景的聚合根互斥：

// packages/core/src/event.ts:254-261 (strictOwner 检查)if (input?.strictOwner && row?.ownerID && row.ownerID !== input.ownerID) {yield* Effect.die(new InvalidDurableEventError({... `Replay owner mismatch for aggregate ${aggregateID}: expected ${row.ownerID}, got ${input.ownerID ?? "none"}`}))}// packages/core/src/event.ts:291-293 (非 strict 时静默跳过)if (input && row?.ownerID && row.ownerID !== input.ownerID) {return// ← 不是自己的聚合，静默跳过}

这支持未来的 clustered ownership（runner/llm.ts:51 的 TODO 项）：多个节点可以 replay 同一聚合的事件，但只有 owner 可以 publish 新事件。

四、Q3：replay 和 replayAll 如何检测事件分歧（divergence）？

4.1 replay 的分歧检测

replay 接收一个 SerializedEvent（从外部存储或日志中恢复的序列化事件），通过 commitDurableEvent 的 input.seq 路径提交：

// packages/core/src/event.ts:441-478function replay(event: SerializedEvent, options?) {return Effect.gen(function* () {const definition = Durable.get(event.type)if (!definition?.durable) {yield* Effect.die(new InvalidDurableEventError({ ... "Unknown durable event type" }))} else {const payload = {id: event.id, type: definition.type,data: Schema.decodeUnknownSync(definition.data)(event.data),} as Payloadconst committed = yield* commitDurableEvent(definition, payload, {seq: event.seq,// ← 指定 seqaggregateID: event.aggregateID,// ← 指定 aggregateIDownerID: options?.ownerID,strictOwner: options?.strictOwner,})if (committed && options?.publish) {yield* notify({ ...payload, durable: { ... } }, true) // ← 可选通知}}})}

分歧检测发生在 commitDurableEvent 的第一层检查（§3.3）：如果 input.seq <= latest，查已存储的同 seq 事件，比较 id + type + data：

• 全部一致 → 幂等返回（事件已存在，不重复处理）
• 任一不一致 → Effect.die(InvalidDurableEventError("Replay diverged"))

isDeepStrictEqual 是 Node.js 的深严格相等——比较两个值的结构、类型、值，不接受 -0 === 0 等隐式转换。这确保了分歧检测的精确性。

4.2 replayAll 的前置校验

replayAll 在逐个 replay 之前做两个前置校验：

// packages/core/src/event.ts:480-512function replayAll(events: SerializedEvent[], options?) {return Effect.gen(function* () {const source = events[0]?.aggregateIDif (!source) return undefined// ── 校验 1：所有事件必须属于同一聚合 ──if (events.some((event) => event.aggregateID !== source)) {yield* Effect.die(new InvalidDurableEventError({ ... "must belong to the same aggregate" }))}// ── 校验 2：序列号必须连续递增 ──const start = events[0]?.seq ?? 0for (const [index, event] of events.entries()) {const seq = start + indexif (event.seq !== seq) {yield* Effect.die(new InvalidDurableEventError({... `Sequence mismatch at index ${index}: expected ${seq}, got ${event.seq}`}))}}// ── 逐个 replay ──for (const event of events) {yield* replay(event, options)}return source})}

这两个校验确保 replay 的事件批次是一个完整的、连续的聚合历史片段。如果批次内部就不一致（跨聚合或跳号），无需等到逐个 replay 才发现错误。

4.3 分歧场景示例

假设聚合 ses_abc 已有序列 0, 1, 2，现在 replay 一批事件 seq=1, 2, 3：

1. replay(seq=1)：1 <= 2(latest)，查已存储的 seq=1 事件 → 比较 id+type+data → 一致 → 幂等返回
2. replay(seq=2)：2 <= 2(latest)，查已存储的 seq=2 事件 → 比较 → 一致 → 幂等返回
3. replay(seq=3)：3 > 2(latest)，新事件 → 检查 seq === latest+1(3) → 通过 → 写入

如果步骤 1 中存储的事件 data 与 replay 的 data 不一致 → Effect.die("Replay diverged")，整个 replayAll 中止。

五、Q4：SessionProjector 注册了多少种事件投影器，各自把哪种事件映射到哪张 SQLite 表？

5.1 投影器总览

projector.ts 的 layer 注册了 28 个投影器，覆盖 Session 生命周期的全部事件。投影目标为 6 张表：

5.2 投影器分类详解

类型 A：直接写 SessionTable（8 个）

类型 B：写 V1 旧表 MessageTable / PartTable（3 个）

类型 C：写 SessionInputTable（2 个）

类型 D：通过 run(db, event) 写 SessionMessageTable（15 个）

这 15 个事件都委托给 run(db, event) → SessionMessageUpdater.update(adapter, event)，后者使用 Immer produce 不可变更新助手消息的部件：

注意：Text.Delta、Reasoning.Delta、Tool.Input.Delta、Compaction.Delta 这些实时流事件没有 durable 配置，不会写入事件日志，只通过 notify 实时推送。它们在 message-updater 中有处理逻辑（如 text.delta 追加文本），但仅在内存 adapter 中生效——持久化的是 text.ended 的最终值。

5.3 run(db, event) 的 Adapter 模式

// packages/core/src/session/projector.ts:112-191function run(db: DatabaseService, event: SessionEvent.Event) {return Effect.gen(function* () {const adapter: SessionMessageUpdater.Adapter = {getCurrentAssistant() { /* 查最新的未完成 assistant 行 */ },getAssistant(messageID) { /* 查指定 assistant 行 */ },getCurrentShell(callID) { /* 查最新的匹配 callID 的 shell 行 */ },updateAssistant: updateMessage, // ← UPDATE SessionMessageTableupdateShell: updateMessage, // ← UPDATE SessionMessageTableappendMessage,// ← INSERT SessionMessageTable}yield* SessionMessageUpdater.update(adapter, event)})}

message-updater.ts 的 update 函数用 SessionEvent.All.match(event, { ... }) 做模式匹配——一个巨大的 switch，每种事件类型对应一个 Immer recipe。Adapter 抽象让同一套更新逻辑可以工作在 SQLite（projector.ts）和内存（memory() adapter，用于测试）两种后端上。

5.4 usage 增量更新

PartUpdated 和 PartRemoved/MessageRemoved 涉及 cost/token 的增量更新：

// packages/core/src/session/projector.ts:312-329 (PartUpdated)yield* events.project(SessionV1.Event.PartUpdated, (event) =>Effect.gen(function* () {// ...const previous = row && usage(row.data) // ← 旧 part 的 usageconst next = usage(event.data.part)// ← 新 part 的 usageif (previous) yield* applyUsage(db, row.session_id, previous, -1)// ← 减去旧值if (next) yield* applyUsage(db, sessionID, next)// ← 加上新值}),)

applyUsage 用 SQL 增量表达式 `cost + ${value.cost * sign}` 更新 SessionTable。sign = -1 时是减法——用于 PartRemoved 或 PartUpdated 的"先减旧值再加新值"。这保证了即使 part 被更新或删除，Session 的累计 cost/tokens 仍然准确。

六、Q5：RevertEvent.Committed 如何通过 seq 边界截断消息实现"回滚"？

6.1 回滚的三阶段流程

回滚功能允许用户"撤销到某条消息"——把该消息之后的所有状态变更"抹掉"。这通过三个事件实现：

1. RevertEvent.Staged：暂存回滚意图（记录目标 messageID + files），写入 SessionTable.revert 字段
2. RevertEvent.Cleared：取消回滚意图，清除 SessionTable.revert 字段
3. RevertEvent.Committed：执行回滚——截断消息 + 清理输入 + 清除意图

6.2 RevertEvent.Committed 的投影逻辑

// packages/core/src/session/projector.ts:415-454yield* events.project(SessionEvent.RevertEvent.Committed, (event) =>Effect.gen(function* () {// ── 步骤 1：查找边界消息的 seq ──const boundary = yield* db.select({ seq: SessionMessageTable.seq }).from(SessionMessageTable).where(and(eq(SessionMessageTable.session_id, event.data.sessionID),eq(SessionMessageTable.id, event.data.messageID), // ← 目标回滚消息)).get().pipe(Effect.orDie)if (!boundary) return yield* Effect.die(`Revert boundary message not found: ${event.data.messageID}`)// ── 步骤 2：截断 SessionMessageTable（删除 seq > boundary 的所有消息）──yield* db.delete(SessionMessageTable).where(and(eq(SessionMessageTable.session_id, event.data.sessionID),gt(SessionMessageTable.seq, boundary.seq),// ← 只删 seq 大于边界的)).run().pipe(Effect.orDie)// ── 步骤 3：截断 SessionInputTable（删除 admitted_seq 或 promoted_seq > boundary 的输入）──yield* db.delete(SessionInputTable).where(and(eq(SessionInputTable.session_id, event.data.sessionID),or(gt(SessionInputTable.admitted_seq, boundary.seq),// ← 回滚后提交的输入gt(SessionInputTable.promoted_seq, boundary.seq),// ← 回滚后提升的输入),)).run().pipe(Effect.orDie)// ── 步骤 4：清除 revert 意图 ──yield* db.update(SessionTable).set({ revert: null, time_updated: DateTime.toEpochMillis(event.data.timestamp) }).where(eq(SessionTable.id, event.data.sessionID)).run().pipe(Effect.orDie)// ── 步骤 5：重置上下文 epoch ──yield* SessionContextEpoch.reset(db, event.data.sessionID)}),)

6.3 seq 边界截断的精妙之处

回滚的核心是 gt(SessionMessageTable.seq, boundary.seq)——按 seq 而非 messageID 截断。这利用了 Event Sourcing 的序号天然时间线特性：

• 每条 SessionMessageTable 行的 seq 来自事件的 event.durable.seq（insertMessage 函数 projector.ts:203）。
• seq 是聚合内单调递增的，所以 seq > boundary.seq 等价于"在边界消息之后创建的所有消息"。
• 一次 DELETE WHERE seq > boundary 就能截断所有 assistant 消息、tool 调用、text 部件、reasoning 部件等——因为它们都存储在同一张 SessionMessageTable 中，通过 seq 排序。

6.4 为什么同时清理 SessionInputTable？

回滚后，在边界之后提交的用户输入（steer/queue）也应该被清除——否则 runner 重启时会尝试 promote 这些输入，但它们对应的状态已经被截断了。

清理条件是 admitted_seq > boundary OR promoted_seq > boundary：

• admitted_seq > boundary：用户在回滚点之后提交了新输入
• promoted_seq > boundary：用户在回滚点之前提交但之后被提升的输入

这两个条件确保回滚后的输入队列与消息历史一致。

6.5 与事件日志的关系

关键：RevertEvent.Committed 不删除 EventTable 中的事件。它只截断读模型（SessionMessageTable/SessionInputTable），事件日志保持完整。

这意味着：

1. 审计性：所有历史事件仍然可查，回滚操作本身也是一个事件（RevertEvent.Committed），记录了回滚到哪个边界。
2. 可重放性：如果需要，可以通过 replayAll 从事件日志重建任意时间点的状态——包括回滚前的状态。
3. Projector 幂等：如果回滚后重新 replay 事件日志，Projector 不会"恢复"被截断的消息——因为 insertMessage 用 INSERT 而非 UPSERT，被删除的行不会被事件日志中的旧事件重新创建（除非显式 replay）。

这是 Event Sourcing 的经典权衡：读模型是"当前状态投影"，事件日志是"完整历史"。回滚修改的是投影，不是历史。

七、记忆系统与崩溃恢复机制

7.1 崩溃恢复的三个层次

基于上述 Event Sourcing 架构，opencode V2 的崩溃恢复分三层：

第一层：事务原子性

commitDurableEvent 在单个 SQLite 事务中完成"事件写入 + Projector 执行 + 序列号更新"。如果崩溃发生在事务提交前，所有变更丢失——状态一致。如果崩溃发生在事务提交后，所有变更已持久化——状态一致。不存在中间状态。

第二层：事件日志重放

如果读模型（SessionMessageTable 等）损坏或不一致，可以从 EventTable 重放。replayAll 逐个事件通过 commitDurableEvent 的 input.seq 路径提交——如果同 seq 事件已存在且内容一致（幂等），直接跳过；如果内容不一致（diverged），报错。这让重放是安全的——不会产生重复或冲突。

第三层：durable stream 续传

durable(aggregateID, after?) 返回一个 Stream，先读取 after 之后的历史事件，然后订阅实时 PubSub 通知：

// packages/core/src/event.ts:585-604const durable = (input: { aggregateID: string; after?: number }) =>Stream.unwrap(Effect.gen(function* () {const wakes = yield* subscribeDurable(input.aggregateID)let sequence = input.after ?? -1const read = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe(Effect.tap((events) => Effect.sync(() => {sequence = events.at(-1)?.durable?.seq ?? sequence // ← 记住已读位置})),)const historical = yield* read// ← 先读历史const live = Stream.fromSubscription(wakes).pipe( // ← 再接实时Stream.mapEffect(() => read), // ← 被 wake 唤醒后增量读Stream.flattenIterable,)return Stream.concat(Stream.fromIterable(historical), live)}))

subscribeDurable 注册一个 PubSub.sliding(1) 到 pubsub.durable.get(aggregateID)。每次该聚合有新事件提交时，commitDurableEvent 事务后会 PubSub.publish(wake, undefined) 唤醒订阅者，订阅者用 readAfter(aggregateID, sequence) 增量读取新事件。

sliding(1) 策略只保留最后一个 wake 信号——多次 publish 合并为一次读取，避免高频事件下的重复查询。

7.2 V1→V2 记忆系统的本质变化

V1 的状态管理是"内存优先 + SQLite 持久化"——SessionV1 在内存中维护消息列表，通过 SessionV1.Event 事件同步到 SQLite。崩溃后从 SQLite 恢复，但恢复逻辑分散在各处。

V2 的状态管理是"事件日志为唯一真相源"——所有状态变更先写事件日志（EventTable），Projector 在事务内同步投影到读模型。崩溃后只需重放事件日志即可重建全部状态。读模型是"事件日志的物化视图"，可以随时删除重建。

这种设计让 V2 具备了 V1 不具备的能力：

1. 精确回滚：RevertEvent.Committed 通过 seq 边界截断，而非 V1 的"删除消息列表中的元素"。
2. 分歧检测：replay 的深比较能发现事件日志与本地状态的不一致。
3. 多节点恢复：owner_id + strictOwner 支持未来的 clustered ownership。
4. durable stream：客户端可以断线重连后从 after 位置续传，不会丢失事件。

八、附录：关键行号索引