第九章：持久化与数据平面

本章目标：帮助你理解 Session 生命周期、日志格式与版本管理、持久化后端、查询、Compaction 和 Python SDK 对接。阅读本章后，你应该能回答"会话数据是如何持久化的"以及"如何扩展持久化后端"。

9.1 Session 生命周期

9.1.1 概念

Session 是一次 Agent 与用户的完整交互。它包含：

SessionHeader：元数据
SessionEventLog：仅追加的事件流
内存状态：运行时投影

stateDiagram
    [*] --> Created: 创建 Session
    Created --> Active: 开始交互
    Active --> Active: 追加事件
    Active --> Paused: 暂停
    Paused --> Active: 恢复
    Active --> Completed: 交互完成
    Active --> Failed: 交互失败
    Completed --> Archived: 归档
    Failed --> Archived: 归档

stateDiagram
    [*] --> Created: 创建 Session
    Created --> Active: 开始交互
    Active --> Active: 追加事件
    Active --> Paused: 暂停
    Paused --> Active: 恢复
    Active --> Completed: 交互完成
    Active --> Failed: 交互失败
    Completed --> Archived: 归档
    Failed --> Archived: 归档

9.1.2 Session 创建

// 来自 packages/core/session/src/index.ts
class Session {
  constructor(options: CreateSessionOptions) {
    this.id = options.id ?? generateSessionId()
    this.header = {
      id: this.id,
      createdAt: new Date(),
      cwd: options.cwd,
      // ... 其他元数据
    }
    this.log = new SessionEventLog()
  }
}

9.1.3 SessionHeader

// 来自 packages/core/session/src/types.ts
interface SessionHeader {
  /** 会话 ID */
  id: SessionId
  /** 创建时间 */
  createdAt: Date
  /** 工作目录 */
  cwd?: string
  /** 父会话 ID（fork 来源）*/
  parentSession?: SessionId
  /** 是否为种子会话 */
  isSeeded?: boolean
  /** 来源 */
  origin?: 'subagent'
  /** 委托深度 */
  delegationDepth?: number
  /** Agent 预设 */
  agentPreset?: string
}

9.2 日志格式

9.2.1 仅追加日志

Session 日志是仅追加的事件流。这是 dsh 的核心设计原则之一：

Model-visible ⟺ logged——任何到达模型请求的内容都必须可从日志重建。

9.2.2 SessionEvent 类型

// 来自 packages/core/session/src/types.ts
type SessionEvent =
  | TurnStartEvent
  | TurnEndEvent
  | StepStartEvent
  | StepEndEvent
  | UserMessageEvent
  | AssistantMessageEvent
  | AssistantChunkEvent
  | ToolCallEvent
  | ToolResultEvent
  | RequestHeaderEvent
  // ... 更多事件类型

9.2.3 事件示例

// 用户消息事件
interface UserMessageEvent {
  type: 'user/message'
  seq: SessionSeq
  data: {
    content: string
    // ... 其他字段
  }
}

// 助手消息事件
interface AssistantMessageEvent {
  type: 'assistant/message'
  seq: SessionSeq
  data: {
    content: string
    toolCalls?: ToolCall[]
    // ... 其他字段
  }
}

// 工具调用事件
interface ToolCallEvent {
  type: 'tool/call'
  seq: SessionSeq
  data: {
    name: string
    parameters: Record<string, unknown>
    // ... 其他字段
  }
}

9.2.4 SessionSeq

// 来自 packages/session/session-persistence/src/index.ts
type SessionSeq = number & { __brand: 'SessionSeq' }

function SessionSeq(value: number): SessionSeq {
  return value as SessionSeq
}

每个事件都有一个单调递增的序列号（seq），用于排序和投影。

9.3 持久化后端

9.3.1 架构

