/// Future returned by [`ThreadStore`] operations. // 统一返回类型：把异步操作装箱成 Send future，trait 才能跨后端共享pub type ThreadStoreFuture<'a, T> = Pin<Box<dyn Future<Output = ThreadStoreResult<T>> + Send + 'a>; // 生命周期 'a 绑定到调用方持有的 store 引用/// Data a [`ThreadStore`] has already loaded while preparing a thread for resume. // 恢复线程前预加载的数据包pub struct ThreadStoreResumePreparation { // core 不解析内容，只负责原样传回给 store    state: Box<dyn Any + Send>, // 不透明类型：里面装的是实现私有的预加载状态（如已打开的游标）}impl ThreadStoreResumePreparation { // 提供装箱/拆箱两个操作    pub fn new<T: Any + Send>(state: T) -> Self { // 把任意类型塞进不透明包        Self { state: Box::new(state) } // 堆上分配，调用方不再持有原对象    }    pub fn downcast<T: Any + Send>(self) -> Result<T, Self> { // 拆箱：类型对就取出，不对就把原包还回去（不丢资源）        match self.state.downcast::<T>() { // 运行时类型检查            Ok(state) => Ok(*state), // 类型匹配：交出内部值            Err(state) => Err(Self { state }), // 类型不符：原样包回，调用方可继续尝试别的实现        }    }}/// Storage-neutral thread persistence boundary. // 存储中立的线程持久化边界：core 只认这个接口pub trait ThreadStore: Any + Send + Sync { // 要求 Any：才能 as_any() 下转；Send+Sync：跨线程共享    /// Return this store as [`Any`] for implementation-owned escape hatches. // 暴露具体类型句柄，供 core 调用实现专属能力    fn as_any(&self) -> &dyn Any; // 返回 trait object 引用，不消耗 self    /// Creates a new live thread. // 创建一条"活"线程并安装它的写者    fn create_thread(&self, params: CreateThreadParams) -> ThreadStoreFuture<'_, ()>; // 成功即线程可追加写入    /// Reopens an existing thread for live appends and returns its authoritative replay context. // 重开旧线程，返回可回放给模型的权威历史    fn resume_thread( // 多行签名：参数与返回值分行写        &self, // 借用 store，不转移所有权        params: ResumeThreadParams, // 携带 thread_id、可选自带历史、修订号等    ) -> ThreadStoreFuture<'_, Arc<Vec<RolloutItem>>>; // 返回值是共享引用：多个调用方复用同一份历史，不拷贝    /// Appends raw rollout items to a live thread. // 追加新对话记录（用户消息/模型输出/工具事件）    fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreFuture<'_, ()>; // 写者必须已存在，否则报错    /// Materializes the thread if persistence is lazy, then persists all queued items. // 持久化检查点：让已入队内容全部落盘可读    fn persist_thread( // 三个参数分行：id + 场景枚举        &self, // 借用 store        thread_id: ThreadId, // 目标线程        context: PersistContext, // 调用方声明"为什么持久化"，store 据此决定能否后台延后    ) -> ThreadStoreFuture<'_, ()>; // Standard 场景必须同步完成；TurnStart 等可先返回    /// Flushes all queued items and returns once they are durable/readable. // flush 屏障：等写入真正完成才返回    fn flush_thread(&self, thread_id: ThreadId) -> ThreadStoreFuture<'_, ()>; // 不产生新数据，只确认已入队内容落盘    /// Flushes pending items and closes the live thread writer. // 正常关闭：flush + 释放写者资源    fn shutdown_thread(&self, thread_id: ThreadId) -> ThreadStoreFuture<'_, ()>; // 之后该线程不再接受追加    /// Discards the live thread writer without forcing pending in-memory items to become durable. // 异常丢弃：初始化失败时不强制把脏数据写盘    fn discard_thread(&self, thread_id: ThreadId) -> ThreadStoreFuture<'_, ()>; // 与 shutdown 的区别：不等落盘，直接释放}

