async fn run_auto_compact( // 自动压缩统一入口：pre-turn / mid-turn / 模型切换三处都调它    sess: &Arc<Session>, // 会话共享指针，历史、遥测、hook 全挂在上面    step_context: Arc<StepContext>, // 当前步上下文（模型 + 工具路由 + 设置快照），压缩请求按它构造    fallback_step_context: Option<Arc<StepContext>>, // 可选的"当前模型"备用上下文——上一模型压缩失败时拿它重试    client_session: &mut ModelClientSession, // 复用 turn 的连接会话，sticky routing / websocket 增量状态不丢    initial_context_injection: InitialContextInjection, // 新历史要不要重注入初始上下文（mid-turn 必须注入）    reason: CompactionReason, // 触发原因：用户请求 / 窗口满 / 模型降级 / comp hash 变化    phase: CompactionPhase, // 所处阶段：独立 turn / pre-turn / mid-turn，决定错误上报时机) -> CodexResult<()> {    let turn_context = &step_context.turn; // 取出 turn 级上下文读配置与 provider 能力    if turn_context.config.features.enabled(Feature::TokenBudget) { // TokenBudget 特性开启 → 压缩退化为"重置窗口"        crate::compact_token_budget::run_inline_auto_compact_task( // 完全不调模型，直接装一个新上下文窗口            Arc::clone(sess), step_context, initial_context_injection).await?;        return Ok(()); // 提前返回：这条路径不走任何摘要逻辑（第六节细讲）    }    match turn_context.provider.capabilities().remote_compaction { // 按 provider 声明的远程压缩能力三分支        RemoteCompactionSupport::V2 => { // V2：服务端直接产出结构化 Compaction item，客户端只负责组装保留历史            emit_compact_metric(&sess.services.session_telemetry, "remote_v2", /*manual*/ false); // 打点区分实现路径（自动触发 manual=false）            run_inline_remote_auto_compact_task_v2( // 注意把 fallback_step_context 传进去——失败可换当前模型重试                Arc::clone(sess), step_context, fallback_step_context,                client_session, initial_context_injection, reason, phase).await?;        }        RemoteCompactionSupport::Unsupported => { // provider 不支持远程压缩 → 本地摘要：把历史+提示词发给模型，取最后一条 assistant 消息当摘要            emit_compact_metric(&sess.services.session_telemetry, "local", /*manual*/ false); // 同样打点，遥测里能对比两条路径的成败率            run_inline_auto_compact_task( // 本地路径不需要 fallback（没有"上一模型"语义）                Arc::clone(sess), Arc::clone(turn_context), initial_context_injection, reason, phase).await?;        }    }    Ok(()) // 三条路径在此汇合：历史已替换、token 用量已重算、窗口号已推进}

三种触发场景（参数矩阵）

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 { // 否则按 provider 能力三分支（与自动路径一致）    RemoteCompactionSupport::V2 => { // V2 远程压缩；manual=true 打点，遥测里可区分手动/自动        emit_compact_metric(&session.services.session_telemetry, "remote_v2", /*manual*/ true);        crate::compact_remote_v2::run_remote_compact_task(session.clone(), ctx).await // 独立 turn：自己 capture step、发 turn_started    }    RemoteCompactionSupport::Unsupported => { // 本地摘要：合成一条用户消息发给模型        emit_compact_metric(&session.services.session_telemetry, "local", /*manual*/ true);        let input = vec![UserInput::Text {            text: ctx.config.compact_prompt.as_deref() // 允许用户在 config.toml 里覆盖摘要提示词                .unwrap_or(crate::compact::SUMMARIZATION_PROMPT).to_string(), // 默认用内置 SUMMARIZATION_PROMPT            text_elements: Vec::new(), // 合成消息没有 UI 元素范围需要保留（自动路径同款注释）        }];        crate::compact::run_compact_task(session.clone(), ctx, input).await // 本地摘要主流程在 core/src/compact.rs    }};

pub(crate) enum InitialContextInjection { // 控制压缩后的替换历史是否必须包含初始上下文    BeforeLastUserMessage { // mid-turn 专用：模型被训练成"摘要必须是历史最后一项"，所以初始上下文要插在最后一条真实用户消息之前        world_state: Arc<WorldState>, // 携带当前世界状态快照——新窗口的 baseline 要用它重建（第 13 讲讲过 WorldState）        step_context: Arc<StepContext>, // 渲染初始上下文需要完整步上下文（模型 + 设置 + 工具路由）    },    DoNotInject, // pre-turn / 手动：只装摘要并清掉 reference item，下一个正常 turn 会完整重注入初始上下文}

Err(e) if matches!(e.details(), CodexErrorDetails::ContextWindowExceeded) => { // 压缩请求自己也超窗 → 裁剪历史重试    if turn_input_len > 1 { // 还有可裁的（至少剩一条）        error!("Context window exceeded while compacting; removing oldest history item. Error: {e}"); // 记日志：正在丢最老的历史项        history.remove_first_item(); // 从头部删一条——保前缀缓存命中，同时保住最近消息完整        retries = 0; // 重置重试计数：裁剪是"换策略"，不该消耗普通重试预算        continue; // 带着更短的历史重新进循环发请求    }    sess.set_total_tokens_full(turn_context.as_ref()).await; // 只剩一条还超窗 → 标记窗口已满，让上层走兜底    sess.track_turn_codex_error(turn_context.as_ref(), &e); // 记错误遥测    if !matches!(compaction_metadata.phase(), CompactionPhase::PreTurn) { // pre-turn 阶段不在这报错——run_turn 会保住用户输入后统一上报        let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));        sess.send_event(&turn_context, event).await; // 其他阶段直接把错误事件推给 UI    }    return Err(e);}Err(e) => { // 其余瞬时错误（网络抖动 / 5xx / 限流等）    if retries < max_retries { // 还有重试预算        retries += 1;        let delay = backoff(retries); // 指数退避，避免打爆服务端        sess.notify_stream_error(turn_context.as_ref(), format!("Reconnecting... {retries}/{max_retries}"), e).await; // UI 提示"重连中 x/y"        tokio::time::sleep(delay).await;        continue; // 同一份历史原样重试    } else { // 预算耗尽 → 上报并失败（pre-turn 同样延迟上报）        sess.track_turn_codex_error(turn_context.as_ref(), &e);        if !matches!(compaction_metadata.phase(), CompactionPhase::PreTurn) {            let event = EventMsg::Error(e.to_error_event(/*message_prefix*/ None));            sess.send_event(&turn_context, event).await;        }        return Err(e);    }}

let mut selected_messages: Vec<CompactedUserMessage> = Vec::new(); // 将被保留进新历史的用户消息if max_tokens > 0 { // 预算即 COMPACT_USER_MESSAGE_MAX_TOKENS（20k）    let mut remaining = max_tokens; // 剩余 token 预算，从最新消息往回扣    for message in user_messages.iter().rev() { // 倒序遍历：优先保留离摘要最近的用户消息        if remaining == 0 { break; } // 预算耗尽 → 更老的消息全部丢弃（信息已在摘要里）        let tokens = approx_token_count(&message.message); // 粗略 token 估算，不追求精确        if tokens <= remaining { // 整条放得下            selected_messages.push(message.clone());            remaining = remaining.saturating_sub(tokens); // 扣减；saturating 防下溢（i64 不会变负）        } else { // 这条太大 → 截断后保留，且只保留"尾部"（离摘要最近的部分）            let truncated = truncate_text(&message.message, TruncationPolicy::Tokens(remaining));            selected_messages.push(CompactedUserMessage { id: message.id.clone(), message: truncated, .. }); // id 保留——回滚时能关联 thread-owned 证据            break; // 截断一条后预算必然耗尽，直接结束        }    }    selected_messages.reverse(); // 恢复时间顺序再写入历史（模型看到的仍是"旧→新"）}

let mut history = sess.clone_history().await; // COW 克隆当前历史——压缩过程不碰活状态，失败可整体丢弃let base_instructions = sess.get_prompt_base_instructions().await; // system instructions 参与 token 估算（它也在窗口里）let (rewritten_outputs, estimated_deleted_tokens) =    trim_function_call_history_to_fit_context_window( // 预裁剪：把超大的工具输出改写/删掉，让请求先塞进窗口        &mut history, turn_context.as_ref(), &base_instructions);if rewritten_outputs > 0 { info!(rewritten_outputs, "rewrote history outputs before remote compaction v2"); } // 有裁剪就记日志（遥测里能算出删了多少 token）input.push(ResponseItem::CompactionTrigger {}); // 关键：在输入末尾追加 CompactionTrigger 标记——这就是 V2 协议的"压缩请求"信号let prompt = Prompt {    input, // 完整历史（预裁剪后）作为输入，服务端据此生成摘要 + 决定保留什么    tools: tool_router.model_visible_specs(), // 带上可见工具规格：服务端需要知道历史里那些 function call 是什么    parallel_tool_calls: true, // 与正常 turn 请求形状保持一致，复用同一套流式管线    base_instructions, // system instructions    output_schema: None, // 不约束结构化输出——靠 Compaction item 类型本身承载结果    output_schema_strict: true, // schema 为 None 时此标志无实际作用（保持字段完整）    cyber_access_program: turn_context.cyber_access_program, // 透传访问计划标记};

while let Some(event) = stream.next().await { // 消费响应流直到 Completed    match event? {        ResponseEvent::OutputItemDone(item) => { // 每个输出 item 完成事件            output_item_count += 1; // 统计总输出数（出错时用于诊断"到底吐了什么"）            if let ResponseItem::Compaction { .. } = item { // 只关心 Compaction 类型的 item                compaction_count += 1; // 数出现了几次                if compaction_output.is_none() { // 第一个作为候选结果收下（后面还有就违反契约了）                    compaction_output = Some(item);                }            }        }        ResponseEvent::Completed { response_id, token_usage, .. } => { // 流结束事件，携带响应 id 与用量            completed_response_id = Some(response_id); // 存下来写进 CompactedItem 检查点（可回溯服务端原始请求）            completed_token_usage = token_usage; // 摘要 token / 缓存命中数都从这里取，喂给分析事件            break; // Completed 之后不再消费        }        _ => {} // 其余事件（reasoning、rate limit 等）在此路径忽略    }}if compaction_count != 1 { // 协议硬契约：必须恰好一个 Compaction item    return Err(CodexErr::Fatal(format!(        "remote compaction v2 expected exactly one compaction output item, got {compaction_count} from {output_item_count} output items"))); // 多一个少一个都是 Fatal——宁可失败，也不能把错误历史装进会话}

pub(crate) const RETAINED_MESSAGE_TOKEN_BUDGET: usize = 64_000; // V2 保留消息总预算：64k tokenconst MAX_RETAINED_AGENT_MESSAGE_TOKENS: i64 = 10_000; // 单条 agent 消息上限：10k（防止一条巨型回复吃掉全部预算）// Compact attempts can run much longer than normal turns, so keep the per-transport retry budget smaller.const MAX_REMOTE_COMPACTION_V2_STREAM_RETRIES: u64 = 2; // V2 流重试只给 2 次——压缩请求耗时远超普通 turn，重试要更克制let retained = v2_history_item_groups(prompt_input) // 把原始输入按来源（用户消息 / 工具调用组等）聚成"保留单元"    .filter(|group| is_retained_for_remote_compaction_v2(&group.source, retain_client_developer_messages)) // 过滤：哪些来源值得跨压缩存活（客户端 developer 消息是否保留由 feature 决定）    .flat_map(HistoryItemGroup::into_items) // 展平回 item 列表    .collect::<Vec<_>>();let mut retained = truncate_retained_messages(retained, RETAINED_MESSAGE_TOKEN_BUDGET, image_budget); // 64k 总预算内从新到旧截断；图片走独立预算（RetainedImageBudget）let retained_image_count = retained.iter().map(|envelope| retained_input_image_count(&envelope.item)).sum::<usize>(); // 统计存活图片数 → 分析事件retained.push(ResponseItemEnvelope::new(compaction_output)); // 最后追加服务端 Compaction 摘要——模型被训练成"它必须是最后一项"

pub(crate) fn should_retry_with_current_model(error: &CodexErr) -> bool { // 判断失败是否"模型相关"、值得换当前模型重试    matches!(error.details(),        CodexErrorDetails::InvalidRequest(_)          | // 请求被拒（如旧模型不接受这种历史形状）——换个模型可能就接受        CodexErrorDetails::UnexpectedStatus(_)        | // 意外 HTTP 状态码        CodexErrorDetails::ContextWindowExceeded     | // 窗口超限——当前模型窗口可能更大，或裁剪后能装下        CodexErrorDetails::UsageLimitReached(_)       | // 该模型的配额用尽——换模型绕开        CodexErrorDetails::ServerOverloaded          | // 服务端过载（可能是该模型专属的拥塞）        CodexErrorDetails::InternalServerError        | // 5xx        CodexErrorDetails::RetryLimit(_))             // 重试预算耗尽——以上全是"换个模型可能成功"的错误}

let outcome = if fallback_error.is_none() { "succeeded" } else { "failed" }; // 兜底尝试本身成功还是失败session_telemetry.counter(    "codex.compaction.model_fallback", // 模型降级事件专用计数器——线上能直接看到"多少压缩靠换模型救回来"    /*inc*/ 1,    &[("reason", reason_tag),           // 为什么压缩：user_requested / context_limit / model_downshift / comp_hash_changed      ("implementation", implementation_tag), // 哪条实现：responses / responses_compaction_v2      ("outcome", outcome)],            // succeeded / failed——三个维度交叉，能定位"哪种原因下降级最有效");

async fn run_compact_task_inner(    sess: &Arc<Session>, step_context: &Arc<StepContext>, world_state: Arc<WorldState>, trigger: CompactionTrigger,) -> CodexResult<()> {    let turn_context = &step_context.turn; // 取 turn 上下文（hook 需要）    let pre_compact_outcome = run_pre_compact_hooks(sess, turn_context, trigger).await; // PreCompact hook 照常跑——用户脚本可以否决这次压缩    match pre_compact_outcome {        PreCompactHookOutcome::Continue => {} // 放行        PreCompactHookOutcome::Stopped => return Err(CodexErr::TurnAborted), // hook 说停 → 整个 turn 中止（fail loud）    }    let compaction_item = TurnItem::ContextCompaction(ContextCompactionItem::new()); // 发 ContextCompaction item——UI / rollout 看到的生命周期与其他路径完全一致    sess.emit_turn_item_started(turn_context, &compaction_item).await;    sess.start_new_context_window(step_context, world_state) // 核心动作：装新上下文窗口（零模型调用、零摘要）        .await;    sess.emit_turn_item_completed(turn_context, compaction_item).await;    let post_compact_outcome = run_post_compact_hooks(sess, turn_context, trigger).await; // PostCompact hook 同样可以否决    if let PostCompactHookOutcome::Stopped = post_compact_outcome { return Err(CodexErr::TurnAborted); }    Ok(())}

pub(super) fn advance(&mut self) -> (u64, AutoCompactWindowIds) { // 压缩成功 → 窗口号 +1，签发新窗口身份    self.window_number = self.window_number.saturating_add(1); // saturating：防溢出（理论上不会发生，但记账代码不留隐患）    self.ids.previous_window_id = Some(self.ids.window_id); // 链式结构：旧窗口变成"上一个"——resume / fork 回滚靠这条链定位    self.ids.window_id = Uuid::now_v7(); // UUIDv7 时间有序：跨机器、跨进程都能按 id 排序，不依赖本地时钟单调性    self.new_context_window_requested = false; // 清掉"请求新窗口"的挂起标志（模型通过 new_context 工具请求的那种）    self.token_budget_reminder_delivered = false; // 每个窗口有且仅有一次"预算告警"机会——one-shot claim    self.auto_compact_fallback_delivered = false; // fallback 提示同理，每窗口一次，防止刷屏    (self.window_number, self.ids) // 返回给调用方写进 CompactedItem 检查点（第八节）}pub(super) fn ensure_server_observed_prefill_from_usage(&mut self, usage: &TokenUsage) { // 用服务端首个用量样本记录"窗口基线"    if matches!(self.prefill_input_tokens, Some(AutoCompactWindowPrefill::ServerObserved(_))) { return; } // 已有实测值 → 绝不用估算覆盖（实测优先）    self.prefill_input_tokens = Some(AutoCompactWindowPrefill::ServerObserved(usage.input_tokens.max(0))); // input tokens 即本窗口的"前缀基线"，后续增量 = 总用量 - 基线}pub(super) fn set_estimated_prefill(&mut self, tokens: i64) { // resume / recompute 场景还没有服务端样本 → 先用本地估算占位    if matches!(self.prefill_input_tokens, Some(AutoCompactWindowPrefill::ServerObserved(_))) { return; } // 实测值优先级更高，估算不能顶掉它    self.prefill_input_tokens = Some(AutoCompactWindowPrefill::Estimated(tokens.max(0))); // 打上 Estimated 标记——之后拿到实测样本会被替换

let mut compacted_item = CompactedItem { // 压缩检查点——持久化进 rollout，resume / fork 全靠它    message: metadata.message, // 本地路径存摘要文本；远程 V2 为空串（摘要在 Compaction item 里）    replacement_history: Some(items.clone()), // 压缩后的完整替换历史——live 与 persisted 必须逐字节一致（前面已补齐缺失 id）    retained_context: None, // 稍后填入当前"保留上下文"快照：用户授权过的事实跨压缩存活（第 13 讲的核心机制）    guardian_history: None, // Guardian thread-owned 模式下的独立历史检查点    mcp_resource_origins: self.services.mcp_runtime.resource_origin_checkpoint(), // MCP 资源来源检查点——resume 后能校验资源出处没变    window_number: Some(metadata.window_number), // 这是第几次压缩（0-based）    first_window_id: Some(metadata.window_ids.first_window_id.to_string()), // 线程首个窗口 id，跨所有窗口不变    previous_window_id: metadata.window_ids.previous_window_id.map(|id| id.to_string()), // 上一个窗口 → 与 first/current 组成回滚链    window_id: Some(metadata.window_ids.window_id.to_string()), // 当前窗口 id（UUIDv7）    compaction_response_id: metadata.compaction_response_id, // 服务端响应 id——出事故可回溯到那次压缩的原始请求    latest_token_usage_record: self.state.lock().await.latest_token_usage_record.clone(), // 压缩时刻的 token 用量快照（审计基线）};

Mid-turn 自动压缩（V2 路径）

▼

▼

▼

▼

▼

▼