graph TB
    subgraph "持久化能力"
        A["session-persistence/ (Service Definition)"]
        B["session-persistence-jsonl/ (Provider)"]
        C["session-persistence-sqlite/ (Provider)"]
        D["session-persistence-memory/ (Provider)"]
    end

    A --> B
    A --> C
    A --> D

graph TB
    subgraph "持久化能力"
        A["session-persistence/ (Service Definition)"]
        B["session-persistence-jsonl/ (Provider)"]
        C["session-persistence-sqlite/ (Provider)"]
        D["session-persistence-memory/ (Provider)"]
    end

    A --> B
    A --> C
    A --> D

9.3.2 SessionPersistence 接口

// 来自 packages/session/session-persistence/src/index.ts
interface SessionPersistence {
  /** 创建新会话 */
  create(options: CreateSessionOptions): Promise<SessionHandle>
  
  /** 打开现有会话 */
  open(sessionId: SessionId): Promise<SessionHandle>
  
  /** 统计会话 */
  stat(sessionId: SessionId): Promise<SessionStat | null>
  
  /** 列出所有会话 */
  list(options?: ListOptions): Promise<SessionHeader[]>
  
  /** 导出会话 */
  export(sessionId: SessionId): Promise<ExportedSession>
}

9.3.3 SessionHandle

interface SessionHandle {
  /** 会话 ID */
  readonly id: SessionId
  
  /** 读取事件 */
  read(offset?: SessionLogOffset): AsyncIterable<SessionEvent>
  
  /** 追加事件 */
  append(events: SessionEvent[]): Promise<void>
  
  /** 获取头部 */
  header(): SessionHeader
  
  /** 关闭句柄 */
  close(): Promise<void>
}

9.3.4 JSONL 后端

// 来自 packages/session/session-persistence-jsonl/src/storage.ts
class JsonlSessionHandle implements SessionHandle {
  private logFile: string
  
  constructor(
    private sessionId: SessionId,
    private dir: string
  ) {
    this.logFile = join(dir, `${sessionId}.jsonl`)
  }
  
  async append(events: SessionEvent[]): Promise<void> {
    const lines = events.map(e => JSON.stringify(e)).join('\n') + '\n'
    await appendFile(this.logFile, lines)
  }
  
  async *read(offset?: SessionLogOffset): AsyncIterable<SessionEvent> {
    const content = await readFile(this.logFile, 'utf-8')
    const lines = content.split('\n').filter(Boolean)
    
    for (const line of lines) {
      const event = JSON.parse(line) as SessionEvent
      if (offset === undefined || event.seq > offset) {
        yield event
      }
    }
  }
}

9.4 日志版本管理

9.4.1 SESSION_FORMAT_VERSION

dsh 使用 SESSION_FORMAT_VERSION 来管理日志格式的兼容性：

// 来自 packages/session/session-persistence/src/index.ts
const SESSION_FORMAT_VERSION = 0

当前版本为 0，表示没有兼容性承诺。每次格式变更都会递增版本号。

9.4.2 版本检查

// 概念性描述
function validateSessionLog(log: SessionEventLog): boolean {
  // 检查日志格式版本
  if (log.version !== SESSION_FORMAT_VERSION) {
    throw new Error(`Incompatible session log version: ${log.version}`)
  }
  
  // 验证事件序列
  // ...
  
  return true
}

9.5 Session 投影

9.5.1 概念

Session 投影从日志增量派生状态。这是 dsh 的数据平面核心。

9.5.2 ProjectionDefinition

// 来自 packages/session/session-projection/src/index.ts
interface ProjectionDefinition<K extends string, S> {
  /** 投影键 */
  key: K
  /** 状态版本 */
  stateVersion: number
  /** 状态 schema */
  stateSchema: ZodType<S>
  /** 初始状态 */
  init(): S
  /** 应用事件 */
  apply(state: S, event: SessionEvent): S
}

9.5.3 turnBoundary 投影

