CompactTask完整流程

本文回答一个具体问题：用户提交 Op::Compact 后，Codex 如何创建一个独立的 CompactTask，选择 TokenBudget、remote V2、remote V1 或 local 后端，最后把新上下文窗口安装到 Session 和 rollout 中。

本文不重新讲 RegularTask 的 sampling、工具循环和 mid-turn compaction；那些路径见 Turn主循环与退出条件 和 RegularTask完整流程。这里会反复区分三种容易混淆的压缩：

压缩类型	创建者	是否是独立 Task	主要入口	历史安装方式
手动 standalone	Op::Compact handler	是	CompactTask::run	独立 compaction turn，完成后 Task 收尾
pre-sampling auto	run_turn	否，inline	run_*_auto_compact_task	首个 sampling 前替换窗口
mid-turn auto	run_turn	否，inline	run_inline_*_auto_compact_task	当前 Turn 内继续 sampling 前替换窗口

1. Op到CompactTask

Op::Compact 不携带用户输入。submission loop 将它交给 handlers::compact，handler 创建新的 TurnContext，再用普通 Session::spawn_task 安装 CompactTask。因此它会替换正在运行的 Task；它 不是把一段“压缩提示词”塞进当前 RegularTask。

源码位置：codex-rs/core/src/session/handlers.rs :: compact

pub async fn compact(sess: &Arc<Session>, sub_id: String) {
    let turn_context = sess.new_default_turn_with_sub_id(sub_id).await;
    // spawn_task 负责替换旧 RunningTask，并把 CompactTask 置入 ActiveTurn。
    sess.spawn_task(Arc::clone(&turn_context), Vec::new(), CompactTask)
        .await;
}

spawn_task 的公共替换语义来自 Task抽象与生命周期：旧任务先按 Replaced 清理，新任务随后拥有自己的 Turn ID。 因此手动压缩的 TurnStarted、ContextCompaction item 和 TurnComplete 都属于新的 Task 生命周期。

这个状态图只描述 standalone CompactTask 的任务边界。Retrying 是 local 压缩请求内部的重试，不会创建 新的 CompactTask；HistoryInstalled 之后才进入 replace_compacted_history 的持久化顺序。

2. CompactTask 选择

后端选择完全集中在 tasks/compact.rs。优先级不是“provider 支持什么就先用什么”，而是先检查 TokenBudget feature；只有它关闭时，才读取 provider 的 remote_compaction 能力，再由 RemoteCompactionV2 feature 决定 V2 还是 legacy remote。

源码位置：codex-rs/core/src/tasks/compact.rs :: CompactTask::run

let _profile_guard = ctx.turn_timing_state.begin_compaction();
if ctx.config.features.enabled(Feature::TokenBudget) {
    // TokenBudget 是本地重建窗口，不请求摘要模型。
    crate::compact_token_budget::run_manual_compact_task(session, ctx).await?;
    return Ok(None);
}

let result = match ctx.provider.capabilities().remote_compaction {
    RemoteCompactionSupport::V2
        if ctx.config.features.enabled(Feature::RemoteCompactionV2) =>
    {
        emit_compact_metric(&session.services.session_telemetry, "remote_v2", true);
        crate::compact_remote_v2::run_remote_compact_task(session.clone(), ctx).await
    }
    RemoteCompactionSupport::V1 | RemoteCompactionSupport::V2 => {
        emit_compact_metric(&session.services.session_telemetry, "remote", true);
        crate::compact_remote::run_remote_compact_task(session.clone(), ctx).await
    }
    RemoteCompactionSupport::Unsupported => {
        emit_compact_metric(&session.services.session_telemetry, "local", true);
        let input = vec![UserInput::Text {
            text: ctx.config.compact_prompt.as_deref()
                .unwrap_or(crate::compact::SUMMARIZATION_PROMPT).to_string(),
            // 合成的压缩提示没有 UI text_elements，避免伪造用户文本范围。
            text_elements: Vec::new(),
        }];
        crate::compact::run_compact_task(session.clone(), ctx, input).await
    }
};

