跳到主要内容

数据截至 (上游 commit 99f6f02fecdb)

第 2 章 · 唯一事实源:append-only 会话事件日志与 surface 投影

30 秒导读: dsh 把一次 agent 会话的全部事实压在一条只能追加、不能改写的事件日志里,模型每次请求看到的消息数组不是被谁维护出来的,而是从这条日志投影出来的。这一章讲这条日志怎么写、怎么投影、怎么落盘、怎么在崩溃后重放。

本章不讲 turn / step 控制流怎么驱动——那是第 3 章。这里只讲数据模型。


1. 这一章在解决什么问题

先说结论:agent 的"记忆"如果由多个地方各自维护,早晚会对不上。

一个朴素实现通常有三份状态:内存里的 messages 数组、UI 上显示的对话、磁盘上的存档。它们靠"每次都记得同步一下"保持一致。只要有一处忘了同步,就会出现三类经典事故:

事故表现
界面有、模型没有用户看到一条注入的上下文,模型请求里却没带上
模型有、存档没有进程崩溃后恢复,模型突然"忘了"上一轮做过什么
压缩把历史改坏了压缩摘要覆盖了原文,UI 上用户已经读过的内容凭空消失

dsh 的处理办法是取消同步这件事:只留一份事实,其余全部是它的投影。

一句话规则

model-visible ⟺ logged:任何能进入一次模型请求的东西,都必须能从会话日志重建;新增一种模型可见输入,就必须新增一种会话事件。

这条规则在代码里有强制检查点(见 §9),不是口号。

三个承重词,先定义清楚

本章反复出现三个词,各有唯一含义,不互相借用:

指什么能不能改
日志(log)SessionEvent[],一次会话发生过的全部事实,按 seq 连续编号只能在尾部追加;已写入的事件深度冻结
surface(模型可见面)日志里"会变成一条 LLM 消息"的那些事件的有序编号列表可以被 replace 操作遮蔽某一段,但不动日志
投影(derive)把 surface 的每个节点算成一条 Message 的纯函数结果缓存产物,随时可从日志重算

一个直觉类比:日志是记账凭证(一张都不能撕),surface 是当期科目余额表(可以把一批旧凭证结转掉),投影是打印出来的报表


2. 顶层全景:一条事件的一生

怎么读这张图:从上往下是一次 session.append() 的时间顺序,"提交线"以下的步骤一旦开始就不可撤销。

调用方: session.append('assistant/message', data, { surfaceOp: 'append' })

├─ ① 快照 + 校验:data 必须是无损 JSON,深拷贝一份
├─ ② 重入拒绝:另一次 append 正在广播时直接抛错
├─ ③ 盖信封:seq = log.length,time = now,整个事件深度冻结
├─ ④ surface 预校验:只"算"出一个 plan,不改任何状态

═════ 提交线 ═══════════════════════════════════════════

├─ ⑤ 先解析监听器快照(此处仍可否决 → 日志未变)
├─ ⑥ push 进 log —— 从这一刻起,这条事件是既成事实
└─ ⑦ 逐个回调 session/event,任何监听器抛错只写 warn 日志

┌───────────┼───────────────┬────────────────────┐
▼ ▼ ▼ ▼
SurfaceManager 持久化后端 session-projection UI / RPC 前端
(投影缓存) (write-behind) (派生视图缓存) (逐字稿)

部件一句话职责

部件干什么在哪个文件
Session持有日志、执行 append、维护三个增量 foldpackages/core/session/src/index.ts:425
SessionStorectx.sessions,会话的在内存注册表与生命周期packages/core/session/src/index.ts:792
SurfaceManager增量维护模型可见面,并在 append 前做校验packages/core/session/src/surface.ts:398
deriveEventMessage单个事件 → 单条 Message 的唯一投影规则packages/core/session/src/surface.ts:83
持久化协调器订阅 session/event,批量落盘,session/flush 是屏障packages/session/session-persistence/src/coordinator.ts:1086
jsonl / sqlite 后端两种物理存储格式,读路径语义对齐packages/session/session-persistence-jsonl/src/index.ts:121packages/session/session-persistence-sqlite/src/index.ts:52
不变量companion运行时检查日志关系与"请求 ⟺ 日志"一致packages/core/session/src/invariant.ts:190packages/core/agent-loop/src/invariant.ts:19

3. append 路径:一次写入要过五道关

这节讲 Session.appendpackages/core/session/src/index.ts:604)——整个系统唯一的写入口。

3.1 为什么第一步是"深拷贝"而不是直接存引用

它要解决的小问题: 调用方传进来的 data 是个活对象。如果日志直接存引用,调用方之后改一个字段,历史就被偷偷改写了;更阴险的是 getter——校验时返回 A,落盘时返回 B。

做法: snapshotJsonValuepackages/core/session/src/json.ts:177)在一次遍历里同时完成"校验 + 拷贝",每个属性只读一次。它拒绝一切 JSON 存不下的东西:

被拒绝的值为什么
BigInt / 函数 / symbol / undefinedJSON 里没有对应表示
-0NaNInfinityround-trip 后不是同一个值
循环引用、稀疏数组无法无损序列化
Map / Set / Date / class 实例原型不是 intrinsic Object.prototype / Array.prototype

