Thread History Projection

本文承接V2 Thread协议和V2 Turn与Item协议，面向已经理解 rollout 与 V2 API view 的读者。本文回答持久化 item 如何重建 Turn、itemsView 如何裁剪内容、resume 如何分页和脱敏；不展开 rollout 文件格式本身。

1. 投影管线

持久化 rollout 是输入事实，V2 Thread 是面向客户端的投影视图。投影会分组 Turn、转换 item、选择视图、分页，并按客户端能力脱敏。

2. Turn重建

源码位置：codex-rs/app-server/src/request_processors.rs :: build_legacy_api_turns_from_rollout_items

pub(crate) fn build_legacy_api_turns_from_rollout_items(
    items: &[RolloutItem],
) -> Vec<Turn> {
    items.iter()
        .filter(|item| is_persisted_rollout_item(item, ThreadHistoryMode::Legacy))
        .fold(TurnReconstruction::default(), |state, item| state.push(item))
        .finish()
}

只有符合 history mode 的 item 进入重建；事件顺序决定 Turn 边界和状态。未知或非持久化 item 可能不会出现在 V2 历史中。

3. Canonical行投影

源码位置：codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs :: project_rollout_line

pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet {
    match &line.item {
        RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => ThreadHistoryChangeSet {
            changed_turns: vec![ThreadHistoryTurnChange {
                turn_id: event.turn_id.clone(),
                status: TurnStatus::InProgress,
                error: None,
                started_at: event.started_at,
                completed_at: None,
                duration_ms: None,
            }],
            ..Default::default()
        },
        RolloutItem::EventMsg(EventMsg::TurnComplete(event)) => ThreadHistoryChangeSet {
            changed_turns: vec![ThreadHistoryTurnChange {
                turn_id: event.turn_id.clone(),
                status: if event.error.is_some() {
                    TurnStatus::Failed
                } else {
                    TurnStatus::Completed
                },
                error: event.error.as_ref().map(|error| TurnError {
                    message: error.message.clone(),
                    codex_error_info: error.codex_error_info.clone().map(Into::into),
                    additional_details: None,
                }),
                started_at: event.started_at,
                completed_at: event.completed_at,
                duration_ms: event.duration_ms,
            }],
            ..Default::default()
        },
        RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) => ThreadHistoryChangeSet {
            changed_items: vec![ThreadHistoryItemChange {
                turn_id: event.turn_id.clone(),
                item: ThreadItem::from(event.item.clone()),
                started_at_ms: event.started_at_ms,
                completed_at_ms: (event.completed_at_ms != 0).then_some(event.completed_at_ms),
            }],
            ..Default::default()
        },
        _ => ThreadHistoryChangeSet::default(),
    }
}

这个函数只处理 canonical paginated rollout 中能直接映射的行：TurnStarted/Complete 更新 turn，ItemCompleted 更新 item；ResponseItem、SessionMeta、RealtimeItem 等行不会凭空制造 ThreadItem。调用方按 rollout ordinal 顺序逐行应用 change set，才能保留重复 item 的首次和最新时间戳。

4. ItemsView

源码位置：codex-rs/app-server/src/request_processors/thread_processor.rs :: apply_thread_turns_items_view

match items_view {
    TurnItemsView::NotLoaded => {
        turn.items.clear();
        turn.items_view = TurnItemsView::NotLoaded;
    }
    TurnItemsView::Summary => {
        turn.items = [first_user_message, last_agent_message]
            .into_iter().flatten().collect();
        turn.items_view = TurnItemsView::Summary;
    }
    TurnItemsView::Full => {
        turn.items_view = TurnItemsView::Full;
    }
}

Summary 不是 Full 的压缩编码，而是只保留代表性 user/agent message；工具调用、reasoning 和中间输出会丢失。NotLoaded 则明确表示 item 未加载，不能解释为空 Turn。

5. Live与Stored

运行中的 listener 已将事件投影进 ThreadState，read/resume 不应再次从 rollout 重复追加 live Turn；stored thread 则从 JSONL/SQLite projection 加载。SQLite 可能滞后，分页路径以 canonical JSONL 为准。

6. 分页读取

源码位置：codex-rs/thread-store/src/local/thread_history/read.rs :: list_turns、list_items、validate_thread_for_paginated_reads