优先级可以写成下面的决策树：

RemoteCompactionSupport::V2 但 feature 未开启时会落入 legacy remote 分支。反过来，provider 不声明 remote capability 时，即使 V2 feature 开启也不会请求 remote endpoint。

当前源码还把 CompactedItem 的 window number、first/previous/current window ID 和 replacement history 作为恢复检查点；RolloutReconstruction 会按这些字段反向重建 history 与 auto-compact window。压缩不是 简单替换一段字符串，而是同时更新模型上下文、窗口身份和持久化基线。

3. Compaction 后端

3.1 TokenBudget

compact_token_budget::run_manual_compact_task 先发 standalone TurnStarted，捕获当前 Step 和 WorldState，然后调用 Session::start_new_context_window。这个分支不生成摘要文本，也不会发 server-side compact request；它用当前初始上下文重建历史，并更新 window IDs。

源码位置：codex-rs/core/src/compact_token_budget.rs :: run_manual_compact_task

let start_event = EventMsg::TurnStarted(TurnStartedEvent {
    turn_id: turn_context.sub_id.clone(),
    trace_id: turn_context.trace_id.clone(),
    started_at: turn_context.turn_timing_state.started_at_unix_secs().await,
    model_context_window: turn_context.model_context_window(),
    collaboration_mode_kind: turn_context.mode,
});
sess.send_event(&turn_context, start_event).await;

// standalone compact 自己捕获 Step，而不是复用上一个 RegularTask 的 Step。
let step_context = sess
    .capture_step_context(Arc::clone(&turn_context), &CancellationToken::new())
    .await?;
let world_state = Arc::new(sess.build_world_state_for_step(&step_context).await?);
run_compact_task_inner(&sess, &step_context, world_state, CompactionTrigger::Manual).await

start_new_context_window 生成初始上下文、TurnContextItem 和完整 WorldState baseline，再调用统一的 replace_compacted_history。测试 token_budget_context_uses_new_window_after_compaction 说明压缩请求不 访问 server-side compact endpoint，后续请求使用新的 window ID 且不再包含压缩前用户消息。

3.2 历史摘要请求

local 后端使用普通 Responses 请求。run_compact_task 会先发 TurnStarted，运行 PreCompact hooks， 把合成提示加入本次压缩请求的临时 history；模型返回后，从当前 Session history 收集用户消息并拼接 SUMMARY_PREFIX，再构造 replacement history。它不把整个压缩请求逐项记录为普通会话消息。

源码位置：codex-rs/core/src/compact.rs :: run_compact_task_inner_impl

let compaction_item = TurnItem::ContextCompaction(ContextCompactionItem::new());
sess.emit_turn_item_started(&turn_context, &compaction_item).await;

let mut history = sess.clone_history().await;
history.record_items(
    &[ResponseInputItem::from(input).into()],
    turn_context.model_info.truncation_policy.into(),
);

// 同一个 ModelClientSession 跨重试复用，保留 sticky routing 和 websocket 请求状态。
let mut client_session = sess.services.model_client.new_session();
loop {
    let turn_input = history.clone().for_prompt(&turn_context.model_info.input_modalities);
    let turn_input_len = turn_input.len();
    let attempt_result = drain_to_completed(
        &sess, turn_context.as_ref(), &mut client_session,
        &responses_metadata,
        &Prompt { input: turn_input, base_instructions: sess.get_base_instructions().await, ..Default::default() },
    ).await;
    match attempt_result {
        Ok(()) => break,
        Err(err) if matches!(err.details(), CodexErrorDetails::Interrupted | CodexErrorDetails::TurnAborted) => {
            return Err(err);
        }
        // 上下文超限时只移除最老 item，保留较新的对话并重新开始 retry 预算。
        Err(err) if matches!(err.details(), CodexErrorDetails::ContextWindowExceeded)
            && turn_input_len > 1 => {
            history.remove_first_item();
            retries = 0;
            continue;
        }
        Err(err) if retries < max_retries => {
            retries += 1;
            let delay = backoff(retries);
            sess.notify_stream_error(
                turn_context.as_ref(),
                format!("Reconnecting... {retries}/{max_retries}"),
                err,
            ).await;
            tokio::time::sleep(delay).await;
            continue;
        }
        Err(err) => {
            sess.send_event(turn_context, EventMsg::Error(err.to_error_event(None))).await;
            return Err(err);
        }
    }
}