// 来自 packages/core/agent-loop/src/index.ts
const turnBoundaryProjectionDefinition = {
  key: 'turnBoundary',
  stateVersion: 2,
  stateSchema: zod.object({
    openTurnStartSeq: zod.number().nullable(),
    lastStepStartSeq: zod.number().nullable(),
    lastStepBoundary: zod.object({
      kind: zod.union([zod.literal('start'), zod.literal('end')]),
      seq: zod.number()
    }).nullable(),
    lastTurn: zod.number()
  }),
  init: () => ({
    openTurnStartSeq: null,
    lastStepStartSeq: null,
    lastStepBoundary: null,
    lastTurn: 0
  }),
  apply: (state, event) => {
    switch (event.type) {
      case 'turn/start':
        return { ...state, openTurnStartSeq: event.seq, lastTurn: event.data.turn }
      case 'turn/end':
        return { ...state, openTurnStartSeq: null }
      case 'step/start':
        return { ...state, lastStepStartSeq: event.seq, lastStepBoundary: { kind: 'start', seq: event.seq } }
      case 'step/end':
        return { ...state, lastStepBoundary: { kind: 'end', seq: event.seq } }
      default:
        return state
    }
  }
}

9.5.4 使用投影

// 获取当前状态
const state = stateOf('turnBoundary')

// 批量快照
const snapshot = snapshot(['turnBoundary', 'otherProjection'])

9.6 Compaction（压缩）

9.6.1 概念

Compaction 机制在上下文窗口溢出时压缩历史消息：

9.6.2 CompactionEngine

// 来自 packages/compaction/compaction/src/index.ts
class CompactionEngine {
  constructor(private ctx: Context) {}
  
  /** 检查是否需要压缩 */
  async needsCompaction(session: Session): Promise<boolean> {
    // 计算当前上下文大小
    const contextSize = this.calculateContextSize(session)
    const maxSize = this.getMaxContextSize()
    
    return contextSize > maxSize
  }
  
  /** 执行压缩 */
  async compact(session: Session): Promise<CompactionResult> {
    // 1. 选择要压缩的消息
    const messagesToCompact = this.selectMessages(session)
    
    // 2. 生成摘要
    const summary = await this.generateSummary(messagesToCompact)
    
    // 3. 替换原始消息
    const compactedSession = this.replaceMessages(session, messagesToCompact, summary)
    
    return { session: compactedSession, summary }
  }
}

9.6.3 Compaction 策略

interface CompactionStrategy {
  /** 选择要压缩的消息 */
  selectMessages(session: Session): SessionEvent[]
  
  /** 生成摘要 */
  generateSummary(messages: SessionEvent[]): Promise<string>
  
  /** 替换消息 */
  replaceMessages(
    session: Session,
    original: SessionEvent[],
    summary: string
  ): Session
}

9.7 Session 查询

9.7.1 架构

graph TB
    subgraph "session-query/ 能力"
        A["session-query/ (Service Definition)"]
        B["session-query-local/ (Provider)"]
        C["tool-session-query/ (Consumer)"]
    end

    A --> B
    C --> A

graph TB
    subgraph "session-query/ 能力"
        A["session-query/ (Service Definition)"]
        B["session-query-local/ (Provider)"]
        C["tool-session-query/ (Consumer)"]
    end

    A --> B
    C --> A

9.7.2 查询接口

// 来自 packages/session-query/session-query/src/index.ts
interface SessionQueryService {
  /** 列出所有会话 */
  list(options?: ListOptions): Promise<SessionHeader[]>
  
  /** 获取会话摘要 */
  inspect(sessionId: SessionId): Promise<SessionSnapshot>
  
  /** 搜索会话 */
  search(query: string): Promise<SessionSearchResult[]>
  
  /** 获取会话日志 */
  readLog(sessionId: SessionId, options?: ReadOptions): Promise<SessionEvent[]>
}

9.7.3 语义搜索

