@deepseek-ai/dsh-session · packages/core/session

dsh-session — 事件溯源会话日志

append-only 的 SessionEvent 日志与内存 Session 存储,以及从日志投影出的派生 LLM 消息历史。持久化是插件关注点。它是"模型可见 ⟺ 已落盘"不变量在代码里的化身。

ctx key: ctx.sessions 10 个 src 文件 · 3300 行 Service 类插件(默认导出 SessionStore) 13 个 spec 测试

01包定位与入口

本包不是函数插件(没有 name/inject/Config/apply),而是 Service 包:默认导出 SessionStore 类(index.ts:1157),经 declare module '@deepseek-ai/cordis' { interface Context { sessions: SessionStore } } 挂到 ctx.sessions(index.ts:37-40)。

项	内容
设计口号	日志是唯一事实来源,模型历史是派生的;每个事件是无损 JSON,`seq` 连续;持久化后端可逐字存储规范日志
Session 本体	纯类,不是 Cordis Service(index.ts:425 `/** @typert object */`);SessionStore 才是 Service
子路径导出	`.`(根)、`./types`(仅类型,client-safe)、`./surface`(browser-safe,无 node: import)、`./invariant`(不变量伴侣插件)
Typert 注册	构造时注册 `session` lookup —— `parameter: 'session'`、`wire: 'sessionId'`(index.ts:89-93)
版本常量	`SESSION_FORMAT_VERSION = 0`(types.ts:56);写在每个 SessionHeader 里,持久化后端加载时强制校验

02核心接口

SessionHeader(types.ts:61-99)

不可变、已校验的存储元数据,不放在会话事件日志里:

字段	语义
`version`	创建时盖 `SESSION_FORMAT_VERSION` 戳;后端拒绝任何其他版本(无迁移)
`id`	`Branded<'SessionId'>`
`cwd? / parentSession? / seedLength?`	工作目录(fork 谱系:父会话 + 继承的前导事件数)
`origin? / delegationDepth? / agentPreset?`	子 agent 分类 / 委托深度 / 组成 agent 的 preset id(持久化因为 preset 决定工具与 prompt)

Session 类(index.ts:425-758)

readonly header: SessionHeader
readonly id: SessionId
readonly firstLiveSeq: number        // 本进程内首个 append 的 seq(= 构造 seed 长度)
readonly surface: SessionSurface        // SurfaceManager 只读视图
readonly events: readonly SessionEvent[]
readonly seq: number                    // 恒为 log.length —— 连续性契约
static create(id, seed?, header?)             // 快照模式
static fromRestore(id, seed, header)          // 恢复模式:取所有权、就地冻结
append<T extends SessionEventType>(type, data, ...opts)
requestHeader(): EpochHeader | undefined   // 增量 fold,冻结缓存
deriveMessages(): Message[]                // 从 surface 投影模型历史

SessionStore(Service,index.ts:792-1155)

create(id?, options?): Session
prepare(id?, options?): Session          // 构造但不发布
enter(session): () => void               // detach disposer(单次幂等)
announce(session): void                // 发 session/created;同步 throw 否决发布
flush(session): Promise<boolean>        // THE 持久性检查点入口
fork(source, boundary?, childSessionId?): Session
get(id): Session | undefined / list(): Session[]

◆ 版本契约 — SESSION_FORMAT_VERSION 是否 bump 由 WRITER 决定:恰好当"旧运行时对新日志无法完全语义正确地重建"时 bump。只有结构性变化才 bump(header 形状、信封、核心事件语义、surface 机制);新增普通事件类型不 bump —— per-event 的 ignorable 守卫覆盖词汇增长。拿不准就 bump。

03事件系统:SessionEventMap 与声明合并

SessionEventMap 是 merge-extensible 的拥有词汇表。插件用 declare module '@deepseek-ai/dsh-session/types' { interface SessionEventMap { ... } } 合并自己的事件(types.ts:335-336)。顶层接口只能有一个家(必须在本包);每个合并成员自动进入生成目录 KNOWN_SESSION_EVENT_TYPES(known-event-types.ts,共 44 个)。

核心 13 个事件

事件	类别	要点
`turn/start · turn/end`	生命周期	turn 打开/关闭;`turn/end.reason` 是 merge-extensible 联合(completed/aborted/blocked/error/max-tokens/interrupted)
`step/start · step/end`	生命周期	一次模型调用 + 其请求的工具执行
`user/message`	surface	模型可见用户消息;content 逐字投影,`source` 区分直接提示 / inject 上下文 / 目标延续
`assistant/chunk`	log-only	原始流块,词元级重放保真
`assistant/message`	surface	每步组装完成的助手消息 + usage
`tool/call`	log-only	模型请求一次工具调用;arguments 是原始 JSON 字符串(未解析)
`tool/result`	surface	模型面结果 + 内部失败身份 + 可选 `meta` 展示负载(须 JSON 可序列化)
`todo/write`	log-only	整表快照 last-write-wins,绝不进派生历史
`request/header · request/context`	log-only	请求头快照(重建请求)/ 路由元数据(不参与重建)
`session/end-seed`	生命周期	构造 seed 结束标记;只有 Session 构造器是合法写入者

