第十一章：Session 日志深度解析

本章目标：帮助你理解 JSONL 格式的逐行结构、事件序列号（SessionSeq）的单调递增保证、投影（Projection）的增量派生算法、以及日志重放（Replay）与 Fork 机制。阅读本章后，你应该能回答"Session 日志在磁盘上长什么样"以及"投影是如何从日志增量计算的"。

11.1 JSONL 格式

11.1.1 什么是 JSONL

JSONL（JSON Lines）是 dsh 选择的会话日志格式。每行是一个独立的 JSON 对象，对应一个 SessionEvent。

{"type":"turn/start","seq":1,"data":{"turn":1}}
{"type":"user/message","seq":2,"data":{"content":"Hello"}}
{"type":"assistant/message","seq":3,"data":{"content":"Hi there!"}}
{"type":"turn/end","seq":4,"data":{"reason":{"kind":"completed"}}}

11.1.2 为什么选择 JSONL

特性	JSONL	SQLite	纯 JSON
追加写入	✅ O(1)	⚠️ 需要事务	❌ 需要重写整个文件
流式读取	✅ 逐行	✅ 游标	❌ 需要完整解析
人类可读	✅	❌	✅
无依赖	✅	❌ 需要 sqlite3	✅
并发安全	✅ 行级	✅ 事务级	❌
压缩友好	✅	❌	✅

11.1.3 文件结构

$dsh-home/sessions/
├── <session-id>.jsonl      # 会话日志
├── <session-id>.meta.json  # 会话元数据
└── index.json              # 会话索引

11.2 SessionEvent 详解

11.2.1 事件类型全景

graph TB
    subgraph "Turn 事件"
        A[turn/start]
        B[turn/end]
    end

    subgraph "Step 事件"
        C[step/start]
        D[step/end]
    end

    subgraph "消息事件"
        E[user/message]
        F[assistant/message]
        G[assistant/chunk]
    end

    subgraph "工具事件"
        H[tool/call]
        I[tool/result]
    end

    subgraph "请求事件"
        J[request/header]
    end

11.2.2 事件类型定义

// 来自 packages/core/session/src/types.ts（概念性）

// Turn 开始事件
interface TurnStartEvent {
  type: 'turn/start'
  seq: SessionSeq
  data: {
    turn: number        // Turn 编号
    source: string      // 触发源
  }
}

// 用户消息事件
interface UserMessageEvent {
  type: 'user/message'
  seq: SessionSeq
  data: {
    content: string     // 消息内容
    role: 'user'
  }
}

// 助手消息事件
interface AssistantMessageEvent {
  type: 'assistant/message'
  seq: SessionSeq
  data: {
    content: string     // 回复内容
    toolCalls?: ToolCall[]  // 工具调用
    role: 'assistant'
  }
}

// 工具调用事件
interface ToolCallEvent {
  type: 'tool/call'
  seq: SessionSeq
  data: {
    name: string        // 工具名称
    parameters: Record<string, unknown>  // 参数
    callId: string      // 调用 ID
  }
}

// 工具结果事件
interface ToolResultEvent {
  type: 'tool/result'
  seq: SessionSeq
  data: {
    callId: string      // 对应的调用 ID
    result: unknown     // 结果
    error?: string      // 错误信息
  }
}

11.3 SessionSeq：事件序列号

11.3.1 单调递增保证

SessionSeq 是一个品牌化数字类型，保证单调递增：

type SessionSeq = number & { __brand: 'SessionSeq' }

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

11.3.2 序列号分配

// 概念性描述
class SessionEventLog {
  private nextSeq: SessionSeq = SessionSeq(1)
  
  /** 追加事件 */
  append(event: Omit<SessionEvent, 'seq'>): SessionEvent {
    const seq = this.nextSeq++
    const fullEvent = { ...event, seq } as SessionEvent
    
    // 写入文件
    this.writeFile(JSON.stringify(fullEvent) + '\n')
    
    return fullEvent
  }
}

11.3.3 序列号的用途

排序：事件按 seq 排序
投影：投影使用 seq 判断是否已处理
Fork：Fork 时记录父会话的 seq 偏移
查询：按 seq 范围查询事件

11.4 投影（Projection）

11.4.1 概念

投影从日志增量派生状态。每次新事件到达时，投影更新其状态，而不需要重新计算整个日志。

11.4.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
}

11.4.3 投影实现示例

// turnBoundary 投影
const turnBoundaryProjection: ProjectionDefinition<'turnBoundary', TurnBoundaryState> = {
  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
    }
  }
}

11.4.4 投影运行时

// 来自 packages/session/session-projection/src/runtime.ts
class ProjectionRuntime {
  private projections = new Map<string, ProjectionInstance>()
  