// 来自 packages/session-query/session-query/src/corpus.ts
class SessionCorpus {
  /** 语义搜索 */
  async semanticSearch(query: string): Promise<SearchResult[]> {
    // 1. 生成查询嵌入
    const queryEmbedding = await this.embed(query)
    
    // 2. 搜索相似会话
    const results = await this.vectorStore.search(queryEmbedding, {
      limit: 10,
      threshold: 0.7
    })
    
    return results
  }
}

9.8 Python SDK 对接

9.8.1 架构

9.8.2 Python SDK 使用

from deepseek_harness import HarnessClient

# 创建客户端
client = HarnessClient(profile='sdk')

# 创建会话
session = client.create_session()

# 发送消息
response = client.send_message(session.id, 'Hello, world!')

# 获取响应
print(response.content)

# 列出所有会话
sessions = client.list_sessions()

9.8.3 SDK 服务器

// 来自 packages/sdk/server/src/server.ts
class HarnessSdkJsonRpcServer {
  constructor(private ctx: Context) {}
  
  /** 初始化服务器 */
  async initialize(): Promise<void> {
    // 1. 注册 JSON-RPC 方法
    this.registerMethods()
    
    // 2. 启动服务器
    await this.startServer()
  }
  
  /** 注册 JSON-RPC 方法 */
  private registerMethods(): void {
    this.server.method('session.create', this.createSession.bind(this))
    this.server.method('session.list', this.listSessions.bind(this))
    this.server.method('session.send', this.sendMessage.bind(this))
    // ... 更多方法
  }
}

9.9 遥测

9.9.1 架构

9.9.2 遥测事件

// 来自 packages/session/session-telemetry-otel/src/index.ts
class OpenTelemetryTelemetryProvider {
  /** 记录会话事件 */
  recordEvent(event: SessionEvent): void {
    const span = this.tracer.startSpan('session.event', {
      attributes: {
        'session.id': event.sessionId,
        'event.type': event.type,
        'event.seq': event.seq
      }
    })
    
    span.end()
  }
  
  /** 记录 LLM 调用 */
  recordLlmCall(call: LlmCall): void {
    const span = this.tracer.startSpan('llm.call', {
      attributes: {
        'llm.model': call.model,
        'llm.tokens.input': call.inputTokens,
        'llm.tokens.output': call.outputTokens,
        'llm.duration': call.duration
      }
    })
    
    span.end()
  }
}

9.10 小结

概念	一句话解释
Session	一次 Agent 与用户的完整交互
SessionEventLog	仅追加的事件流
SessionHeader	会话元数据
SessionSeq	事件序列号
SessionPersistence	持久化接口
Projection	从日志增量派生状态
Compaction	上下文窗口溢出时压缩历史
SESSION_FORMAT_VERSION	日志格式版本管理

9.11 完整教程总结

恭喜你完成了 DeepSeek Harness 教程的全部九章！让我们回顾一下你学到的内容：

核心概念

总体架构：dsh 是全插件化的 Agent 运行时
环境与运行：Profile 系统、CLI 启动链、构建系统
Cordis 框架：Plugin/Service/Context 模型、事件派发
Agent 循环：Turn/Step 生命周期、Cancel 机制
LLM 能力：适配器机制、流式调用、重试策略
工具能力：三角色模式、执行管道、能力插件
Agent 编排：Subagent、Worker Thread、Workflow
人类协作：Approval、Questions、Guard、Plan
持久化：Session、日志、投影、Compaction

下一步

实践：尝试修改一个插件或添加新功能
深入源码：使用 GitNexus 探索更多细节
贡献：参与 dsh 的开发和改进

参考资源

本教程基于 DeepSeek Harness 源码编写，内容可能随项目更新而变化。

2026 年 9 月
日	一	二	三	四	五	六
		1	2	3	4	5
6	7	8	9	10	11	12
13	14	15	16	17	18	19
20	21	22	23	24	25	26
27	28	29	30