上面省略了 SessionBudgetExceeded 和只剩一个输入 item 时的终止分支，但没有改变展示分支的顺序。测试 manual_compact_retries_after_context_window_error 让第一次请求返回 context_length_exceeded，断言第二次请求少一个最老 history item；这不是普通网络 retry，而是为压缩请求 减小输入后重新构造 prompt。

3.3 Remote V1

legacy remote 后端不在本地生成摘要。run_remote_compact_attempt 复制当前 history，必要时先把超出上下文 的 function-call output 改写为截断提示，然后调用 model_client.compact_conversation_history。服务端返回 的 new_history 经过 process_compacted_history 过滤和初始上下文插入后，进入统一安装边界。

3.4 Remote V2

V2 使用 Responses stream。collect_compaction_output 要求先收到 response.completed，并且输出中恰好有 一个 ResponseItem::Compaction；没有 completed 或数量不是 1 都是错误。成功后保留符合规则的 user、developer、 system 和有限长度的 agent message，再把 compaction item 放在末尾。

Remote 后端的格式边界

“remote”不是一个统一的响应格式。V1 返回完整 replacement history，V2 返回 compaction output， 二者都共享安装和持久化边界，但保留历史的算法不同。

4. 统一安装边界

四种后端只有在压缩结果准备好后才调用 Session::replace_compacted_history。这是从临时请求视图切换到 Session 真正可见历史的语义边界。

源码位置：codex-rs/core/src/session/mod.rs :: Session::replace_compacted_history

let items = Self::assign_missing_response_item_ids(Cow::Owned(items)).into_owned();
let compacted_item = CompactedItem {
    message: metadata.message,
    replacement_history: Some(items.clone()),
    window_number: Some(metadata.window_number),
    first_window_id: Some(metadata.window_ids.first_window_id.to_string()),
    previous_window_id: metadata.window_ids.previous_window_id.map(|id| id.to_string()),
    window_id: Some(metadata.window_ids.window_id.to_string()),
};

// 先在锁内替换 live history，并为新窗口建立完整 WorldState baseline。
{
    let mut state = self.state.lock().await;
    state.replace_history(items, reference_context_item.clone());
    if let Some(world_state) = world_state_baseline {
        let snapshot = world_state.snapshot();
        world_state_item = Some(WorldStateItem::full(snapshot.clone().into_value()));
        state.history.set_world_state_baseline(snapshot);
    }
}

// rollout 顺序反映恢复依赖：Compacted 建立替换历史，WorldState 建立 baseline，
// TurnContext 记录引用上下文；不能把它们混成普通 ResponseItem 追加。
self.persist_rollout_items(&[RolloutItem::Compacted(compacted_item)]).await;
if let Some(world_state_item) = world_state_item {
    self.persist_rollout_items(&[RolloutItem::WorldState(world_state_item)]).await;
}
if let Some(turn_context_item) = reference_context_item {
    self.persist_rollout_items(&[RolloutItem::TurnContext(turn_context_item)]).await;
}

安装后都会调用 recompute_token_usage，再发 ItemCompleted(ContextCompaction)。CompactedItem 保存 window number、first/previous/current window ID 和 replacement history，恢复时可以重放压缩检查点，而不是 只看到一条模糊的 summary 文本。

图中的继承箭头表示 RolloutItem 的枚举变体，而不是 Rust trait 继承。Session 先更新 live history，再 按 Compacted、WorldState、TurnContext 的顺序写入这些变体；它们共同构成恢复压缩检查点所需的信息。

