Event Sourcing 事件溯源系统详解

数码系统

相机 win10

测评 win11

手机智车

华为 Tesla

小米理想

苹果蔚来

游戏软件

LOL 抖音

原神微信

当前位置：首页

AI教程

Event Sourcing 事件溯源系统详解

热心网友时间：2026-07-15

转载

05-Event Sourcing 事件系统在 opencode V2 中，Event Sourcing 是整个系统的核心机制。它要同时保证事件的幂等性、顺序性、原子性，还要支持精确回滚和崩溃恢复。这套设计背后有一套精心打磨的存储模型和事务逻辑。下面我们就从最基础的几张表开始拆解。一、事件存储的

05-Event Sourcing 事件系统

在 opencode V2 中，Event Sourcing 是整个系统的核心机制。它要同时保证事件的幂等性、顺序性、原子性，还要支持精确回滚和崩溃恢复。这套设计背后有一套精心打磨的存储模型和事务逻辑。下面我们就从最基础的几张表开始拆解。

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

// packages/core/src/event/sql.ts
export 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(), // ← 事件 UUID
  aggregate_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 存储模型：

05-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 }
    }),
    { beha vior: "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 事务中。好处很明确：

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

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

2.4 事务后的通知

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

// packages/core/src/event.ts:406-417
function 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-43
const options = {
  durable: {
    aggregate: "sessionID",          // ← 从 data.sessionID 取值作为 aggregateID
    version: 1,
  },
} as const

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

// packages/core/src/event.ts:218-219
const 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:294
const 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-289
if (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-315
const 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-301
if (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-478
function 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 Payload
      const committed = yield* commitDurableEvent(definition, payload, {
        seq: event.seq,              // ← 指定 seq
        aggregateID: event.aggregateID,   // ← 指定 aggregateID
        ownerID: 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-512
function replayAll(events: SerializedEvent[], options?) {
  return Effect.gen(function* () {
    const source = events[0]?.aggregateID
    if (!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 ?? 0
    for (const [index, event] of events.entries()) {
      const seq = start + index
      if (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：

replay(seq=1)：1 <= 2(latest)，查已存储的 seq=1 事件 → 比较 id+type+data → 一致 → 幂等返回
replay(seq=2)：2 <= 2(latest)，查已存储的 seq=2 事件 → 比较 → 一致 → 幂等返回
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 张表：

目标表	用途
`SessionTable`	Session 元信息（标题、agent、model、cost、tokens、revert 状态）
`MessageTable`	V1 旧消息表（MessageUpdated/MessageRemoved）
`PartTable`	V1 旧消息部件表（PartUpdated/PartRemoved）
`SessionMessageTable`	V2 新消息表（Step/Text/Tool/Reasoning/Compaction 等）
`SessionInputTable`	用户输入队列（Prompted/PromptAdmitted）
`WorkspaceTable`	工作区时间戳（Session.Created 时更新）

5.2 投影器分类详解

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

事件	投影行为
`SessionV1.Event.Created`	`INSERT INTO SessionTable` + 更新 `WorkspaceTable.time_used`
`SessionV1.Event.Updated`	`UPDATE SessionTable SET ...`
`SessionEvent.Moved`	`UPDATE SessionTable SET directory/path/workspace_id` + `SessionContextEpoch.reset`
`SessionV1.Event.Deleted`	`DELETE FROM SessionTable`
`SessionEvent.AgentSwitched`	`UPDATE SessionTable SET agent` + `run(db, event)`
`SessionEvent.ModelSwitched`	`UPDATE SessionTable SET model` + `run(db, event)`
`SessionEvent.RevertEvent.Staged`	`UPDATE SessionTable SET revert = ...`
`SessionEvent.RevertEvent.Cleared`	`UPDATE SessionTable SET revert = null`
`SessionEvent.RevertEvent.Committed`	`DELETE FROM SessionMessageTable WHERE seq > boundary` + 清理输入 + 清除 revert

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

事件	投影行为
`SessionV1.Event.MessageUpdated`	`INSERT INTO MessageTable ON CONFLICT DO UPDATE`
`SessionV1.Event.MessageRemoved`	撤销 usage → `DELETE FROM MessageTable`
`SessionV1.Event.PartUpdated`	`INSERT INTO PartTable ON CONFLICT DO UPDATE` + usage 增量更新
`SessionV1.Event.PartRemoved`	撤销 usage → `DELETE FROM PartTable`

类型 C：写 `SessionInputTable`（2 个）

事件	投影行为
`SessionEvent.Prompted`	`SessionInput.projectPrompted`（设置 `promoted_seq`）+ `run(db, event)`
`SessionEvent.PromptAdmitted`	`SessionInput.projectAdmitted`（设置 `admitted_seq`）

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

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

事件	message-updater 行为
`AgentSwitched`	`appendMessage(AgentSwitched)`
`ModelSwitched`	`appendMessage(ModelSwitched)`
`ContextUpdated`	`appendMessage(System)`
`Synthetic`	`appendMessage(Synthetic)`
`Shell.Started`	`appendMessage(Shell)`
`Shell.Ended`	`updateShell`（设置 output + completed）
`Step.Started`	完成当前 assistant → `appendMessage(Assistant)`
`Step.Ended`	`updateAssistant`（设置 completed + finish + cost + tokens + snapshot）
`Step.Failed`	`updateAssistant`（设置 completed + error）
`Text.Started`	`updateAssistant`（push AssistantText part）
`Text.Ended`	`updateAssistant`（设置 text 最终值）
`Tool.Input.Started`	`updateAssistant`（push AssistantTool part, status=pending）
`Tool.Input.Ended`	`updateAssistant`（设置 tool.input）
`Tool.Called`	`updateAssistant`（status=running, 设置 input + provider）
`Tool.Progress`	`updateAssistant`（更新 structured + content）
`Tool.Success`	`updateAssistant`（status=completed, 设置 result + content + outputPaths）
`Tool.Failed`	`updateAssistant`（status=error, 设置 error）
`Reasoning.Started`	`updateAssistant`（push AssistantReasoning part）
`Reasoning.Ended`	`updateAssistant`（设置 text 最终值）
`Compaction.Ended`	`appendMessage(Compaction)`

注意：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-191
function 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 SessionMessageTable
      updateShell: updateMessage,     // ← UPDATE SessionMessageTable
      appendMessage,                   // ← 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 的 usage
    const next = usage(event.data.part)                // ← 新 part 的 usage
    if (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 回滚的三阶段流程

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

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

6.2 `RevertEvent.Committed` 的投影逻辑

// packages/core/src/session/projector.ts:415-454
yield* 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），事件日志保持完整。

这意味着：

审计性：所有历史事件仍然可查，回滚操作本身也是一个事件（RevertEvent.Committed），记录了回滚到哪个边界。
可重放性：如果需要，可以通过 replayAll 从事件日志重建任意时间点的状态——包括回滚前的状态。
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-604
const durable = (input: { aggregateID: string; after?: number }) =>
  Stream.unwrap(
    Effect.gen(function* () {
      const wakes = yield* subscribeDurable(input.aggregateID)
      let sequence = input.after ?? -1
      const 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 不具备的能力：

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

八、附录：关键行号索引

概念	文件	行号
`commitDurableEvent` 事务体	`packages/core/src/event.ts`	205-367
Projector 事务内执行	同上	320-322
序列号分配 `latest + 1`	同上	294
幂等深比较 `isDeepStrictEqual`	同上	269-273
分歧检测 `Replay diverged`	同上	284-289
Event ID 唯一性检查	同上	303-315
序列号连续性检查	同上	295-301
owner_id 互斥	同上	254-261, 291-293
`publishEvent` + `notify`	同上	369-396
`replay`	同上	441-478
`replayAll` 前置校验	同上	480-512
`durable` stream	同上	585-604
`subscribeDurable` PubSub	同上	565-583
`EventSequenceTable` / `EventTable` 定义	`packages/core/src/event/sql.ts`	4-25
`SessionProjector` 28 个投影器注册	`packages/core/src/session/projector.ts`	211-456
`run(db, event)` Adapter 模式	同上	112-191
`insertMessage`（seq 来自事件）	同上	193-209
`RevertEvent.Committed` 截断逻辑	同上	415-454
`applyUsage` 增量更新	同上	90-110
`SessionMessageUpdater.update` 模式匹配	`packages/core/src/session/message-updater.ts`	78-395
事件定义 + durable 配置	`packages/schema/src/session-event.ts`	38-43, 54+
`Event.define` 工厂	`packages/schema/src/event.ts`	—
`Durable` manifest	`packages/schema/src/durable-event-manifest.ts`	—