  /** 注册投影 */
  register<K extends string, S>(definition: ProjectionDefinition<K, S>): void {
    this.projections.set(definition.key, {
      definition,
      state: definition.init(),
      lastSeq: SessionSeq(0)
    })
  }
  
  /** 应用事件 */
  apply(event: SessionEvent): void {
    for (const instance of this.projections.values()) {
      if (event.seq > instance.lastSeq) {
        instance.state = instance.definition.apply(instance.state, event)
        instance.lastSeq = event.seq
      }
    }
  }
  
  /** 获取状态 */
  stateOf<K extends string>(key: K): ProjectionState<K> {
    const instance = this.projections.get(key)
    if (!instance) throw new Error(`Projection "${key}" not found`)
    return instance.state
  }
  
  /** 批量快照 */
  snapshot(keys: string[]): Record<string, unknown> {
    const result: Record<string, unknown> = {}
    for (const key of keys) {
      result[key] = this.stateOf(key)
    }
    return result
  }
}

11.5 日志重放（Replay）

11.5.1 概念

日志重放允许从现有日志重建会话状态。这是 Fork 和恢复的基础。

11.5.2 重放流程

// 概念性描述
class SessionReplay {
  /** 重放日志 */
  async replay(log: SessionEventLog, options: ReplayOptions): Promise<SessionState> {
    let state = this.createInitialState()
    
    for await (const event of log.read(options.offset)) {
      // 应用事件到状态
      state = this.applyEvent(state, event)
      
      // 检查是否需要停止
      if (options.maxSeq && event.seq > options.maxSeq) {
        break
      }
    }
    
    return state
  }
  
  /** 应用事件 */
  private applyEvent(state: SessionState, event: SessionEvent): SessionState {
    switch (event.type) {
      case 'user/message':
        return { ...state, messages: [...state.messages, event.data] }
      case 'assistant/message':
        return { ...state, messages: [...state.messages, event.data] }
      case 'tool/call':
        return { ...state, pendingToolCalls: [...state.pendingToolCalls, event.data] }
      case 'tool/result':
        return {
          ...state,
          pendingToolCalls: state.pendingToolCalls.filter(tc => tc.callId !== event.data.callId),
          toolResults: [...state.toolResults, event.data]
        }
      default:
        return state
    }
  }
}

11.6 Fork 机制

11.6.1 概念

Fork 允许从现有会话创建新会话，共享部分历史。

11.6.2 Fork 实现

// 概念性描述
class SessionFork {
  /** Fork 会话 */
  async fork(
    parentSession: Session,
    options: ForkOptions
  ): Promise<Session> {
    // 1. 确定 Fork 点
    const forkPoint = options.forkPoint ?? parentSession.currentSeq
    
    // 2. 复制历史事件
    const history = []
    for await (const event of parentSession.log.read()) {
      if (event.seq > forkPoint) break
      history.push(event)
    }
    
    // 3. 创建新会话
    const childSession = await this.createSession({
      parentSession: parentSession.id,
      isSeeded: true,
      inheritedEventCount: forkPoint
    })
    
    // 4. 写入历史事件
    await childSession.log.append(history)
    
    return childSession
  }
}

11.6.3 Fork 选项

interface ForkOptions {
  /** Fork 点（seq）*/
  forkPoint?: SessionSeq
  /** 是否复制元数据 */
  copyMetadata?: boolean
  /** 子会话的额外元数据 */
  meta?: Partial<SessionHeader>
}

11.7 SessionQuery 服务

11.7.1 查询接口

// 来自 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[]>
}

11.7.2 语义搜索

// 来自 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
  }
}

11.8 配置

11.8.1 持久化配置

# cordis.patch.yml
- id: session-persistence
  config:
    # 后端类型
    backend: 'jsonl'  # 'jsonl' | 'sqlite' | 'memory'
    
    # JSONL 配置
    jsonl:
      dir: '${DSH_HOME}/sessions'
      compress: true  # 启用压缩
      maxFileSize: 10485760  # 10MB
    
    # 自动清理
    cleanup:
      enabled: true
      maxAge: 2592000000  # 30 天
      maxCount: 1000

11.8.2 投影配置

# cordis.patch.yml
- id: session-projection
  config:
    # 启用的投影
    projections:
      - turnBoundary
      - toolUsage
      - errorSummary
    
    # 缓存配置
    cache:
      enabled: true
      maxSize: 1000

11.9 小结

概念	一句话解释
JSONL	每行一个 JSON 对象的日志格式
SessionSeq	单调递增的事件序列号
Projection	从日志增量派生状态
Replay	从日志重建会话状态
Fork	从现有会话创建新会话
SessionQuery	会话查询和语义搜索

下一步：第十二章：Typert 类型图系统——深入理解类型图生成、运行时类型注册和跨仓库类型发现。

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