/// Takes initial turn input and runs a loop where, at each sampling request, // 文档注释：run_turn 拿到本轮输入后进入循环，每次采样请求模型要么要工具、要么回消息/// the model replies with either: // ——回复只有两种形态/// - requested function calls // 形态一：函数调用（harness 执行完把结果喂回去）/// - an assistant message // 形态二：助手消息（记入历史，本轮结束）pub(crate) async fn run_turn( // Turn 的执行入口：一个 turn = "用户输入 → 模型/工具多轮往返 → 最终回复"    sess: Arc<Session>, // 会话本体（共享状态、事件总线、input_queue），Arc 因为要跨任务传递    turn_context: Arc<TurnContext>, // 本轮上下文快照：配置、审批策略、provider、sub_id    mut input: Vec<TurnInput>, // 用户输入列表（文本/图片/@提及等）；mut 因为 guardian 路径会改写它    mcp_startup_requirements: &mut McpStartupRequirements, // MCP 启动需求跨 turn 保留：显式 @ 的 server/plugin 必须拉起    prewarmed_client_session: Option<ModelClientSession>, // 预热好的模型会话（缓存 WebSocket/粘性路由），没有就新建    cancellation_token: CancellationToken, // 取消令牌：用户中断时所有 await 点都能快速退出) -> CodexResult<Option<String>> { // 返回本轮最后一条助手消息；None = 被 hook 拦截或出错提前结束

let mut client_session = // 复用预热会话，否则新建——WebSocket + sticky routing 状态是 turn 级缓存    prewarmed_client_session.unwrap_or_else(|| sess.services.model_client.new_session()); // 有预热会话就复用，没有才新建——省掉 WebSocket 握手开销if let Err(err) = run_pre_sampling_compact( // 采样前先压缩：历史超预算就先 compact，避免第一轮请求直接爆窗口    &sess, // 传会话本体：compact 要读写历史和事件总线    &turn_context, // 传本轮上下文快照：读配置里的压缩阈值/模型信息    &mut client_session, // 可变借用：remote compact 可能切换底层连接状态    &cancellation_token, // 取消令牌：用户中断时 compact 立即停止).await{ // ——压缩失败分支开始    // Compaction runs before the new input is recorded, so preserve it on every failure. // 压缩发生在"新输入落库之前"，失败时必须先把用户输入记下来，否则丢消息    run_hooks_and_record_inputs(&sess, &turn_context, &turn_context.capture_current_model_info(), &input, PersistContext::Standard).await;    if matches!(err.details(), CodexErrorDetails::TurnAborted) { // 取消类错误直接上抛，不当普通错误广播        return Err(err);    }    let error = err.to_codex_protocol_error(); // 其余失败：先走 turn 错误生命周期（遥测/状态机）    sess.emit_turn_error_lifecycle(turn_context.as_ref(), error.clone()).await;    sess.send_event(turn_context.as_ref(), EventMsg::Error(err.to_error_event(message_prefix))).await; // 再广播 Error 事件——顺序保证 hook 先跑完，防止客户端抢着往"还在保 prompt 的 turn"里塞新输入    return Ok(None); // 本轮到此为止：无最终消息，但用户可继续对话}

let mut last_agent_message: Option<String> = None; // 记录本轮最后一条助手消息，作为 run_turn 的返回值let mut stop_hook_active: bool = false; // Stop hook 是否已生效过——防止"hook 要求继续 → 再触发 hook"死循环let turn_diff_tracker = Arc::new(tokio::sync::Mutex::new( // TurnDiffTracker：跟踪本轮文件改动供 UI 展示 diff；用户视角一个 turn 内共享    TurnDiffTracker::with_environment_display_roots(display_roots), // 用各环境的展示根目录初始化 diff 跟踪器)); // ——Mutex 包裹：工具执行与 UI 读取并发访问同一 tracker// `ModelClientSession` is turn-scoped and caches WebSocket + sticky routing state, so we reuse // 客户端会话是 turn 级缓存，重试间复用同一实例// one instance across retries within this turn. // ——所以整个 loop 里 client_session 不重建let mut next_step_context = Some(first_step_context); // 下一步的 StepContext（工具/权限/MCP 快照）；Some=复用，None=重新捕获let mut guardian_budget_compacted = false; // Guardian 预算压缩只允许一次——防止无效压缩无限循环loop { // ===== agent loop 主循环：采样 → 分发输出 → 决策是否再来一轮 =====    let pending_input = if can_drain_pending_input { // 只有"允许排空"时才取队列里用户中途塞进来的新输入（steer）        sess.input_queue.get_pending_input(&sess.active_turn).await.0 // 取出并清空该 turn 的待处理输入    } else {        Vec::new() // 否则留队：比如 compact 之后要先让模型把工具续跑完，不能插话打断    };    if run_hooks_and_record_inputs(&sess, &turn_context, &turn_context.capture_current_model_info(), &pending_input, PersistContext::Standard).await { // 排空出来的输入同样要过 hook：拒绝则 break 结束本轮        break;    }

