会话状态存储

async def pre_agent_execute(self, session: BaseSession, inputs):
    """Prepare agent execution by recovering agent state."""
    session_logger.info(
        "Agent checkpoint restore initiated",
        event_type=LogEventType.CHECKPOINT_RESTORE,
        session_id=session.session_id(),
        agent_id=session.agent_id(),
        metadata={"operation": "pre_execute", "storage_type": "persistence"}
    )
    await self._agent_storage.recover(session)
    if inputs is not None:
        session.state().update({INTERACTIVE_INPUT: [inputs]})
# ...（758-781 行：interrupt_agent_execute：需要人工交互时同样调用 save）
async def post_agent_execute(self, session: BaseSession):
    """Save agent state after execution completes."""
    session_logger.info(
        "Agent checkpoint save on completion",
        event_type=LogEventType.CHECKPOINT_SAVE,
        session_id=session.session_id(),
        agent_id=session.agent_id(),
        metadata={"reason": "agent_finished", "storage_type": "persistence"}
    )
    await self._agent_storage.save(session)

/// Writes canonical session rollout items to JSONL.
///
/// Rollouts are recorded as JSONL and can be inspected with tools such as:
///
/// ```ignore
/// $ jq -C . ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl
/// $ fx ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl
/// ```
#[derive(Clone)]
pub struct RolloutRecorder {
    tx: Sender<RolloutCmd>,
    writer_task: Arc<RolloutWriterTask>,
    pub(crate) rollout_path: PathBuf,
}
// ...（91-1652 行：RolloutRecorderParams、写入命令与录制器实现）
/// Mutable state owned by the background rollout writer.
///
/// Items are first appended to `pending_items`; persist/flush/shutdown remove each item from that
/// queue only after it is written successfully. I/O failures drop the file handle but keep the
/// unwritten suffix so the next barrier can reopen the file and retry.
struct RolloutWriterState {
    writer: Option<JsonlWriter>,
    /// True until a newly created rollout is first materialized.
    deferred_creation: bool,
    pending_items: Vec<RolloutItem>,
    meta: Option<SessionMeta>,
    cwd: PathBuf,
    rollout_path: PathBuf,
    ordinal_state: RolloutOrdinalState,
    last_logged_error: Option<String>,
}

/**
 * Durable session-persistence Service Definition (`ctx.sessionPersistence`). Backends store
 * {@link SessionEvent}s as the event-sourced log and carry non-replayable
 * {@link SessionHeader} metadata separately.
 * @module @deepseek-ai/dsh-session-persistence
 */
  // ...（7-134 行：快照、检视、原始制品等辅助类型与方法）
  /**
   * Durably persist a batch of events. Honors the append-only and contiguous-
   * seq contracts: the first event's `seq` MUST equal the stored next-seq
   * (after `load` has durably closed any interrupted turn). Rejects non-JSON-
   * serializable `event.data` with an error naming the offending event type.
   * @param id - the session the batch belongs to.
   * @param events - the contiguous batch to persist, in seq order.
   */
  abstract append(id: SessionId, events: readonly SessionEvent[]): Promise<void>
  // ...（144-154 行：prepare 的文档注释）
  async prepare(id: SessionId, signal?: AbortSignal): Promise<SessionPreparation> {
    signal?.throwIfAborted()
    const loaded = await this.load(id)
    signal?.throwIfAborted()
    const sessions = this.ctx.get('sessions')
    if (sessions === undefined) {
      throw new Error('cannot prepare a session: SessionStore is not configured')
    }
    return SessionPreparation.create(sessions.prepare(id, {
      seed: loaded.events.map(event => structuredClone(event)),
      meta: structuredClone(loaded.meta),
      seedSource: 'persistence',
    }))
  }

export interface EntryBase {
	type: string;
	id: string;
	seq: number; // shared sequence; read-side, storage-assigned
	parentId: string | null; // storage-assigned: the appending lane's leaf
	timestamp: number; // Unix ms, storage-assigned
}
// ...（21-289 行：各类条目与记录的类型定义）
export interface SessionStorage<TMetadata extends SessionMetadata = SessionMetadata> {
	getMetadata(): Promise<TMetadata>;

	// Lanes
	getLanes(): Promise<{ lane: string; leafId: string | null }[]>;
	createLane(lane: string, at: string | null): Promise<void>;
	moveLane(lane: string, to: string | null): Promise<void>;

	// Entries and Records
	appendEntry<TEntry extends Entry>(entry: ProvisionedEntry<TEntry>, lane: string): Promise<TEntry>;
	appendRecord<TRecord extends LaneRecord>(record: NewRecord<TRecord>): Promise<TRecord>;

	// Reads
	getEntry(id: string): Promise<Entry | undefined>;
	findEntries(query?: EntryQuery): Promise<Entry[]>;
	/** start is mandatory here (as opposed to SessionTree's findEntriesOnBranch); defaulting to a lane's leaf is view sugar. */
	findEntriesOnBranch(query: EntryQuery & BranchBounds & { start: string }): Promise<Entry[]>;
	findRecords<K extends LaneRecord["type"]>(
		query: RecordQuery & { type: K },
	): Promise<Extract<LaneRecord, { type: K }>[]>;
	findRecords(query?: RecordQuery): Promise<LaneRecord[]>;
	/**
	 * Returns unfinished operation starts newest first. Recovery uses `limit: 2`:
	 * zero results mean the lane is idle, one means it is suspended, and two
	 * mean at least two operations are open, which is corruption. Further
	 * results provide no additional recovery state.
	 */
	findOpenOperations(lane: string, options?: { limit?: number }): Promise<OperationStartedRecord[]>;
	getLog(options?: { afterSeq?: number; limit?: number }): Promise<LogItem[]>;