04 — 会话持久化 (pi-protocol 与 session-backends)

问题

Agent 的对话状态（消息、工具结果、用量统计）需要持久化，以支持：

• 进程重启后恢复对话
• 创建会话分支（fork）
• 导航到历史中的某一点

同时，Agent 可能以 client-server 模式运行，需要一个传输中立的协议来在进程间通信。

Pi 将这两个关注点分别放在 pi-protocol（协议层）和 pi-session-backends（存储层）。

第一部分：pi-protocol — 传输中立协议

packages/protocol/src/

协议概述

一个二进制 RPC 协议，用于远程 pi 会话。客户端发送 hello + 请求/取消；服务端回复 hello + 响应 + 服务更新 + 附件通知。

协议版本：PROTOCOL_VERSION = 8（protocol.ts:5）

消息类型

packages/protocol/src/protocol.ts

客户端消息（ClientMessage，line 63）：

消息	说明
ClientHello	{ type: "hello", version }— 必须是客户端发送的第一帧
RequestEnvelope	{ type: "request", id, target, call }— RPC 调用
CancelEnvelope	{ type: "cancel", id, target }— 取消请求

服务端消息（ServerMessage，line 110）：

消息	说明
ServerHello	{ type: "hello", version: 8, serverId }
ServerHelloError	握手失败
ResponseEnvelope	{ type: "response", id, ok, result?/error? }— RPC 响应
ServiceEventEnvelope	{ type: "service_update", subscriptionId, update }— 订阅更新
AttachmentEnvelope	{ type: "attachment", attachment }— 附件路由更新

RPC 目标

RpcTarget（line 47）定义调用目标：

// 服务端级调用
ServerTarget: { serverId: ServerId }

// 会话级调用（绑定到具体会话和附件）
SessionTarget: { serverId, sessionId, attachmentId }

call 和 result 字段是 JsonValue（不透明 JSON 值）——协议层不关心这些字段的具体内容，由上层序列化 agent-core 操作。

关键设计：所有 schema 使用 StrictObject（additionalProperties: false），拒绝未知字段。这保证了协议演化的严格性。

帧封装

packages/protocol/src/framing.ts

4 字节大端长度前缀 + CBOR 载荷：

┌─────────────┬───────────────────────┐
│ 4 bytes     │ payload (CBOR)        │
│ length (BE) │                       │
└─────────────┴───────────────────────┘

• DEFAULT_MAX_FRAME_LENGTH = 16 MiB（line 6）
• FrameDecoder（line 44）：流式增量解码器，处理部分 header、部分 payload（64 KiB 块）、一 chunk 多帧

CBOR 编码

packages/protocol/src/cbor/

RFC 8949 的严格子集：

• 不允许：不定长项、tag、break 标记、循环引用、undefined、非字符串 map key
• 支持类型：null、bool、整数（安全整数范围）、float64、UTF-8 字符串、byte string、数组、map

限制（options.ts）：

• DEFAULT_MAX_CBOR_BYTE_LENGTH = 16 MiB
• DEFAULT_MAX_CBOR_CONTAINER_LENGTH = 1,000,000
• DEFAULT_MAX_CBOR_DEPTH = 64

为什么用 CBOR 而非 JSON：CBOR 是二进制格式，更紧凑、解析更快。对于可能包含大量消息的会话数据，二进制编码显著减少传输量和存储空间。

Codec

packages/protocol/src/codec.ts：

// 编码：验证 → CBOR 编码 → 帧封装
encodeClientMessage(message) → Uint8Array
encodeServerMessage(message) → Uint8Array

// 解码：帧解码 → CBOR 解码 → schema 验证
ClientMessageDecoder.push(chunk) → ClientMessage[]
ServerMessageDecoder.push(chunk) → ServerMessage[]

isSupportedProtocolVersion(version)（line 139）：精确匹配 PROTOCOL_VERSION，不支持跨版本通信。

传输中立

协议层定义了消息格式和编码，但不定义传输方式。server 和 client 包可以选择 WebSocket、stdio、Unix socket 等传输实现。

第二部分：pi-session-backends — SQLite 会话存储

