pub struct CodexThread { // 线程门面：外部调用方（TUI/app-server/子代理）看到的只有它，看不到 Session    pub(crate) session: Arc<Session>, // 真正的运行时在 Session；thread 只持 Arc 引用、不拥有其生命周期    pub(crate) io: SessionIo, // 双向通道端点：提交 op / 接收事件都走这里（下一节细拆）    pub(crate) session_source: SessionSource, // 注册来源决定访问权与生命周期钩子（用户/内部/Guardian）    session_configured: SessionConfiguredEvent, // 启动时首个事件的快照，供外部查询线程配置    rollout_path: Option<PathBuf>, // rollout 录制文件路径；ephemeral 或未落盘时为 None    out_of_band_elicitations: Mutex<OutOfBandElicitations>, // 带外 elicitation 状态（计数+注册），用互斥锁保护    _diagnostics_guard: GaugeGuard, // drop guard：线程被丢弃时自动把 LIVE_THREADS 指标减一}impl CodexThread { // 门面方法组：所有对外入口都挂在这里    pub(crate) fn new( // 内部构造器：只有 ThreadManager 能创建线程（注册是它的职责）        session: Arc<Session>, // 运行时本体，与 SessionIo 等持有者共享同一份 Arc        io: SessionIo, // 通道端点；所有发送方 drop 掉即终止 session loop        session_configured: SessionConfiguredEvent, // 首事件必须是它，否则 finalize_thread_spawn 会报错        rollout_path: Option<PathBuf>, // 录制路径由外部传入（来自 session 配置）        session_source: SessionSource, // 来源决定它对客户端是否可见、能否被移除    ) -> Self { // 构造并返回门面实例        Self { // 逐字段初始化            session, // 原样透传运行时引用            io, // 原样透传通道端点            session_source, // 原样透注册来源            session_configured, // 原样透传配置快照            rollout_path, // 原样透传录制路径            out_of_band_elicitations: Mutex::new(OutOfBandElicitations::default()), // elicitation 状态从零开始            _diagnostics_guard: LIVE_THREADS.track(), // 登记进存活线程指标，drop 时自动释放        } // 结构体字面量结束    } // new() 结束} // impl 块结束

pub(crate) struct SessionIo { // 会话双向流的通道端点，与运行时状态刻意分离    pub(crate) tx_sub: Sender<Submission>, // 发送端：所有 op（用户输入/工具结果/控制命令）都走这条 channel    pub(crate) rx_event: Receiver<Event>, // 接收端：turn 进度/完成等事件从这里拉取    pub(crate) agent_status: watch::Receiver<AgentStatus>, // agent 最新状态，watch channel 可订阅    pub(crate) session_loop_termination: SessionLoopTermination, // 共享 future：多个调用方都能等待 loop 退出}pub async fn submit(&self, op: Op) -> CodexResult<String> { // 最基础入口：提交一个 op，返回 submission id    self.io.submit(op).await // 直接转发给 SessionIo；thread 自身不持有任何业务逻辑}

pub async fn start_or_steer_turn( // 最通用入口：空闲就开新 turn，忙碌就把输入 steer 进当前 turn    &self, // 门面是 Arc 共享的，方法不要求独占    request: TurnInputRequest, // 用户输入（文本 / 命名函数调用结果）) -> CodexResult<TurnInputSubmission> { // 结果是枚举：Started / Steered / NotSubmitted{reason}    self.submit_turn_input_with_mode(request, TurnInputMode::StartOrSteer).await // 以 StartOrSteer 模式委托统一分发器}async fn submit_turn_input_with_mode( // 统一分发：容量检查 + 通道提交，三模式的公共底座    &self, // 同上：共享门面    request: TurnInputRequest, // 与上面相同的输入载荷    mode: TurnInputMode, // StartOrSteer / StartIfIdle / Steer{expected_turn_id}) -> CodexResult<TurnInputSubmission> { // 返回统一的结果枚举    if !matches!(mode, TurnInputMode::Steer { .. }) { // steer 不需要容量检查（它不启动新 turn）        self.session.services.agent_control.ensure_execution_capacity_for_turn_start(self).await?; // 多智能体预算：先确认有执行配额才允许开新 turn    } // if 结束    self.io.submit_turn_input(request, mode).await // "到底 start 还是 steer"由 session loop 单线程裁决，门面不猜} // submit_turn_input_with_mode 结束

pub async fn suspend_turn_and_shutdown(&self) -> CodexResult<SuspendTurnOutcome> { // 挂起未完成的 root turn 并关闭 session，供另一个 worker 接管恢复    if self.session_source.is_non_root_agent() { // 子代理没有这个权利：只有拥有它的 root thread 能发起挂起        return Err(CodexErr::UnsupportedOperation("turn suspension requires the owning root thread".to_string())); // fail loud：明确报错而不是静默降级    } // 子代理守卫结束    let (reply, result) = oneshot::channel(); // 一次性应答通道：整个交接过程的所有权交给 session    self.io.tx_sub.send(Submission { // 把控制 op 送进主循环；即使本调用方断开，session 也会继续处理它        id: new_submission_id(), // submission id，用于追踪这次控制请求        op: Op::SuspendTurnAndShutdown { reply }, // op 里带着通道的一半（reply），session 完成后用它回话        trace: current_span_w3c_trace_context(), // 附带 W3C trace context，跨进程关联同一条链路        parent_turn_id: None, // 挂起是线程级动作，不绑定任何具体 turn        root_turn_id: None, // 同上：由 session loop 自己定位要挂起的 root turn    }).await.map_err(|_| CodexErr::Fatal("thread session has stopped".to_string()))?; // channel 已关 = loop 已死，直接报 fatal    let outcome = result.await.map_err(|_| CodexErr::Fatal("thread suspension reply was lost".to_string()))??; // 等 session 回话（停执行 + flush 历史 + 关闭 writer）    if matches!(&outcome, SuspendTurnOutcome::Suspended { .. }) { // 挂起成功才需要等 loop 真正退出        self.io.session_loop_termination.clone().await; // await 共享终止 future：确认资源全部释放后才返回调用方    } // if 结束    Ok(outcome) // 返回结果（Suspended / Refused{reason}）给接管方决策} // suspend_turn_and_shutdown 结束

pub(crate) struct ThreadManagerState { // 管理器的共享状态；用 Arc 包一层，让 AgentControl 能弱引用它而不必处处传 Arc<&Self>    threads: Arc<RwLock<HashMap<ThreadId, Arc<CodexThread>>>, // 存活线程表：id → 门面；读写锁分离热读与冷写    thread_created_tx: broadcast::Sender<ThreadId>, // 广播通道：新线程诞生时通知订阅方（TUI 列表 / app-server）    thread_id_generator: ThreadIdGenerator, // ID 生成器（支持 reserve_thread_id 预占号）    auth_manager: Arc<AuthManager>, // 认证凭证全线程共享一份    models_manager: SharedModelsManager, // 模型目录与 fallback 策略也共享    git_root_discovery: Arc<GitRootDiscovery>, // Git root 发现结果在管理器级缓存，避免每个线程重复扫盘    environment_manager: Arc<EnvironmentManager>, // 环境（cwd / workspace roots）统一管理    starting_mcp_runtimes: std::sync::Mutex<Vec<std::sync::Weak<AtomicBool>>>, // 正在启动的 MCP runtime：用 Weak 引用，死线程不阻塞清理    skills_service: Arc<HostSkillsService>, // Skills 服务共享    plugins_manager: Arc<PluginsManager>, // 插件管理器共享    mcp_manager: Arc<McpManager>, // MCP 连接管理器（隔离会话会另建独立实例，见第七节）    code_mode_session_provider: Arc<dyn CodeModeSessionProvider>, // Code Mode 会话提供者（trait object，宿主注入）    extensions: Arc<ExtensionRegistry<Config>>, // 扩展注册表；隔离线程改用空注册表    user_instructions_provider: Arc<dyn UserInstructionsProvider>, // 全局指令提供者（宿主负责抓取与缓存）    image_store: Arc<dyn AttachmentStore>, // 图片附件存储    thread_store: Arc<dyn ThreadStore>, // 持久化层：JSONL rollout + SQLite 元数据    agent_graph_store: Option<Arc<dyn AgentGraphStore>>, // 多智能体图存储（可选，依赖 state db）    attestation_provider: Option<Arc<dyn AttestationProvider>>, // workload identity 证明（可选）    external_time_provider: Option<Arc<dyn TimeProvider>>, // 外部时间源（可选，测试/特殊宿主用）    session_source: SessionSource, // 本管理器实例的默认注册来源（cli / app-server / internal）    installation_id: String, // 安装标识，遥测用    analytics_events_client: Option<AnalyticsEventsClient>, // 分析客户端（可被配置禁用）    ops_log: Option<SharedCapturedOps>, // 测试模式：捕获提交的 op 供断言} // ThreadManagerState 结束

if let InitialHistory::Resumed(resumed) = &initial_history { // resume 路径：先查这个线程是否已经在表里活着    let mut threads = self.threads.write().await; // 拿写锁做"检查+去重"，原子化防止两个 resume 竞态出两份运行时    if let Some(thread) = threads.get(&resumed.conversation_id).cloned() { // 表里命中：这个 id 已有运行时        if thread.is_running() { // 还在跑：直接返回它，而不是再造一个（resume 幂等）            if matches!(thread.session_source, SessionSource::Internal(InternalSessionSource::Guardian)) { // Guardian 审查器是特例                return Err(CodexErr::InvalidRequest("cannot resume a live Guardian reviewer; use thread/read to inspect it".to_owned())); // 活着的 Guardian 归父线程池所有：拒绝交出，防止两个 owner 抢一个 writer            } // Guardian 特判结束            if let Some(requested_rollout_path) = resumed.rollout_path.as_deref() && thread.rollout_path().as_deref() != Some(requested_rollout_path) { // 请求的 rollout 路径与在跑的不一致                return Err(CodexErr::InvalidRequest(format!("thread {} is already running with a different rollout path", resumed.conversation_id))); // 两边写不同文件，绝不能混用——直接报错            } // 路径校验结束            return Ok(NewThread { thread_id: resumed.conversation_id, session_configured: thread.session_configured(), thread }); // 原样返回现有运行时（幂等成功）        } // "已在跑"分支结束        threads.remove(&resumed.conversation_id); // 没在跑（僵尸条目）：先移除，让下面的流程重建    } // 表内命中分支结束} // resume 去重块结束

let thread_id = session.thread_id(); // 线程 id 由 session 自己决定（或来自预占的 reserved_thread_id）let event = io.next_event().await?; // 等新 loop 的第一个事件——这是启动"握手"，不通过就不注册let session_configured = match event { // 校验首事件的形状    Event { id, msg: EventMsg::SessionConfigured(session_configured) } if id == INITIAL_SUBMIT_ID => session_configured, // 必须是 SessionConfigured 且来自初始提交；否则视为协议违规    _ => return Err(CodexErr::SessionConfiguredNotFirstEvent), // 其他任何形状 = 启动失败，fail fast 不留半成品}; // 握手校验结束{ // 注册块（作用域锁：离开即释放写锁）    let mut threads = self.threads.write().await; // 拿写锁做重复检查    if let std::collections::hash_map::Entry::Vacant(e) = threads.entry(thread_id) { // Vacant = 表里没有这个 id → 可以安全注册        let thread = Arc::new(CodexThread::new(session, io, session_configured.clone(), session_configured.rollout_path.clone(), session_source)); // 组装门面（结构定义见第二节）        e.insert(thread.clone()); // 插入存活线程表；此后 get_thread 就能找到它        return Ok(NewThread { thread_id, thread, session_configured }); // 返回给调用方：id + 门面 + 配置快照    } // Vacant 分支结束} // 注册块结束if let Err(err) = io.shutdown_and_wait().await { warn!("failed to shut down duplicate thread {thread_id}: {err}"); } // 撞车：刚启动的这个是多余的——先把它关掉（尽力而为）Err(CodexErr::InvalidRequest(format!("thread {thread_id} is already running"))) // 再报错；表里保留的是旧运行时，新来的让路

pub(crate) async fn get_thread(&self, thread_id: ThreadId) -> CodexResult<Arc<CodexThread>> { // 客户端侧查询：只放行非内部线程    let threads = self.threads.read().await; // 读锁足够（查询不改表）    match threads.get(&thread_id) { // 按 id 查表        Some(thread) if !thread.session_source.is_internal() => Ok(thread.clone()), // 非内部 → 返回 Arc clone（引用计数 +1，出锁后安全使用）        Some(_) | None => Err(CodexErr::ThreadNotFound(thread_id)), // 内部线程一律报"不存在"：对客户端彻底隐藏    } // match 结束} // get_thread 结束pub async fn remove_thread_for_client(&self, thread_id: &ThreadId) -> CodexResult<Option<Arc<CodexThread>>> { // 客户端发起的移除    let mut threads = self.state.threads.write().await; // 检查与删除共用同一把写锁：被拒绝的 worker 原样留在表里    if threads.get(thread_id).is_some_and(|thread| thread.session_source.is_internal()) { // 目标是内部线程？        return Err(CodexErr::InvalidRequest("live internal threads can only be removed by their owner".to_owned())); // 内部 worker 归其父所有，客户端无权动它    } // 内部线程守卫结束    Ok(threads.remove(thread_id)) // 从表移除：若无其他 Arc 引用，门面 drop → session loop 终止} // remove_thread_for_client 结束

线程创建路径（start / resume / fork 共用）

▼

▼

▼

▼