/// Why thread persistence is being requested. // 为什么要求持久化：不同场景的耐久性要求完全不同#[derive(Clone, Copy, Debug, Eq, PartialEq)] // 纯数据枚举：可拷贝、可比对，方便在参数里传递pub enum PersistContext { // 五种持久化场景，语义互斥    ThreadPreparation, // 线程初始化/上下文注入完成，但还没开始任何 turn    Standard,          // 标准检查点：让线程和所有已入队条目都变持久、可读    SubagentSpawn,     // 子代理派生：子历史可先入队，其耐久性栅栏与"父子关系落盘"重叠执行    TurnStart,         // turn 即将开始采样：输入已经记录完毕    SteeredUserInput,  // 用户中途注入新指令：在下一轮采样请求前记录下来}impl PersistContext { // 把"能否后台化"的策略收敛到一处，不散落在各调用点    /// Whether a store may enqueue this checkpoint before returning and fence it at a later durability barrier. // 文档注释：哪些检查点允许先返回、由后续栅栏兜底    pub fn allows_background_persistence(self) -> bool { // 按场景回答"能否后台写"        match self { // 按场景分派策略            Self::ThreadPreparation | Self::Standard => false, // 这两类必须同步完成：调用方紧接着就要读这份数据            Self::SubagentSpawn | Self::TurnStart | Self::SteeredUserInput => true, // 这三类可先入队，由后续 flush/shutdown 栅栏兜底        }    }}