5. Hooks失败

所有后端都会进入 PreCompact/PostCompact hook 生命周期，但失败传播并不完全相同：

阶段	后端行为	用户可观察结果
PreCompact 返回 Stopped	以 TurnAborted 结束	不安装 replacement history
后端被取消	local/remote 返回 Interrupted 或 TurnAborted	Task abort 生命周期，通常有 TurnAborted
remote V1/V2 普通失败	后端发送 Error 后返回错误	当前历史不换窗，Task 继续统一收尾
local 普通失败	local 发送 Error 后返回错误	与 remote 类似，但重试/忽略由 local 实现决定
TokenBudget 构造失败	CompactTask 直接 ? 返回	错误进入 Task 完成处理，不经过 local summary 请求

CompactTask::run 对非 TurnAborted 的 local/remote result 最终返回 Ok(None)，因此后端必须自己发送 普通 Error；否则外层看不到失败。TokenBudget 分支没有这个包装，构造 WorldState 或捕获 Step 失败会直接传播。

错误传播不能只看返回类型

这不是“所有压缩失败都会让 Task 返回 Err”。阅读错误行为时必须同时看 CompactTask::run 和具体后端； 只看后端函数的返回类型会误判外层事件语义。

该版本还保留一个重要测试边界：manual_compact_non_context_failure_retries_then_emits_task_error 被 #[ignore] 标记，并注明 non-context manual /compact failure 的行为等待后续 PR。它证明该问题已被测试 作者识别，但不是当前可宣称的已验证保证。

6. 分支阅读练习

实现最值得先运行的测试不是“请求成功”一项，而是下面这些互补断言：

1. manual_compact_uses_custom_prompt：配置自定义 compact_prompt 后，压缩请求使用自定义文本而不是默认 SUMMARIZATION_PROMPT。
2. manual_compact_retries_after_context_window_error：第一次压缩请求因上下文超限失败，第二次请求删除最老 history item 后重试；它不证明所有 provider 错误都可恢复。
3. manual_compact_emits_context_compaction_items：ItemStarted 与 ItemCompleted 使用同一个 ContextCompaction ID，并最终收到 TurnComplete。
4. token_budget_context_uses_new_window_after_compaction：TokenBudget 分支不访问远端 compact endpoint， 后续请求使用新的 window ID 且不包含旧用户消息。
5. remote_compact_replaces_history_for_followups：remote 返回的替换历史会成为后续 Turn 的 live history； 专门检查 rollout 持久化的 remote_compact_persists_replacement_history_in_rollout 虽仍带 #[ignore]， 但已单独以 --run-ignored 通过。它说明该 fixture 的 replacement history 会写入 rollout，不外推所有 remote failure 分支。

故障定位时可按这条链检查：

rg -n "Op::Compact|CompactTask::run|replace_compacted_history|run_remote_compact_task|run_manual_compact_task" \
  codex-rs/core/src

如果事件有 ContextCompaction 但下一请求仍带旧历史，先检查安装前的 backend response 和 replace_compacted_history；如果完全没有 compact item，再检查 handler 是否创建了新 Task、PreCompact hook 是否 stopped，以及 provider/feature 分支是否走到了预期后端。

7. 压缩分支验证

1. 给定 TokenBudget 已开启、provider 同时声明 V2，按源码说明实际选择哪个后端、是否发远端 compact 请求、 以及哪个函数安装新的 window IDs。
2. 第一次 local compact 请求返回 ContextWindowExceeded 时，解释为什么第二次请求可以少一个最老 history item， 但普通 server error 不一定走相同路径。
3. 读取 replace_compacted_history 的三个 rollout 写入点，说明恢复时为什么必须保留 Compacted、完整 WorldState baseline 和可选 TurnContext 的顺序。

继续阅读时，回到 Turn主循环与退出条件 对照 inline auto compaction， 再阅读 Turn中断与运行中注入 了解 standalone CompactTask 被替换时 Replaced 与 TurnAborted 的差异。