信封与 ignorable(types.ts:404-435)

SessionEvent 是 type 上的真判别联合;所有字段 append 时深冻结。surfaceOp/sourceEventSeqs 是条件字段—— 仅存在于 3 个 surface 类型上,编译器在调用点强制(非 surface 类型传 opts 直接类型错误)。

◆ ignorable 的默认是 required — 标记 = "reader 不认识此类型时可以安全跳过"。缺席 = 必需:遇到未知必需事件必须拒绝重建而非静默丢弃。忘记标记导致过度拒绝(可接受),漏标导致静默读错(不可接受)。

服务事件(index.ts:42-86)

事件	mode	契约
`session/created`	emit	同步 throw 否决并回滚(配对 disposal);scoped 分发
`session/disposed`	emit	宣布过的会话离开 store 时恰好一次,含发布回滚
`session/event`	emit	post-commit、fire-and-forget;构造 seed 不发出
`session/flush`	parallel	被 await 的持久性检查点;每个监听器都运行,无 waterfall 否决

04关键机制深度解析

append 流水线(index.ts:604-655)

snapshotJsonValue(data)

▼ 无损 JSON 校验 + 单遍脱拷贝(热路径从不阻塞 I/O)

重入守卫 + deepFreeze 信封 {type, seq: log.length, time, data, surfaceMeta}

▼

surfaceManager.validateNext(event)

▼ plan-before-commit:验证但不提交

collectCallbacks → log.push(event)

▼ 事件入 log 即已提交;观察者失败不影响返回值

invokeContainedSessionObservers 逐监听器 try/catch,拒绝记 warn

返回的是入 log 的快照而非调用者的可变输入;坏事件在 append 点失败,而非后端 flush 时。种子(seed)与 append 走 相同的不变量 —— 这样 replay/fork 无法构造任何后端都无法存储的活日志。

deriveMessages():从日志投影模型历史(index.ts:726-747)