为什么在这里拒绝而不是落盘时拒绝: 日志是唯一事实源,一条存不下的事件必须在 append 现场炸掉,而不是等到几百毫秒后后端 flush 时才失败——那时调用方早就返回了(packages/core/session/src/index.ts:590-602 的契约注释)。

原型检查是逐层做的(json.ts:16-49hasIntrinsicConstructor / hasPlainObjectPrototype),所以跨 realm 的普通对象也接受,伪造原型的对象不接受。遍历是迭代式的(显式任务栈,json.ts:89-161),所以深层嵌套受内存限制,而不是受 JS 调用栈限制。

3.2 信封:seq = log.length 是全系统的地基

const event = deepFreeze({
type,
seq: this.log.length, // 契约:seq 必须与数组下标一致
time: Date.now(),
data: dataSnapshot,
...surfaceMetadataSnapshot,
})

packages/core/session/src/index.ts:627-633

seq 不是自增计数器,就是当前数组长度。这条契约让"seq → 事件"永远是 O(1) 下标访问,surface 的节点列表因此可以只存数字(surface.ts:139nodes: readonly number[])。种子(seed)载入时也用同一条规则强校验:第 index 个事件的 seq 必须等于 index,否则构造直接失败(index.ts:525-527)。

deepFreeze 作用在整个事件上。所以 session.events 返回的东西,即使调用方用 as any 强转,也改不动(index.ts:556-562)。

3.3 重入拒绝:不许在广播里再写一条

const entry = attachments.get(this)
if (entry?.appending) {
throw new Error('session append cannot reenter while another append is being published')
}

packages/core/session/src/index.ts:623-626

为什么必须禁: session/event同步广播。如果某个监听器在回调里又 append 一条,第二条的 seq 会在第一条的广播还没走完时就被派发出去,下游拿到的顺序与日志顺序不一致。直接拒绝比让下游各自防御便宜得多。

注意这个标志位挂在 store 条目上(SessionEntry.appendingindex.ts:409),所以没进 store 的游离 Session 不受这条限制——它本来也没有广播。

3.4 surface 预校验:先"算",后"改"

this.surfaceManager.validateNext(event)index.ts:634)在事件进日志之前跑完全部 surface 规则,但只产出一个 plan,不动 nodes。真正的 splice/push 发生在下一次读取 surface 时(surface.ts:444_processDelta)。

好处: 校验失败时 surface 状态一个字节都没变,不存在"改了一半"的中间态。种子载入走的是同一条路径(index.ts:531-535),所以磁盘上的日志和内存里的 append 受完全相同的约束。

3.5 提交与容错分发:顺序被刻意排过

这是全章最值得抄的一段设计:

let callbacks: SessionCallback[] | undefined
if (entry !== undefined) {
callbacks = collectSessionCallbacks(entry.emitCtx, [entry.carrier, 'session/event', ...])
}
this.log.push(event as SessionEvent) // ← 提交点
this.eventsSnapshot = undefined
if (callbacks !== undefined && entry !== undefined) {
invokeContainedSessionObservers(entry.emitCtx, 'session/event', entry.id, callbackArgs, callbacks)
}

packages/core/session/src/index.ts:638-648

三步的顺序各有理由:

步骤时机理由
解析监听器快照push 之前Cordis 的 internal/dispatch 校验此时还能抛错否决,而日志还没变
log.push中间这是提交点;之后 append 一定返回成功
逐个调用回调push 之后监听器读到的 session.events 必须已经包含这条事件

回调的容错在 invokeContainedSessionObserversindex.ts:382-399):每个监听器单独 try/catch,同步抛错记 warn,返回的 promise 若 reject 也记 warn。一个坏掉的持久化插件不会让 append 失败,也不会让排在它后面的监听器收不到事件。

finally 里还有一处细节:如果某个监听器在回调中触发了 detach,detach 会被推迟到本次广播结束(index.ts:649-654index.ts:934-945),保证"先把这条事件广播完,再拆钩子"。

3.6 两个入口函数:adopt 还是 snapshot

外部把事件"送进"会话时(持久化恢复、跨进程导入),有两个语义不同的入口:

函数前提行为用在哪
adoptSessionEvent调用方独占这份对象图,不再持有可变别名原地校验 + 冻结 message,不拷贝崩溃修复合成的闭合事件(coordinator.ts:903
snapshotSessionEvent所有权不确定structuredClone 后再 adopt查询 / 持久化边界

packages/core/session/src/index.ts:167:192

adopt 里的校验不是走形式:assertMessageEventShapeindex.ts:301)要求 user/message 必须有非空 id、role 必须匹配、tool/result 必须恰好一个 tool-result 块且 toolCallIdsource.callId 对得上。一条对不上的历史宁可载入失败,也不能进日志。


4. 事件族:一张能被插件扩表的类型表

4.1 合并可扩展

事件词表是一个 interface,不是 enum:

export interface SessionEventMap {
'turn/start': { turn: number }
'user/message': UserMessage
'assistant/chunk': { turn: number; step: number; chunk: StreamChunk }
'assistant/message': { turn: number; step: number; message: AssistantMessage; usage?: TokenUsage }
'tool/result': { turn: number; step: number; message: ToolResultMessage; error?:; meta?: JsonValue }
// …
}