packages/session-backends/sqlite-node/src/

SQLite 能力抽象

types.ts:16-31：

interface SqliteDatabase {
  exec(sql: string): void;
  prepare(sql: string): SqliteStatement;
  transaction<T>(callback: () => T): T;  // 同步写事务
  close(): void;
}

interface SqliteDatabaseFactory {
  open(path): SqliteDatabase;       // 创建或打开
  openExisting(path): SqliteDatabase;
  openReadOnly(path): SqliteDatabase;
}

index.ts 基于 Node 内置的 node:sqlite``DatabaseSync 实现。事务使用 BEGIN IMMEDIATE / COMMIT / ROLLBACK。

关键设计：抽象层让后端可以替换 SQLite 实现（如 better-sqlite3）而不修改会话逻辑。

SQL 辅助

sql.ts：sql 标签模板构建参数化查询，嵌套 SqlQuery 可内联组合，joinSqlFragments 用于 IN (...) 和动态 WHERE。

数据库 Schema

migrations/001_initial.sql — 存储格式版本 4 / storageVersion 1

所有表 WITHOUT ROWID，按 session_id 分区：

权威数据表

表	主键	说明
sessions	id	会话元数据：创建时间、父会话、存储版本、消息计数、用量、next_seq
entries	(session_id, id)	Entry 树：parent_id、seq、type、timestamp、payload(JSON)
scalar_values	(session_id, namespace, key)	标量值：seq、value(JSON)
list_values	(session_id, namespace, key, seq)	列表值：value(JSON)
usage_ledger	(session_id, id)	用量账本：seq、entry_id、adjustment、usage(JSON)

投影缓存表

表	主键	说明
branch_entries	(session_id, branch_id, entry_id)	分支索引：entry_seq、entry_type
branch_meta	(session_id, branch_id)	分支元数据：tip_entry_id、tip_seq、base_branch_id、base_seq

权威 vs 投影

schema 注释明确说明：

“Authoritative durable state is entries + scalarvalues + list_values + usage_ledger; branch* and stats columns on sessions are maintained projections/caches.”

为什么这样设计：

• 分支遍历需要按顺序读取大量 entry。如果每次从 entries 的 parent_id 链遍历，性能很差
• branch_entries 预计算了分支的线性顺序，加速查询
• 但 branch_entries 是可重建的缓存——即使损坏，也能从 entries 重建

完整性触发器

migrations/001_initial.sql:69-91：

• trg_entries_validate：如果 parent_id 不存在或 id 与 usage_ledger 冲突则中止
• trg_usage_ledger_validate：如果 id 与 entries 冲突则中止

这保证了 entry 和 usage 共享 ID 命名空间，且 parent 必须先于 child 插入——在数据库层面强制数据完整性。

Storage 实现

packages/session-backends/sqlite-node/src/storage.ts

SqliteStorage implements Storage（line 49）：

commit — 核心写入路径

async commit(writes, context): Promise<CommitResult> {
  // 1. 通过 commitQueue 串行化
  // 2. prepareStorageCommit() 分配序列号
  // 3. 在事务中执行写入：
  //    - entry → EntryRowWriter.insert + appendEntryToBranchIndex
  //    - usage → UsageLedgerRowWriter.insert + addUsageToSessionStats
  //    - value → setScalarValueRow / deleteScalarValueRow
  //    - list → appendListValueRow / deleteListValueRows
  // 4. advanceNextSeq
  // 5. 返回 { ...result, stats }
}

commitQueue 是一个 promise chain，确保提交串行化——避免并发写入导致序列号冲突。

读取方法

方法	说明
getEntries(ids)	批量读取 entries
getValue(address)	读取标量值
scanValues(prefix)	前缀扫描标量值
readList(address, options)	分页读取列表值
scanBranch(query)	扫描分支 entries（使用投影缓存）
scanEntries(query)	扫描 entries（权威数据）
scanUsage(query)	扫描用量账本
getStats()	读取会话统计

snapshot — 用于 fork