投影只走 surface 节点(3 种 surface 类型中带了 surfaceOp 者);chunk、turn/step 边界、usage、todo、request/* 等 log-only 事件不参与。投影规则(surface.ts:83-114):

user/message → data 逐字(不做 framing —— framing 由 producer 烘进 content)
assistant/message → data.message;空 content 的投影为 null(只为承载 max-tokens 的 usage)
tool/result → data.message;default → null,无 assertNever(merge-extensible 联合)

◆ Surface 机制 — SurfaceOp 只有 append 和 replace 两个变体。replace 必须覆盖每个被遮蔽节点、 tool/result 的 replace 只能改 content、只能重写一个节点(surface.ts:287-318)。缓存模型:surface 重写使 generation 变化 → 重建整个派生缓存; 平时只投影新节点。人工 transcript 必须投影 append-origin 事件(isAppendSurfaceEvent 守卫),不要直接读 session.surface(landed replace 会抹掉用户已见的对话)。

chunk-rows:流块打包(压缩而非数据损失)

流式逐词元 delta 会让日志行数爆炸(实测 ~56 倍)。packChunkRuns(chunk-rows.ts:192)把至少 3 个成员的连续 run 打包成 ChunkRow(text-chunks / reasoning-chunks / tool-call-chunks);文本绝不 join(词元边界是数据); 不认识的块原样存储。存储行是持久编码词汇、不是会话事件(从不进 Session.events)。读侧必须无条件 decodeStorageRecord,畸形行 fail loud。

request-header 重建(可重建请求)

request/header 携带 EpochHeader —— 日志中派生态之外的完整请求状态(config + adapterDefaults + system + tools)。 foldRequestHeader(events) 纯离线重建:按序取最后一个;活 Session 用同一 fold 增量维护(冻结缓存)。 request/context 不参与请求重建或 header 相等性。每次请求前 loop 记录 header;仅当 !headerEquals 时记 reason:'change'。

repair:崩溃恢复(repair.ts)

interruptedTurnClosers(events) 扫描有效提交前缀(可能带崩溃尾巴),为孤儿 turn 合成:

每个 pending tool call 一个合成 tool/result(surfaceOp: append)—— 已记录 call 用 TOOL_OUTCOME_UNKNOWN、未记录 start 用 TOOL_NOT_STARTED,确定性消息 id
合成 step/end(turn/end 在 step 开着时是 invariant 违规)
合成 turn/end 带 reason: { kind: 'interrupted' } —— loop 从不发出此 marker,只有持久化后端重载时用它关崩溃孤儿

seq 从 last.seq + 1 续,时间复用最后真实事件的 time(确定性,不发明"未来"时间)。修复只发生一次、由持久化侧触发。

invariant:关系不变量(invariant.ts)

SessionTrace(lastSeq / openTurn / openStep / nextTurn / nextStep / pendingCalls)逐事件先验后提交。核心断言: turn/step 必须匹配递增计数器;assistant/tool 事件必须落在开放 turn+step 内; tool/result 的 append 变体要求 pendingCalls 里有 callId;merge-extensible 事件的关系属于其拥有插件(default 放行)。安装采用两段式:经 internal/dispatch 在发布前暂存验证,session/event 在事件确实发布后提交转换 —— 保证"先验证后提交"。

05自定制插件指南

◆ 通用门槛 — 每个包必须拥有 ./invariant 伴侣插件注册清单名;事件 JSDoc 需要 @mode 与 payload @param; 模型可见输入必须配新会话事件。新增事件后运行 pnpm run gen-persistence-catalog 重新生成目录(否则 doc-sync 失败)。

(a) 往会话日志加一个新事件类型(模型可见输入)

声明合并:declare module '@deepseek-ai/dsh-session/types' { interface SessionEventMap { 'your/event': Payload } } —— 不得创建顶层接口
append:session.append('your/event', data);payload 必须无损 JSON
surface 抉择:若产生 LLM 消息 → 必须是三种 surface 类型之一并传 SurfaceIntent(SurfaceEventType 集合本身不可扩展); 纯 log-only → 不带 surface 元数据;丢失不影响重建则设 ignorable: true
注册目录:重跑 gen-persistence-catalog;事件进入 docs/persistence-catalog.md
关系不变量:事件与 turn/step 的关系在你的包自己的 ./invariant 中注册(会话包对未知类型 default 放行)
测试:包级单测 + 关键路径无 key 快照(装配应用 transcript)

(b) 替换持久化后端(JSONL/SQLite 之外)

实现 PersistenceBackend 接口(coordinator.ts:128-209):name / loadStored(返回 header + 有效连续前缀 + torn 标记,返回的图必须全新、无别名)/ readStoredRevision(源限定 revision) / appendBatch(物化写与首批事件必须原子提交) / commitRepair / list;SQLite 类可寻址后端可加 loadStoredFrom 做 seek 读
挂接生命周期:订阅 session/created / session/event(写后缓冲)/ session/flush(持久性屏障)/ session/disposed(最终 drain)
加载校验链:header version === SESSION_FORMAT_VERSION(方向感知拒绝:newer → "upgrade",older → "no upgrade path");seq 从 0 连续; 未知类型无 ignorable → 拒绝(SessionFormatUnsupportedError);崩溃修复在 commitRepair 前用 interruptedTurnClosers 生成 closers
契约:后端拒绝旧版本格式(无迁移);Session.fromRestore 所有权转移 —— 调用方不得保留可变别名;测试含损坏/torn 行拒绝、原子性、修复后合法重放

(c) fork / resume 会话

fork:ctx.sessions.fork(source, boundary?, childSessionId?) —— boundary 包含式、必须连续、前缀不得以 turn/start 结尾(OPEN_TURN); 子会话继承 cwd,记 parentSession + seedLength(index.ts:1087-1094)
resume:后端 loadStored → prepare(id, { seed, meta, seedSource: 'persistence' }) → Session.fromRestore 就地校验冻结; 构造器自动 append session/end-seed;firstLiveSeq(本进程构造事实)≠ header.seedLength(持久化 fork 边界)

(d) 监听会话事件做 UI / 遥测

ctx.on('session/event', (session, event) => …):post-commit、fire-and-forget、逐监听器包含;观察者失败只记 warn
session/created 监听器不要 throw(会否决发布);持久性屏障用 session/flush
所有事件 scoped 分发 —— agent-scoped 监听器只收到该 agent 上下文进入的会话
种子不发布:telemetry 采纳必须从 session.events 的 firstLiveSeq 起重放
UI transcript 用 isAppendSurfaceEvent 守卫投影 append-origin 事件

06关键陷阱与约定

seq = log.length 连续性契约:seed 校验逐条强制 snapshot.seq === index;chunk 解码也以此重构 seq
事件深冻结 + 快照缓存:events getter 的数组 append 前复用、之后失效;cast 也无法改写持久历史
session/end-seed 只有构造器写 —— 插件 append 一个会把其前的活括号静默分类为 seed 历史
append 不可重入:附加会话的 append 在另一个 append 发布期间被拒(index.ts:623-626)
flush 唯一入口是 ctx.sessions.flush(session),别裸 dispatch session/flush(store 拥有 carrier);返回是否至少一个持久化监听器参与
loop 不在 turn 边界等 flush;dsh-session-checkpoint-policy 在每模型请求 dispatch 前、顶层工具执行前、每步前 flush,失败 fail-closed
所有权区分:snapshotSessionEvent(借阅,先 clone)vs adoptSessionEvent(排他,就地);meta 必须是 plain JSON record
surface.ts 必须 browser-safe(无 node: import);types.ts 只含类型,./types 子路径是 client-safe 面
HMR:invariant 与持久化协调器都通过 ctx.sessions.list() 播种已存在活会话(热重载不重放 session/created)