let max_retries = turn_context.provider.info().stream_max_retries(); // 重试上限来自 provider 配置——不同后端容忍度不同let mut retry_state = ResponsesStreamRetryState::default(); // 重试状态机：记录已试过的退避/切换策略，避免重复同一动作let mut initial_input = Some(input); // 首次请求用调用方准备好的输入；Some→None 一次性消费let mut original_input = None; // 记住"原始 prompt 输入"，采样结束后回写历史（保证记录的是真实发过的内容）let mut executed_tool_calls_by_output = HashMap::new(); // output → 已执行工具调用的映射：重试时把旧结果重新挂到新请求上loop { // ===== 流式请求 + 可恢复错误恢复循环 =====    turn_context.extension_data.remove::<codex_api::ResponseId>(); // 清掉上一响应的 ResponseId——重试后新工具调用不能归属到旧的 response，否则审计/回放错乱    let prompt_input = if let Some(input) = initial_input.take() { // 第一次：用现成输入        input    } else { // 重试：从历史重新拼 prompt（历史里已含上一轮的工具结果）        sess.clone_history().await.for_prompt(&step_context.settings.model_info.input_modalities)    };    let mut prompt_input = prompt_input;    sess.services.executed_tool_calls.attach_to_prompt(&mut prompt_input, &mut executed_tool_calls_by_output); // 把已执行的工具调用结果重新附着到重试请求上——模型看到的上下文不丢

let mut in_flight: FuturesOrdered<InFlightFuture<'static>> = FuturesOrdered::new(); // 在途工具调用队列：FuturesOrdered 保证按发出顺序收集结果，历史顺序不乱let mut needs_follow_up = false; // 本轮采样是否还需要"下一轮请求"（有工具调用/服务端要求续跑）let outcome: CodexResult<SamplingRequestResult> = loop { // ===== SSE 事件消费主循环：一直读到 response.completed =====    let event = match stream.next().instrument(trace_span!(parent: &receiving_span, "receiving")).or_cancel(&cancellation_token).await { // or_cancel：取消令牌一触发，整个流立即中断        Ok(event) => event,        Err(codex_async_utils::CancelErr::Cancelled) => { break Err(CodexErr::TurnAborted); } // 被取消 → TurnAborted，上层识别后直接返回（不当普通错误广播）    };    let event = match event { // 流结束三态：正常事件 / 传输错误 / 提前断流        Some(Ok(event)) => event,        Some(Err(err)) => break Err(err), // 网络/协议错误上抛，交给外层重试状态机判断是否可恢复        None => { break Err(CodexErr::Stream("stream closed before response.completed".into())); } // 服务端没发 completed 就断流 = 不完整响应，必须报错而不是静默收尾    };    sess.services.session_telemetry.record_responses(&handle_responses, &event); // 每个事件都过遥测：OTEL span 上挂 token/工具字段（第 44 讲展开）

match ToolRouter::build_tool_call(item.clone()) { // 每个完成的输出项先问路由：这是工具调用吗？    Ok(Some(call)) => { // ——是：记日志、落库、把执行 future 挂进在途队列（不阻塞流消费）        call_trace::received(ctx.sess.thread_id, &call.tool_name, &call.call_id, call_trace::Receipt::ModelTurn(&ctx.step_context.turn.sub_id)); // 调用链追踪：记录"模型第 N 轮发起了这个工具"，供审计回放        ctx.sess.input_queue.accept_mailbox_delivery_for_current_turn(&ctx.sess.active_turn, &ctx.step_context.turn.sub_id).await; // 有工具要跑 = 本轮还没结束，放行 mailbox（多智能体邮件）投递        record_completed_response_item(ctx.sess.as_ref(), ctx.step_context.as_ref(), &item).await; // 立即持久化工具调用项——即使 turn 随后被取消，历史/rollout 也保持完整        let cancellation_token = ctx.cancellation_token.child_token(); // 派生子令牌：取消 turn 时正在跑的工具一起停        let tool_future: InFlightFuture<'static> = Box::pin( // 工具执行是异步 future，先挂起不 await——流里还有后续事件要处理            ctx.tool_runtime.clone().handle_tool_call(call, cancellation_token),        );        output.needs_follow_up = true; // 有工具调用 ⇒ 采样结束后必须再发一轮请求把结果喂回模型        output.tool_future = Some(tool_future);    }    Ok(None) => { // ——不是工具：消息/推理等，走"非工具项"收尾（finalize → emit_turn_item_completed → 记历史）

let should_roll_over = needs_follow_up && (sess.take_new_context_window_request().await || token_limit_reached); // 需要续跑 且（服务端要求开新窗口 或 本地 token 到限）⇒ 先压缩再续let allow_auto_compact_fallback = !should_roll_over && !token_limit_reached; // 不满足 rollover 但还有余量 ⇒ 允许"自动压缩兜底"路径if should_roll_over { // ===== 上下文窗口滚动：MidTurn 压缩，把历史压到预算内 =====    if let Err(err) = run_auto_compact(&sess, Arc::clone(&step_context), /*fallback_step_context*/ None, &mut client_session, InitialContextInjection::BeforeLastUserMessage { world_state: Arc::clone(&world_state), step_context: Arc::clone(&step_context) }, CompactionReason::ContextLimit, CompactionPhase::MidTurn).await { // 压缩失败：取消直接上抛；其余走错误生命周期后结束本轮        if matches!(err.details(), CodexErrorDetails::TurnAborted) { return Err(err); }        let error = err.to_codex_protocol_error();        sess.emit_turn_error_lifecycle(turn_context.as_ref(), error.clone()).await;        return Ok(None);    }    can_drain_pending_input = !model_needs_follow_up; // 压缩后若模型还要续跑，先让它把活干完再排空用户插话——顺序敏感    continue; // 回到 loop 顶部：用压缩后的历史重新发起采样请求}

一个 turn 的完整生命周期（run_turn L163-783）

▼

▼

▼

▼

▼