pub(in crate::local) async fn list_turns(
    store: &LocalThreadStore,
    params: ListTurnsParams,
) -> ThreadStoreResult<TurnPage> {
    validate_thread_for_paginated_reads(
        store,
        params.thread_id,
        params.include_archived,
        "list_turns",
    )
    .await?;
    validate_page_size(params.page_size)?;
    let lineage = store.resolve_rollout_lineage(params.thread_id).await?;
    let pool = store.thread_history_db().await?;
    let page = page_turn_rows(
        pool,
        params.thread_id,
        &lineage,
        params.cursor.as_deref(),
        params.page_size,
        params.sort_direction,
        params.items_view,
    )
    .await?;
    let turns = page
        .rows
        .into_iter()
        .map(|turn| StoredTurn {
            turn_id: turn.turn_id,
            items: match params.items_view {
                StoredTurnItemsView::NotLoaded => Vec::new(),
                StoredTurnItemsView::Summary => turn.summary_items,
            },
            items_view: params.items_view,
            status: turn.status,
            error: turn.error,
            started_at: turn.started_at,
            completed_at: turn.completed_at,
            duration_ms: turn.duration_ms,
        })
        .collect();
    Ok(TurnPage {
        turns,
        next_cursor: page.next_cursor,
        backwards_cursor: page.backwards_cursor,
    })
}

分页读取先验证线程存在、未归档或明确允许归档，并拒绝 legacy history mode；随后沿 RolloutLineage 查询 SQLite history。items_view=NotLoaded 不读取 item，Summary 只联结首个 user 与最终 agent 摘要，Full 的完整 item 列表由独立 item 查询提供。

源码位置：codex-rs/thread-store/src/local/thread_history/read.rs :: parse_cursor

pub(super) fn parse_cursor(
    cursor: Option<&str>,
    requested_thread_id: ThreadId,
    scope: CursorScope,
) -> ThreadStoreResult<Option<HistoryCursor>> {
    let Some(cursor) = cursor else {
        return Ok(None);
    };
    let cursor_value: HistoryCursor =
        serde_json::from_str(cursor).map_err(|_| invalid_cursor(cursor))?;
    if cursor_value.requested_thread_id != requested_thread_id
        || cursor_value.scope != scope
    {
        return Err(invalid_cursor(cursor));
    }
    Ok(Some(cursor_value))
}

cursor 绑定请求线程和读取 scope，因此不能把 turns cursor 用到 items 查询，也不能跨线程复用。分页不是 rollout 字节切片，而是带 lineage、ordinal 和 scope 的结构化位置。

7. Live Turn合并

源码位置：codex-rs/app-server/src/request_processors/thread_lifecycle.rs :: merge_turn_history_with_active_turn

pub(super) fn merge_turn_history_with_active_turn(
    turns: &mut Vec<Turn>,
    active_turn: Turn,
) {
    if let Some(existing) = turns.iter_mut().find(|turn| turn.id == active_turn.id) {
        *existing = active_turn;
    } else {
        turns.push(active_turn);
    }
}

resume/read 组装历史时会把 listener 中的 active turn 合并到已加载 turns；同 id 替换，缺失则追加。这个步骤避免实时事件和持久化 history 同时出现时产生重复 Turn。

8. Resume分页与脱敏

initialTurnsPage、turn cursor 和 item cursor 允许 resume/read 分批返回历史。cursor 表示投影位置，不是 rollout 字节偏移；客户端应原样回传，并保持相同 itemsView 参数。

9. Resume脱敏

源码位置：codex-rs/app-server/src/request_processors/thread_resume_redaction.rs :: redact_thread_resume_payloads

for turn in turns {
    turn.items.retain_mut(|item| match item {
        ThreadItem::McpToolCall { result, error, .. } => {
            *result = Some(Box::new(redacted_mcp_tool_call_result()));
            if let Some(error) = error { error.message = "[redacted]".to_string(); }
            true
        }
        ThreadItem::ImageGeneration(_) => false,
        _ => true,
    });
}

某些客户端 resume 会将 MCP 结果/错误替换为 [redacted]，并移除图片生成 item。脱敏后的响应不能用于完整重放或调试原始 payload。

10. 失败与信息损失

损坏 item JSON、缺失 rollout、unsupported store operation 和 cursor 不一致会产生不同错误。Summary、redaction 和未知 item 过滤属于有意信息损失；它们不是存储损坏。恢复逻辑只能基于仍保留的字段工作。

11. 源码验证

itemsView 测试验证 NotLoaded 清空、Summary 只保留首个 user 和最后 agent message；redaction 测试验证 MCP payload 替换和 image generation 移除；resume 分页测试验证 cursor 与 canonical JSONL。它们证明公开视图边界，不证明 Full view 无任何 mapper 信息损失。

源码位置：

• codex-rs/app-server/src/request_processors/thread_processor.rs :: apply_thread_turns_items_view
• codex-rs/app-server/src/request_processors/thread_resume_redaction.rs :: redaction tests
• codex-rs/app-server/src/request_processors/thread_lifecycle.rs :: paginated resume

cd codex-rs
cargo test -p codex-app-server items_view
cargo test -p codex-app-server thread_resume_redaction
cargo test -p codex-app-server paginated_resume

12. 历史排查

遇到历史“缺工具调用”，先看 itemsView；遇到 resume 与本地 rollout 不一致，检查 redaction 和分页 cursor；遇到 live Turn 重复，检查是否同时使用 ThreadState 投影和 rollout 重建。V2 历史是视图，不是 rollout 的无损镜像。