packages/core/session/src/types.ts:236

插件用 TypeScript 的 declaration merging 往这张表里加自己的键(compaction 加 compaction/*,hook 桥加 hook/*),SessionEvent 是对这张表的映射类型联合types.ts:404-436),所以 switch (event.type) 能自动收窄 event.data,不需要任何断言。

代价是这个联合不封闭:核心代码里所有对 event.type 的 switch 都必须有 default 分支,不能用 assertNeversurface.ts:109-113 明确写了这一点)。

核心事件按用途分四类:

类别事件进不进模型可见面
控制流边界turn/startturn/endstep/startstep/end
模型可见消息user/messageassistant/messagetool/result(且必须带 surfaceOp
请求重建request/headerrequest/context否(走单独的 fold)
追踪 / UI 状态assistant/chunktool/calltodo/writesession/end-seed

tool/call 不进可见面,因为工具调用块本来就嵌在 assistant/message 里;它存在只是为了让"这次调用什么时候开始的"有个时间点,崩溃修复要读它(repair.ts:59-67)。

4.2 三层兼容规则

一个"新写、旧读"的日志会怎样?dsh 用三层机制回答,各管一段:

机制粒度遇到不认识的东西时
SESSION_FORMAT_VERSION整个日志格式版本不等 → 直接拒绝载入,提示"升级 harness"
KNOWN_SESSION_EVENT_TYPES单个事件类型类型不在集合里 → 拒绝解释整条日志
ignorable: true单条事件写入方声明"丢了也不影响重建" → 允许跳过

(依据:types.ts:56known-event-types.ts:19types.ts:422

默认是"必需",不是"可忽略"——读到不认识且没标记的事件,读方必须拒绝重建而不是静默跳过:

if (KNOWN_SESSION_EVENT_TYPES.has(event.type) || event.ignorable === true) continue
throw this.unsupported(meta, `session "…" contains event type "${event.type}" … refusing to interpret the log`)

packages/session/session-persistence/src/coordinator.ts:1061-1066

理由写在类型注释里:一条不认识的必需事件可能改变后面整段日志的解释方式(比如一个新的 surface 操作),跳过它等于重建出一个错误的会话。忘记标 ignorable 的后果是"过度拒绝"(不方便),忘记检查的后果是"静默读错"(灾难)——所以默认取前者。

KNOWN_SESSION_EVENT_TYPES 这份清单是生成的scripts/gen-persistence-catalog.ts 扫全仓库的 SessionEventMap 声明),不是手写维护的。

什么时候该 bump 版本号? 判据是写入方发出了什么,不是读方能接受什么(types.ts:47-54):只有 header 结构、事件信封、核心事件语义、或 surface 机制本身(SurfaceEventType 集合与 SurfaceOp 变体)变了才 bump;新增一个普通事件类型不 bump——那由 ignorable 那层负责。当前值锁在 0,未发布期不承诺任何兼容。


5. surface:模型可见面

5.1 只有三类事件能进

const SURFACE_EVENT_TYPES = new Set<string>([
'user/message',
'assistant/message',
'tool/result',
])

packages/core/session/src/surface.ts:15-19

这是个双向约束,由 surfaceOpOfsurface.ts:185-208)在每次 append 时执行:

  • 这三类事件必须surfaceOp,否则抛错;
  • 其它任何事件不许surfaceOpsourceEventSeqs,否则抛错。

TypeScript 层面也拦了一道:append 的第三参数是条件类型,非 surface 类型传 opts 编译不过(index.ts:604-608)。运行时那道是给"从磁盘/网络读进来的日志"准备的。

这条约束的价值: 不存在"这条事件到底算不算模型可见"的模糊地带。每一条消息型事件都自己声明了它怎么进入可见面。

5.2 surfaceOp:append 还是 replace

export type SurfaceOp =
| 'append'
| { op: 'replace'; start: number; end: number }

packages/core/session/src/types.ts:376-378

replace 是整章的关键设计。看图(怎么读:上面是日志,下面是可见面;replace 只改下面那行):

日志(永不删改,seq 连续)
seq 0 1 2 3 4 5 6 7
user asst tool user asst · · user'
· = 非 surface 事件(chunk / turn 边界 / compaction/summary)
└────────── 被 7 号遮蔽 ──────────┘ user' 带 surfaceOp = {op:'replace', start:0, end:4}

surface.nodes
之前: [0, 1, 2, 3, 4]
之后: [7] replaceGeneration: 0 → 1

压缩因此不是"改写历史",而是"在可见面上遮蔽一段旧节点"。 原文 5 条事件一个字节都没动,只是不再进入下一次模型请求。

真实使用者有两个,取舍不同:

使用者遮蔽范围替代节点代码
dsh-compaction-basic一整段区间 start..end一条摘要 user/messagepackages/compaction/compaction-basic/src/region.ts:462-465
dsh-compaction-tool-result-pruner恰好一个节点内容被裁剪的同一条 tool/resultpackages/compaction/compaction-tool-result-pruner/src/index.ts:168-173

5.3 replace 的四道校验

一次 replace 要过四关,任何一关不过就在 append 现场抛错:

校验规则代码
标记结构{op,start,end} 必须恰好三个键,start/end 是非负安全整数surface.ts:173-182 (isReplaceOp)
区间有效startend 必须都是当前可见面上的节点,且 start 不在 end 之后surface.ts:246-266 (replacementRange)
溯源完整sourceEventSeqs 必须包含每一个被遮蔽的节点,元素不重复且都早于自己surface.ts:211-243 (assertProvenance)
工具结果专项tool/result 的替换只能覆盖一个当前 tool/result 节点,且只允许改 contentsurface.ts:287-318 (assertToolResultRewrite)

第三关是"可审计"的来源:拿着替换节点,就能查到它到底吃掉了哪些原始事件。第四关最有意思——它把 original.dataevent.datacontent 都置成 null 后做深比较(surface.ts:302-315),于是"裁剪输出"合法,"顺手改掉 callId 或错误标记"非法。裁剪器不能借压缩之名改写工具身份。

5.4 为什么人类逐字稿不能读 surface

这是本节最容易踩的坑,而且代码里专门留了一对类型守卫来防:

export function isAppendSurfaceEvent(event: SessionEvent):
event is SurfaceEvent & { surfaceOp: 'append' } {
return isSurfaceEvent(event) && event.surfaceOp === 'append'
}

packages/core/session/src/surface.ts:51-55;对偶的 isReplacementSurfaceEvent:64

道理: 可见面是故意遮蔽历史的——那是给模型省 token 用的。但用户已经在屏幕上读过被遮蔽的那五条消息。如果 UI 也照着 surface.nodes 渲染,一次压缩落地会让用户眼前的对话凭空消失

所以分工是:

消费者读什么结果
模型请求session.surface.nodes被压缩的旧内容消失 → 省 token
人类逐字稿 / UI全日志里 isAppendSurfaceEvent 为真的事件用户读过的东西永远在

UI 侧的确是这么用的:packages/client/ui-conversation/src/client/conversation-nodes/turn-tail.ts:39packages/client/ui-conversation/src/client/conversation-nodes/assistant.ts:251packages/client/ui-conversation/src/client/conversation-nodes/tool.ts:241 全都用 isAppendSurfaceEvent 过滤;替换节点则由专门的节点类型渲染成"这里发生过一次压缩"(packages/client/ui-conversation/src/client/conversation-nodes/command.ts:83isReplacementSurfaceEvent)。API 代理导出对话时同样只取 append-origin(packages/host/apiproxy/src/api-proxy.ts:238)。

5.5 SurfaceManager:增量维护

SurfaceManagersurface.ts:398)实现只读接口 SessionSurface,直接引用会话的私有 log 数组(index.ts:428),因此不需要任何"通知我日志变了"的机制——每次读 nodes 时,它先把上次处理位置之后的新事件补折一遍(surface.ts:444_processDelta)。

validateNext 算出的 plan 会被缓存在 _pendingPlan_processDelta 走到那条事件时直接复用,不重算(surface.ts:450-456)。

同一套折叠逻辑还有一个纯函数出口 foldSurface(events)surface.ts:387),给离线重建用:拿到任意一段日志前缀,能算出当时的可见面和全部替换历史。在线增量与离线全量共用 planSurfaceEvent / applySurfacePlan,所以两条路径不可能算出不同结果。


6. 三个增量 fold:投影缓存长什么样

Session 上挂着三个"折叠缓存",形状统一:一份状态 + 一个水位线(已消费到哪个 seq)。

方法折叠什么水位字段失效条件
deriveMessages()surface 的每个节点投影成一条 MessagederivedNodesreplaceGeneration 变了 → 整份重建
requestHeader()request/header 事件,取最后一个快照headerFoldSeq只增不减,无需重建
requestContext()request/context 事件,取最后一个contextFoldSeq同上

packages/core/session/src/index.ts:726:670:691

6.1 deriveMessages:每个节点只投影一次

const generation = surface.replaceGeneration
if (generation !== this.derivedGeneration) {
this.derived = []; this.derivedNodes = 0; this.derivedGeneration = generation
}
for (const seq of nodes.slice(this.derivedNodes)) {
const msg = this.deriveEventMessage(this.log[seq]!)
if (msg) this.derived.push(msg)
}

packages/core/session/src/index.ts:729-744

三个要点:

  1. 正常情况是 O(新增节点):一次 step 只投影新长出来的那几条。
  2. replace 触发整份重建replaceGeneration 是单调计数器(surface.ts:370),一变就清空缓存重来。压缩不常发生,所以用"简单正确"换"偶尔重算"是划算的。
  3. 返回的数组是新的,里面的 Message 是共享且深冻结的index.ts:746[...this.derived])。调用方拿到的数组不会在下次 append 后偷偷变长;而消息对象复用事件里已经冻结的那份,省掉第二次深拷贝。

没有原始日志兜底路径。 一条消息要么在 surface 上,要么模型看不见——不存在"surface 没有但从 log 里捞出来"的后门。

6.2 deriveEventMessage:唯一的投影规则

switch (event.type) {
case 'user/message': return event.data
case 'assistant/message':
if (event.data.message.content.length === 0) return null // 只承载 usage 的空消息
return event.data.message
case 'tool/result': return event.data.message
default: return null
}

packages/core/session/src/surface.ts:83-114

它是纯函数、可单独导出,这点很重要:在线的 deriveMessages、离线重建器、以及 §9 的一致性检查,折叠的是同一个函数,因此不可能出现"在线算出来的请求"和"离线重放出来的请求"不一致。

投影是逐字透传的——不加任何 <context> 之类的包装。注释里明确写了这是刻意的:包装归生产方所有(比如 agent-instructions 自己把 <system-reminder> 烘进 content),投影层保持哑管道(surface.ts:89-95)。

空内容的 assistant/message 返回 null 是个真实边界:一个撞到 max-tokens 的 step 会产出一条只带 usage 的空消息,它必须留在日志里(token 账要算),但不能变成一个空的 assistant 轮次塞给 provider。

6.3 requestHeader:模型请求的另一半

模型请求 = 消息数组 + 请求头(system prompt、工具 schema、call config)。消息数组走 surface,请求头走 request/header 事件的折叠:

this.headerFold = deepFreeze(foldRequestHeader(this.log.slice(this.headerFoldSeq), this.headerFold))

packages/core/session/src/index.ts:676

foldRequestHeaderrequest-header.ts:65)就是"取最后一个 request/header 快照"——每个 header 事件都是全量快照,不是 delta(老的 request/header-delta 格式在载入时被显式拒绝,index.ts:215-217)。

配套的 canonicalHeaderrequest-header.ts:21)把空 system 和空 tools 规范成"字段缺席",headerEquals:44)按字段比较——循环用它判断"这次请求的头和上次一样吗",一样就不写事件。折叠结果被 deepFreeze,因为它是按引用暴露的会话状态,就地改会让后续所有比较失准。


7. 生命周期:SessionStore 与所有权移交

7.1 三段式发布

SessionStore 提供的不是一个 create(),而是三个可分离的原语:

prepare(id, options) ──► [调用方自己组装] ──► enter(session) ──► announce(session)
构造 Session 还没进 store 装广播钩子 发 session/created
校验 header/seed + 进 store 监听器同步抛错
不广播任何东西 返回 detach 闭包 ⇒ 整个发布回滚

packages/core/session/src/index.ts:863:913:968

create()index.ts:830)只是把三步串起来的便利函数,但串的顺序有讲究

this.ctx.effect(function* (this: SessionStore) {
yield this.enter(session) // 先把 detach 交出去
this.announce(session) // 再广播
}, 'sessions.create()')

index.ts:836-839

yield detach 再 announce,是为了让"session/created 监听器抛错"这件事能回滚——生成器 effect 在抛错时会依次释放已 yield 的 disposer,于是 store 条目和钩子一起消失,而不是留下一个半死不活的会话。

那为什么还要暴露拆开的三步?因为 agent 需要把会话的生命周期折进自己那一个 effect:如果会话和 agent 是两个平行 effect,卸载时它们会赛跑,可能在循环写完最后几条事件之前就把广播钩子拆了,事件直接丢失(index.ts:843-855 的契约说明)。

7.2 SessionPreparation:未发布状态的所有权

prepare 出来的 Session 还没进 store,可能永远不进(组装失败)。谁来释放 provider 侧为它保留的状态?

export class SessionPreparation implements Disposable {
[Symbol.dispose](): void {
if (this.released) return
this.released = true
this.options.release?.()
}
}

packages/core/session/src/preparation.ts:20-48

using 语法就能保证"要么发布、要么释放",且释放是同步幂等的。发布路径可以先把 provider 状态消费掉,让 release 变成空操作。

7.3 fork:从一个稳定前缀分叉

fork(source, boundary?, childSessionId?): Session

packages/core/session/src/index.ts:1081

边界规则由 _forkSeedindex.ts:1097)执行,四种拒绝码都是 SessionForkError

情况
源 id 不在 store / 传进来的对象不是 store 里那个实例SESSION_NOT_FOUND / SESSION_NOT_LIVE
子 id 已被占用SESSION_ALREADY_EXISTS
边界不是一个连续存在的 seqINVALID_BOUNDARY
选中的前缀停在一个尚未闭合的 turn 里OPEN_TURN

最后一条是关键:往回找最近的 turn/start / turn/end,如果找到的是 turn/start,说明这个前缀中间截断了一轮对话——分叉出来的会话会带着"一个开着的 turn",工具调用没有结果,provider 会直接拒收(index.ts:1128-1135)。

分叉出的子会话在 header 里记 parentSessionseedLengthindex.ts:1088-1093),血缘可查。

7.4 header 与 session/end-seed:两种"种子边界"

这两个概念很容易混,列表对照:

概念含义存在哪
header.seedLength持久的分叉血缘边界:从父会话继承了多少条会话 header(不在日志里)
Session.firstLiveSeq本进程第一条自己写的事件的 seq内存字段,不持久化
session/end-seed 事件firstLiveSeq日志投影,给读存档的人看日志里

packages/core/session/src/types.ts:80index.ts:472types.ts:332

差别在恢复场景:一个 resume 出来的会话,它的构造种子是整份存档日志,但 header 里的 seedLength 还是当初那个分叉值。

session/end-seed 有个防重复的细节:种子最后一条已经是它就不再补(index.ts:545-547),否则反复打开一个没动过的会话,日志会每次长一条。读存档时要找最后一个该事件,而不是假设它在 firstLiveSeq 位置上。

header 本身刻意不进日志index.ts:436-443):它是存储关切(格式版本、cwd、血缘),不是可重放的对话状态。


8. 落盘与回放

8.1 写路径:日志广播 → 有界批写 → 后端

持久化是纯粹的插件关切,核心包里一行 I/O 都没有。协调器订阅三个事件就够了:

session/created → 记下 header,分叉种子落一次盘
session/event → live.writes.enqueue(event) ← 只入队,不阻塞 append
session/flush → this.flush(session) ← 调用方的即时耐久屏障
session/disposed → retire(session)(最后一次排空)

packages/session/session-persistence/src/coordinator.ts:1117-1132

中间那层是 SessionWriteBehindpackages/session/session-persistence/src/write-behind.ts:22):拿一份持久化自己的事件副本,用一个固定的批处理截止时间攒批,后台写失败单独上报而不去毒化生产者。

于是热路径(append)永远不等 I/O,需要确定性的地方(每次请求前的检查点、空闲检查点、要读存档之前)用 ctx.sessions.flush(session) 当屏障——那是一个并行 awaited 的事件(index.ts:1022,所有监听器都跑、都等、第一个失败在全部落定后抛出)。

8.2 两种物理格式

jsonl 后端sqlite 后端
布局每会话一个文件;第一行是 type:'session' 的 header 行,后续每行一条记录一个 db;sessions 元数据表 + events 表,事件 1:1 一行
版本SESSION_FORMAT_VERSION另有 SCHEMA_VERSION = 15,与会话格式版本正交
惰性物化首次 append 前不建文件首次 append 前不写 sessions 行,因此不出现在 list()
列表加速parseHeaderMeta 只解析第一行直接查元数据表
相关代码packages/session/session-persistence-jsonl/src/format.ts:33,221,404packages/session/session-persistence-sqlite/src/schema.ts:20,32,49

sqlite 的 events 表把 sourceEventSeqssurfaceOp 存成 JSON 文本列、ignorable 存成 1/null(schema.ts:49-61)——信封字段是,不是塞在 data 里,所以按 surface 属性筛事件不用解析 payload。

8.3 chunk 行:为压缩流式增量而生的存储词表

流式 provider 每个 token 发一个 delta,于是日志里出现成百上千条几乎一样的行,JSON 信封比 payload 还大(模块注释里写的实测约 56 倍,packages/core/session/src/chunk-rows.ts:1-6)。

packChunkRunschunk-rows.ts:192)把连续的同块同类 delta 打包成一行:

行类型打包什么存法
text-chunks连续 text-deltaseq0 + time0 + 各成员 texts[] + 时间差 dt[]
reasoning-chunks连续 reasoning-delta同上
tool-call-chunks连续 tool-call-delta多一个整段恒定的 id/name + args[]

三条设计约束值得抄:

  1. 存储行不是会话事件。 它们用不带斜杠的 tag(text-chunks 而非 text/chunks),永远不会进 Session.events,读方一眼能分辨(chunk-rows.ts:10-14)。
  2. 白名单编码。 classify:96)逐键精确匹配 shape,任何不完全认识的东西原样存——未来新增的 chunk 变体只会失去压缩,绝不会丢数据。
  3. 解码严格。 decodeStorageRecord:339)遇到打了 chunk-row tag 却结构不对的行直接抛错,因为把它当普通事件放过去等于静默丢掉一整段流(validateRow 连"成员 seq / time 加下去会不会溢出安全整数"都查了,:281-288)。
  4. MIN_RUN = 3:78)是格式常量不是可调参数:两种布局解码结果完全一致,所以改它不会让旧日志失效。

8.4 读路径:三道拒绝 + 一次修复

载入一份存档要过的顺序是固定的(coordinator.ts:895-908):

读原始工件

├─ ① 身份核对:header.id 必须等于请求的 id
├─ ② 格式版本:meta.version ≠ SESSION_FORMAT_VERSION → 拒绝(提示升级,不说"日志损坏")
├─ ③ 事件词表:未知类型且未标 ignorable → 拒绝

├─ ④ 崩溃修复:interruptedTurnClosers(events) 合成缺失的闭合事件

└─ ⑤ sessions.prepare(id, { seed: balanced, meta, seedSource: 'persistence' })
└─ 走 Session.fromRestore:所有权移交,原地校验 + 冻结,不再拷贝一次

②③ 的顺序不能反。 jsonl 后端甚至在校验 header 结构之前就先看版本号(format.ts:240-247refuseForeignFormatVersion):一个未来格式没义务满足今天的结构检查,用户该看到的是"升级 harness",绝不是"会话日志损坏"。

截断修复: jsonl 扫描器把最后一条没有换行的记录当作"撕裂的尾巴"直接忽略,并返回一个可安全续写的字节偏移 committedBytesformat.ts:341-344)。但如果一条坏行后面还出现了 turn/end,说明损坏发生在已提交区间里,这时直接抛错而不是悄悄截断(format.ts:347-372)。sqlite 后端用"最后一条 turn/end 作为切口"实现同一语义(schema.ts:232-257scanRows),两个后端的崩溃语义因此对齐。

崩溃闭合: interruptedTurnCloserspackages/core/session/src/repair.ts:27)扫日志找出"assistant 请求了工具但没有结果"的调用,按顺序合成:

合成事件内容为什么
每个悬空调用一条 tool/resultisError,错误码 TOOL_OUTCOME_UNKNOWNTOOL_NOT_STARTEDprovider 拒收带悬空调用的 transcript
一条 step/end(若有开着的 step)开着 step 时发 turn/end 违反不变量
一条 turn/endreason: { kind: 'interrupted' }标记这是后端补的,不是循环发的

两个错误码的模型面文案不同,这是个细节但很关键:已记录开始的调用(TOOL_OUTCOME_UNKNOWN)告诉模型"结果未知,只有只读或幂等操作才能盲重试";未记录开始的(TOOL_NOT_STARTED)直接说"需要就重试"(repair.ts:100-107)。合成事件的 time 复用最后一条真实事件的时间戳,保证确定性、不发明未来时间。

另外:一个 turn 还开着的会话不允许被当存档读走(coordinator.ts:981-983),必须用活的 Session 或等 turn 结束。

8.5 投影缓存:让派生视图也别每次重算

日志之上还有一层通用的派生视图机制 session-projectionpackages/session/session-projection/src/index.ts)。它的形状和 §6 一模一样,只是开放给插件:

interface ProjectionDefinition<K, S> {
key: K
schema: ZodType<> // view 输出的线上校验
init(): S // 空日志的初值
apply(state, event): S // 纯转移;不关心的事件必须返回同一个引用
view(state):// 状态 → 线上整值
stateVersion: number // 持久缓存失效版本
}

packages/session/session-projection/src/index.ts:42

四条规则决定了它能安全缓存:

  • 三个函数都必须同步——异步单元会撕裂消费者的一致性切面。
  • 不感兴趣就返回同一个引用Object.is 不变 ⇒ 下游零工作。
  • 事件必须携带完整的变更后状态,不能是裸 delta(模块注释里叫 whole-value event rule)。
  • 持久化下来的 (sessionId, key, ver, seq, val)永远不是权威,只是折叠的捷径:版本对不上、或它声称的 seq 超过了存档末尾,就丢掉重折(:108-118)。

注册表自己订阅一次 session/event 然后驱动所有单元(:181),域插件只提供数学,不碰订阅。


9. 不变量检查点:把"model-visible ⟺ logged"变成可执行断言

规则写在文档里没用,dsh 把它编译成两个运行时 companion 插件。

9.1 会话包:日志的关系不变量

packages/core/session/src/invariant.ts:190 挂两个钩子,分工很精巧:

钩子时机做什么
internal/dispatchappend 提交前validateEvent,算出一个"待提交转移",暂存在以事件为弱键的 Map 里
session/eventappend 提交后取出暂存的转移并应用;取不到就直接判失败

invariant.ts:223-241

为什么要拆两半?因为校验必须在日志变更之前(这样才能否决),而状态推进必须在提交之后(否则被否决的事件会污染追踪状态)。校验是纯函数,被丢弃的转移不会留下任何痕迹。

它检查的关系包括:seq 严格递增、turn/step 正确嵌套、tool/result 必须配得上同一 step 里的 tool/calltodo/write/request/header/request/context 必须被 turn 包住(invariant.ts:55-166)。

两个刻意的例外值得看:

  • session/end-seed 不受任何约束——一段不平衡的种子完全合法地会把它放在一个开着的 turn 里(invariant.ts:147-149)。
  • tool/result替换节点不要求配对 tool/call——它是被 Session 校验过的内容改写,不是第二次执行(invariant.ts:128-135)。

9.2 循环包:请求必须等于日志的投影

这是"model-visible ⟺ logged"最直接的那个检查点:

const expected = session.deriveMessages()
if (JSON.stringify(options.messages) !== JSON.stringify(expected)) {
fail(`llm request for session "…" diverges from the dispatch-time durable derivation (log-reconstruction desync)`)
}

packages/core/agent-loop/src/invariant.ts:39-42

它挂在 llm/stream 上并且 prepend——放在最前面,免得某个短路的 replay 监听器把检查跳过去(invariant.ts:55)。同一处还核对请求头:model / system / temperature / maxTokens / stop / tools 必须与 foldRequestHeader(events) 的结果逐项相等(:44-52)。

读法: 任何一次真实发出的模型请求,如果不能被"当时那份日志"精确重放出来,就是 bug,立刻炸。这条断言把整章的所有设计钉在地上。


10. 巧妙之处(可以直接搬走的几条)

  1. 校验与拷贝在同一次遍历里完成。 每个属性只读一次,所以一个有状态的 getter 没法"给校验一个值、给存储另一个值"(packages/core/session/src/json.ts:165-178)。

  2. 先算 plan、后提交。 surface 转移分成 planSurfaceEvent(纯计算)和 applySurfacePlan(纯变更)两半,失败永远不会留下半改的状态;在线增量和离线全量共用这两个函数,因此结果必然一致(surface.ts:321:362:387)。

  3. 监听器快照在 push 之前、调用在 push 之后。 一句话卡住了"能否决"和"能看到"这两个互相矛盾的需求(index.ts:638-648)。

  4. 压缩用遮蔽而非改写,再用一对类型守卫把两类消费者分开。 模型省 token,用户的逐字稿不丢(surface.ts:51:64)。

  5. 未知事件默认拒绝,ignorable 由写入方主动声明。 把"忘了标记"的后果从"静默读错"降级成"过度拒绝"(types.ts:414-422coordinator.ts:1061)。

  6. 存储词表和事件词表用命名区分。 chunk 行用不带斜杠的 tag,物理上无法与事件类型混淆(chunk-rows.ts:10-14)。

  7. 崩溃合成事件复用最后一条真实事件的时间戳。 修复因此是确定性的、可快照测试的,也不会在日志里出现"未来时间"(repair.ts:82-86)。


11. 边界与局限(诚实版)

  • 单进程单写者。 session/end-seed 的注释明确说了它不是关于其它写入者的存活信号——要容忍并发写入者,需要日志之外的机制(types.ts:325-330)。
  • 无迁移。 SESSION_FORMAT_VERSION 锁在 0,旧格式一律拒绝载入,不提供 migration(types.ts:49-51)。完整的升级链机制只存在于设计笔记里。
  • replace 会让派生缓存整份重建。 replaceGeneration 一变,deriveMessages 的缓存全清(index.ts:730-734)。压缩频繁的会话会付这个代价。
  • replace 也会打断 KV cache 复用。 日志虽仍是追加的,但可见面从第一个被遮蔽的消息起就变了,前缀复用从那里断掉(依据:packages/core/session/README.md 的 KV Cache effect 段)。
  • fork 只接受落在完整 turn 边界上的前缀,中途分叉会被 OPEN_TURN 拒绝(index.ts:1128-1135)。
  • 插件事件的关系不变量归插件自己管。 核心的 validateEvent 对合并进来的类型走 default 分支,什么都不查(invariant.ts:158-161)。
  • KNOWN_SESSION_EVENT_TYPES 只覆盖本仓库。 仓库外的下游插件事件天然不在这张表里,注册接口被显式推迟到"真有这种消费者"再做(known-event-types.ts:15-18)。

12. 代码地图(导航索引)

主题文件路径符号名
会话本体与写入口packages/core/session/src/index.tsSessionSession.appendSession.events
事件采纳 / 快照packages/core/session/src/index.tsadoptSessionEventsnapshotSessionEventassertMessageEventShape
容错广播packages/core/session/src/index.tsinvokeContainedSessionObserverscollectSessionCallbacks
三个增量 foldpackages/core/session/src/index.tsderiveMessagesrequestHeaderrequestContext
存储与生命周期packages/core/session/src/index.tsSessionStore.prepare.enter.announce.flush.forkSessionForkError
事件词表与信封packages/core/session/src/types.tsSessionEventMapSessionEventSurfaceOpSurfaceIntentSESSION_FORMAT_VERSION
可见面投影packages/core/session/src/surface.tsderiveEventMessagefoldSurfaceSurfaceManager
可见面类型守卫packages/core/session/src/surface.tsisSurfaceEventisAppendSurfaceEventisReplacementSurfaceEvent
replace 校验packages/core/session/src/surface.tsassertProvenancereplacementRangeassertToolResultRewrite
无损 JSON 边界packages/core/session/src/json.tssnapshotJsonValueisJsonValue
请求头折叠packages/core/session/src/request-header.tsfoldRequestHeadercanonicalHeaderheaderEquals
崩溃修复packages/core/session/src/repair.tsinterruptedTurnClosersTOOL_OUTCOME_UNKNOWNTOOL_NOT_STARTED
chunk 打包packages/core/session/src/chunk-rows.tspackChunkRunsdecodeStorageRecordChunkRow
事件词表清单(生成)packages/core/session/src/known-event-types.tsKNOWN_SESSION_EVENT_TYPES
未发布所有权packages/core/session/src/preparation.tsSessionPreparation
日志关系不变量packages/core/session/src/invariant.tsvalidateEventapplyTransition
请求重建不变量packages/core/agent-loop/src/invariant.tsinstallllm/stream prepend 监听器)
持久化协调packages/session/session-persistence/src/coordinator.tsassertVersionassertEventsSupportedinstallWritePath
有界批写packages/session/session-persistence/src/write-behind.tsSessionWriteBehind
jsonl 格式packages/session/session-persistence-jsonl/src/format.tsHeaderLineeventLinesSessionLogScannerparseHeaderMeta
sqlite 格式packages/session/session-persistence-sqlite/src/schema.tsSCHEMA_VERSIONSessionRowEventRowscanRows
派生视图缓存packages/session/session-projection/src/index.tsProjectionDefinitionSessionProjectionRegistry
可见面替换的使用方packages/compaction/compaction-basic/src/region.tspackages/compaction/compaction-tool-result-pruner/src/index.tssurfaceOp: { op: 'replace', … } 的两个 append 点

接着读: 日志之上是怎么被驱动起来的——turn / step 状态机、Inbox、一次模型请求的生死,见第 3 章 · 主循环。日志之下的插件树与配置组装,见第 1 章 · 底座。工具执行如何产出 tool/calltool/result,见第 4 章 · 手脚