struct LiveRecorderEntry { // 进程内"活写者表"的一条记录：一个线程同一时刻只能有一个    recorder: RolloutRecorder, // 真正的 JSONL 写者（rollout crate 提供）    rollout_id: ThreadId,      // 不可变的 rollout ID：SQLite 投影行按它索引，而不是按查找用的稳定 thread id    history_mode: ThreadHistoryMode, // Legacy 或 Paginated：决定写入后要不要做 SQLite 投影    writer_lock: WriterLockGuard,    // 跨进程文件锁的守卫：条目被移除时自动释放}#[derive(Default)] // 空表即可用，无需手动初始化struct LiveWriterLocks { // 进程内每线程一组协调锁的注册表    // Keep per-thread locks after a writer goes idle. Removing one while another caller is about to acquire it could let two operations for the same thread run at once. // 中文：写者空闲后保留锁，防止"刚要拿就被回收"导致同线程两个操作并发    by_thread: Mutex<HashMap<ThreadId, Arc<ThreadCoordination>>>, // 锁本身常驻：key 是线程 id，value 是该线程的协调锁组}#[derive(Default)] // 默认构造即两把空锁struct ThreadCoordination { // 每个线程一组进程内协调锁    // Serialize writes and capture consistent fork snapshots. // 中文：写串行化 + fork 快照一致性都靠它    writer: Arc<Mutex<()>>,   // 写串行化锁：append/flush/fork 快照都靠它独占    // Forks hold a shared lease until their child reference is durable; deletion, archive, and unarchive require exclusive access. Keeping this separate from `writer` lets the source accept writes during child initialization, including MCP startup that can take 30 seconds. // 中文：fork 持共享租约直到子引用落盘；删除/归档才需独占。与 writer 分开，源线程在子初始化（MCP 启动可达 30 秒）期间仍可接受写入    lifecycle: Arc<RwLock<()>>, // 生命周期读写锁：fork 持共享读，删除/归档持独占写}

fn acquire_writer_lock(&self, thread_id: ThreadId) -> ThreadStoreResult<WriterLockGuard> { // 跨进程抢"该线程的写者身份"    self.writer_lock_coordinator // 文件锁协调器（按 codex_home 初始化，所有本地 store 共享同一把物理锁）        .acquire(thread_id) // 非阻塞尝试获取该线程的文件锁：拿不到立刻失败，不死等        .map(Arc::new) // 包成 Arc：guard 可随 LiveRecorderEntry 共享持有、延迟释放        .map_err(|err| { // IO 错误翻译成业务错误类型            if err.kind() == std::io::ErrorKind::WouldBlock { // WouldBlock = 别的进程正持有写者                ThreadStoreError::Conflict { message: err.to_string() } // 翻译成业务错误：冲突，调用方可以重试或报错            } else { // 非 WouldBlock 的其他 IO 错误                ThreadStoreError::Internal { message: err.to_string() } // 归为内部故障            }        }) // 结束错误映射闭包}async fn acquire_writer_locks(&self, thread_ids: &[ThreadId]) -> ThreadStoreResult<Vec<WriterLockGuard>> { // 批量抢锁（fork 场景要同时锁住父线程）    let mut writer_locks = Vec::with_capacity(thread_ids.len()); // 预分配容量，逐个收集锁守卫    for &thread_id in thread_ids { // 遍历要锁的线程列表        if self.live_recorders.lock().await.contains_key(&thread_id) { // 本进程已经是该线程的写者：跳过，避免自己锁死自己（文件锁不可重入）            continue; // 不重复获取文件锁        }        writer_locks.push(self.acquire_writer_lock(thread_id)?); // 逐个获取，任一失败整体回滚（已拿到的 guard drop 时自动释放）    }    Ok(writer_locks) // 全部拿到：guard 存活期间锁持续有效}

let history = match params.history { // 调用方可能自带一份历史（比如 core 内存里还留着），先判断能不能直接复用    Some(history) // 分支一：带了历史，且首行不是本线程的 SessionMeta        if !matches!( // 守卫条件：检查第一行是不是"本线程的会话头"            history.first(), // 取历史首条记录            Some(RolloutItem::SessionMeta(meta)) if meta.meta.id == params.thread_id // 是别的线程/导入的历史 → 走导入语义，原样采用        ) =>    {        history // 直接复用内存副本，零磁盘 IO    }    Some(history) // 分支二：带了历史且首行是本线程 SessionMeta，再看修订号        if params.history_revision.is_some() // 守卫条件一：必须带了修订号（缺失不能当作"没变过"的证据）            && params.history_revision // 守卫条件二：修订号与磁盘当前值相等 = 期间无人写入                == super::history_revision::read(&rollout_path).await // 异步读一次小文件，比回放整个 rollout 便宜得多    =>    {        history // 校验通过：复用内存副本    }    _ => Arc::new( // 分支三（兜底）：从磁盘重新加载权威历史——revision 缺失或不匹配都走这里        super::model_context::load_from_rollout_path(store, params.thread_id, &rollout_path) // 全量读取正本文件并重建模型可见上下文            .await? // 异步读盘，失败向上抛            .items, // 取出历史条目列表    ), // Arc 包装：共享给多个调用方，避免拷贝};

/// The rollout writer has three distinct lifecycle moments: // 写者有三个不同的生命周期时刻（下面枚举逐一对应）/// - `AppendItems` is normal turn/event persistence and adds new rollout records. // AppendItems = 追加新记录（turn/事件持久化）/// - `Persist` makes the thread durable before any turn items exist; locally this can write the initial `SessionMeta`. // Persist = 让线程变持久：本地可能写出首个 SessionMeta/// - `Flush` writes any rollout records already queued in the recorder and ensures they are durably persisted. // Flush = 把已入队记录全部刷盘并确认持久enum RolloutWriteOp { // 三种写操作统一建模，后续分支只认这个枚举    AppendItems(Vec<RolloutItem>), // 追加新对话记录    Persist,                       // 让线程在磁盘上"存在"（可能写出初始 SessionMeta）    Flush,                         // 纯屏障：不产生新数据，只等已入队内容落盘}async fn write_and_project( // 所有活写操作的唯一入口（异步函数）    store: &LocalThreadStore, // store 引用：查锁表与写者表    thread_id: ThreadId, // 目标线程    write_op: RolloutWriteOp, // 三种写操作之一（追加/持久化/刷盘）) -> ThreadStoreResult<()> { // 成功返回 ()，失败带业务错误    let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; // 先拿进程内写锁：串行化本线程的所有写入    let (recorder, rollout_id, history_mode) = live_writer_parts(store, thread_id).await?; // 取出该线程的写者三元组（找不到说明已拆除，迟到写入在此失败）    let sync_rollout_path = matches!(&write_op, RolloutWriteOp::Persist | RolloutWriteOp::Flush); // Persist/Flush 之后要同步 state_db 里的 rollout 路径    if matches!(history_mode, ThreadHistoryMode::Legacy) { // Legacy 模式：只有正本，没有投影这回事        durable_write(&recorder, write_op).await?; // 只写 JSONL，失败直接报错（正本没写成就是真失败）    } else { // Paginated 模式：正本 + SQLite 视图两步走        let rollout_path = recorder.rollout_path(); // 记下正本路径，供投影与同步使用        // SQLite is a rebuildable view. The flush barrier must win before projection starts so it can lag JSONL after failure, but can never get ahead of canonical history. // 中文：SQLite 是可重建视图——失败后它可以落后于 JSONL，但绝不能跑到正本前面        durable_write(&recorder, write_op).await?; // 第一步永远先写正本：JSONL 落盘成功才算数        if let Err(err) = super::thread_history_materialization::materialize_to_sqlite( // 第二步才做投影；失败不致命——捕获错误只告警（视图可重建）            store, rollout_id, rollout_path, // 按不可变 rollout_id 物化进 turns/items 表        ).await { // 异步物化完成或失败            warn!("failed to project durable rollout for {thread_id}: {err}"); // 投影失败只告警不报错——正本没丢，下次可追平        }    }    if sync_rollout_path { // Persist/Flush 之后才需要同步路径引用（压缩/归档会换文件名）        sync_materialized_rollout_path(store, thread_id, recorder.rollout_path()).await?; // 把"当前正本文件路径"写回 state_db    }    Ok(()) // 写 + 投影全部完成}async fn durable_write(recorder: &RolloutRecorder, write: RolloutWriteOp) -> ThreadStoreResult<()> { // 对正本的唯一写入口    match write { // 按操作类型分派到 recorder 对应方法        RolloutWriteOp::AppendItems(items) => { // 追加：先记录再刷盘，两步都成功才算数            recorder.record_canonical_items(items.as_slice()).await.map_err(thread_store_io_error)?; // 记录规范化条目（内部按 history_mode 过滤）            recorder.flush().await.map_err(thread_store_io_error) // 立即刷盘：append 的语义就是"写完即可读"        }        RolloutWriteOp::Persist => recorder.persist().await.map_err(thread_store_io_error), // 写出初始 SessionMeta，让线程在磁盘上"存在"        RolloutWriteOp::Flush => recorder.flush().await.map_err(thread_store_io_error), // 纯屏障：不产生新数据，只等已入队内容落盘    }}

写路径（live append + projection）

▼

▼

▼

pub(super) async fn migrate_rollouts_on_startup(store: &LocalThreadStore) -> ThreadStoreResult<()> { // 启动迁移入口（crate 内可见）    let Some(state_db) = store.state_db.as_ref() else { // let-else：没有 SQLite 状态库就直接返回（Legacy-only 部署不受影响）        return Ok(()); // 分页模式不可用，跳过整个迁移    };    let paths = find_all_rollout_paths(&store.config.codex_home).await?; // 扫描 codex_home 下所有 rollout JSONL 文件    let mut skipped_rollouts = state_db // 从 SQLite 读"跳过清单"（可变，后面会更新）        .list_rollout_migration_skipped_rollouts(LEGACY_TO_PAGINATED_MIGRATION_ID) // 按迁移 ID 查询：不同迁移互不干扰        .await // 异步读库        .map_err(migration_error)?; // 统一错误包装，避免反复处理已知问题文件    retry_busy_rollouts(store, skipped_rollouts.as_slice(), paths.as_slice()).await?; // 先重试上次"忙"的文件：当时被写者占着，现在可能空闲了    if !pending_migration_thread_ids(&store.config.codex_home).await?.is_empty() { // 检查是否有上次崩溃留下的未完成迁移标记        return migrate_all_rollouts(store, paths, skipped_rollouts.as_slice()).await; // 有 → 全量重跑（迁移幂等 + journal 兜底）    }    let state = state_db.get_rollout_migration_state(LEGACY_TO_PAGINATED_MIGRATION_ID) // 读迁移状态（含增量游标 last_checked_thread）        .await        .map_err(migration_error)?; // 统一错误包装    if state.is_none() { // 从未迁移过：首次全量        return migrate_all_rollouts(store, paths, skipped_rollouts.as_slice()).await; // 一次性把所有 legacy 文件升级    }    let last_checked_thread = state.and_then(|state| state.last_checked_thread); // 增量游标：上次检查到哪个线程为止（无则 None）    let lookback_created_at = last_checked_thread.as_ref().map(|cursor| { // 计算回看窗口起点，覆盖"上次检查窗口内刚写入"的边界竞态        cursor.thread_created_at.saturating_sub(CURSOR_LOOKBACK_SECONDS) // 游标时间 - N 秒（saturating 防下溢）    }); // 无游标时为 None，表示不设下限    let candidates = paths.iter() // 只挑候选：不在跳过清单里、且创建时间落在回看窗口内的新文件        .filter(|path| { /* 按文件名排除 skipped；thread_creation_cursor(path) >= lookback_created_at */ }) // 过滤条件见源码（此处节选）        .collect::<Vec<_>>(); // 收集成候选列表    if candidates.is_empty() { // 没有新文件：日常启动到此结束，开销趋近于零        return Ok(()); // 直接返回成功    }    let mut unresolved = false; // 标记是否有"正在被别的进程写"的文件    for path in candidates { // 逐个体检候选文件        match inspect_rollout_path(store, path).await? { // 读 SessionMeta + 前后指纹比对，判定状态            StartupInspection::Paginated | StartupInspection::Skipped => {} // 已是新格式 / 已知跳过：无事发生            StartupInspection::Legacy | StartupInspection::NeedsMigration => { // 发现需要迁移的 → 升级为全量迁移（保证整批一致）                return migrate_all_rollouts(store, paths, skipped_rollouts.as_slice()).await; // 立即执行并返回结果            }            StartupInspection::Unresolved => unresolved = true; // 文件正被别的进程写：本轮放弃，等下次启动再试        }    }    if unresolved { return Ok(()); } // 有未决文件：本轮收手（宁可这次不迁，不能迁坏在用文件）    advance_last_checked_thread(store, paths.as_slice()).await // 全部干净：把游标推进到当前快照，下次只查增量}

#[derive(Clone)] // ActiveTurn 需要 clone 进合成事件结构里struct ActiveTurn { // 重放过程中"当前正在写的 turn"状态    id: String,      // turn ID：显式的来自 TurnStarted 事件；隐式的合成 rollout-{源行号}（可追溯）    explicit: bool,  // 是否由旧文件显式开启（隐式 turn 结束时要补一个合成的 TurnComplete）    saw_user: bool,  // 本 turn 是否已见用户消息：决定隐式 turn 何时该收尾}async fn write_item( // 写一条规范化记录的唯一出口（异步函数）    &mut self, // 可变借用：要更新序号/偏移等状态    writer: &mut BufWriter<File>, // 缓冲文件写者：批量落盘减少 syscall    timestamp: &str, // 源行原始时间戳（RFC3339）——迁移时间不是对话发生时间    item: RolloutItem, // 待写入的规范化条目) -> ThreadStoreResult<()> { // 成功返回 ()，失败带迁移错误    let item = match codex_rollout::persisted_rollout_item(&item, ThreadHistoryMode::Paginated) { // 按 Paginated 语义过滤/改写：返回借用、新值或 None        Some(Cow::Borrowed(_)) => item,   // 无需改写：原样写入（零拷贝）        Some(Cow::Owned(item)) => item,   // 需要按分页语义改写：用改写后的版本        None => return Ok(()),            // 该记录不属于分页历史（如旧版一次性事件）：静默丢弃，不报错    };    let mut bytes = serde_json::to_vec(&RolloutLine { // 序列化成 JSONL 一行（时间戳 + ordinal + 条目）        timestamp: timestamp.to_string(), // 保留源行原始时间戳        ordinal: Some(self.next_ordinal), // 补上全局单调递增的序号（legacy 文件没有 ordinal，分页格式必须有）        item, // 条目本体    }).map_err(migration_error)?; // 序列化失败视为迁移错误，中止本文件    bytes.push(b'\n');                    // JSONL 约定：一条记录一行    writer.write_all(&bytes).await.map_err(migration_error)?; // 写入缓冲；IO 错误直接上抛（外层有回滚规划）    let byte_count = u64::try_from(bytes.len()).map_err(|_| migration_error("rollout record exceeds addressable size"))?; // 单条记录超过可寻址大小：视为损坏，报错中止    self.output_byte_offset = self.output_byte_offset.checked_add(byte_count) // 累计输出字节偏移：回滚规划和进度上报都靠它（checked 防溢出）        .ok_or_else(|| migration_error("paginated rollout byte offset overflow"))?; // 偏移溢出：宁可报错也不静默回绕    self.next_ordinal = self.next_ordinal.checked_add(1).ok_or_else(|| migration_error("paginated rollout ordinal overflow"))?; // 序号 +1（同样 checked）    Ok(()) // 本条写入成功，序号已推进}

启动迁移路径（legacy → paginated）

▼

▼

▼

▼