async snapshot(options, context): Promise<SqliteStorageSnapshot> {
  // 1. 等待 commit 队列完成
  // 2. 读取所有标量值和 entries
  //    - 树范围：所有 entries
  //    - 分支范围：从分支 tip 开始，尊重 compaction 边界
  // 3. 返回 { entries, scalarValues, entriesComplete }
}

Session Repo

packages/session-backends/sqlite-node/src/repo.ts

SqliteSessionRepo（line 157）管理会话容器：

方法	说明
create(options, context)	生成 uuidv7 ID，创建 SQLite 文件（PRAGMA journal_mode=WAL; busy_timeout=5000），应用 schema，插入 session 行
open(metadata, context)	打开现有数据库（读写），验证元数据
list(options, context)	列出所有*.sqlite 文件，best-effort 跳过损坏文件
delete(metadata, context)	删除行（共享 DB）或删除文件（独立文件）
fork(source, options, context)	从源会话快照创建新会话
close(context)	幂等关闭所有打开的会话

文件命名：安全 ID 使用 ${id}.sqlite，不安全字符用 base64url 编码加 ~ 前缀。

分支索引维护

packages/session-backends/sqlite-node/src/session/branch-entries.ts

appendEntryToBranchIndex()（line 166）：

• 根 entry：创建根分支
• 延伸分支 tip 的 entry：追加到当前分支
• parent 不是 tip 的 entry：创建分叉分支（createDivergentBranchForEntry），从 parent 分支复制片段直到最近的 compaction 边界

这是分支索引投影的维护逻辑——当用户在历史中某点创建新分支时，需要正确地将新 entry 关联到正确的分支。

Fork 机制

repo.ts:312 — fork()：

1. 如果源会话当前在 repo 中打开：使用内存中的 snapshot()
2. 否则：以只读方式打开源会话，外部构建 fork 快照
3. buildForkSnapshot() 使用 agent-core 的 createForkSnapshot()，可选树范围或分支范围
4. 在一个事务中写入所有 entries、标量值和更新后的统计

Entry 解码

session/entries.ts：

• entryPayload(entry)（line 28）：剥离身份字段，JSON 序列化剩余部分
• decodeEntryRow()（line 103）：从数据库行重建类型化 Entry
• scanEntryRows()（line 150）：构建动态 WHERE/ORDER BY/LIMIT

协议层与存储层的连接

┌─────────────┐     ┌──────────────┐     ┌─────────────────┐
│   client    │────▶│   protocol   │────▶│     server      │
│             │     │  (CBOR+帧)   │     │  SessionRouter  │
└─────────────┘     └──────────────┘     └────────┬────────┘
                                                   │
                                          ┌────────▼────────┐
                                          │  session-       │
                                          │  backends       │
                                          │  (SQLite)       │
                                          └────────┬────────┘
                                                   │
                                          ┌────────▼────────┐
                                          │  agent-core     │
                                          │  (Session/      │
                                          │   Storage 接口)  │
                                          └─────────────────┘

• pi-protocol 定义”如何传输”（消息格式、编码）
• pi-session-backends 定义”如何存储”（SQLite schema、读写逻辑）
• agent-core 定义”存储什么”（Entry 类型、Session/Storage 接口）
• server/client 定义”如何连接”（传输实现）

学习要点

• 权威数据 vs 投影缓存的分离是数据库设计的经典模式：entries 是事实来源，branch_entries 是为查询性能优化的缓存，可随时重建
• 触发器保证完整性：在数据库层面（而非应用层面）强制 parent 先于 child 插入，避免应用 bug 导致数据不一致
• CBOR 比 JSON 更适合二进制协议：更紧凑、解析更快，但需要自定义编解码器
• 传输中立协议：协议层只定义消息格式，不关心传输方式，让同一套协议可用于多种场景
• commit 串行化：通过 promise chain 保证写入串行化，避免序列号冲突，比数据库锁更灵活
• snapshot + 事务写入实现 fork：先读取源会话快照，然后在一个事务中写入新会话，保证 fork 的原子性

上一篇：03-Agent 运行时核心-pi-agent-core
下一篇：05-编码 Agent-pi-coding-agent