async fn spawn_agent_internal( // 所有派生入口的汇聚点：一次子代理出生要过的全部闸门都在这里    &self,    config: Config, // 父会话配置：容量上限、角色、feature flag 都从这里取    initial_input: SpawnInitialInput, // 初始输入二选一：用户 prompt，或 agent 间通信消息（V2）    session_source: Option<SessionSource>, // 会话来源：ThreadSpawn 才走子代理分支，None 表示普通新线程    options: SpawnAgentOptions, // 可选参数：fork 模式、父 turn id、环境选择等) -> CodexResult<LiveAgent> {    let state = self.upgrade()?; // 弱引用升级成强引用，thread manager 已销毁就直接报错退出    let multi_agent_version = state // 先定版本：V1/V2/Disabled 决定后面走哪套容量与通知机制        .effective_multi_agent_version_for_spawn(            &InitialHistory::New,            session_source.as_ref(),            options.parent_thread_id,            /*forked_from_thread_id*/ None, // fork 场景由 spawn_forked_thread 单独处理，这里传空            &config,        )        .await;    if let Some(session_source) = session_source.as_ref() {        self.ensure_execution_capacity(multi_agent_version, session_source)?; // 执行容量硬检查：V2 子代理超限直接 AgentLimitReached，fail loud    }    let agent_max_threads = config.effective_agent_max_threads(multi_agent_version); // V2 上限=配置值-1（给父会话留一个位置），V1 用默认常量    let spawn_uses_v2_residency = multi_agent_version == MultiAgentVersion::V2 // 只有"V2 + SubAgent 来源"才占用 LRU 常驻槽位        && session_source.as_ref().is_some_and(is_v2_resident_session_source);    let residency_slot = if spawn_uses_v2_residency {        Some(            self.reserve_v2_residency_slot(&state, &config, /*protected_thread_id*/ None) // 预留常驻槽位：满了会先尝试 LRU 驱逐一个闲子代理（第三节）                .await?,        )    } else {        None    };    let reservation_max_threads = if spawn_uses_v2_residency {        None // V2 走常驻槽位后不再重复计数，避免双重限流    } else {        agent_max_threads    };    let mut reservation = self.state.reserve_spawn_slot(reservation_max_threads)?; // RAII 预留：函数中途 return/panic 时 Drop 自动归还（第二节）    let inheritance = SpawnAgentThreadInheritance { // 继承清单：子代理默认复用父会话的运行环境与安全策略        environments: self.inherited_environments_for_source(&state, session_source.as_ref()).await, // 快照父线程的 turn 环境（工作目录、工具配置）        exec_policy: self.inherited_exec_policy_for_source(&state, session_source.as_ref(), &config).await, // 复用父级 ExecPolicyManager，避免子代理重新解析策略文件    };    let (session_source, mut agent_metadata) = match session_source {        Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { parent_thread_id, depth, agent_path, agent_role, .. })) => { // 只有 ThreadSpawn 来源才需要登记父子关系与元数据            let (session_source, agent_metadata) = self.prepare_thread_spawn( // 预留 agent path + 昵称，构造带 depth/role 的 SessionSource                &mut reservation, &config, parent_thread_id, depth, agent_path, agent_role, /*preferred_agent_nickname*/ None,            )?;            (Some(session_source), agent_metadata)        }        other => (other, AgentMetadata::default()), // 非子代理来源：不登记元数据，走普通线程路径    };    let notification_source = session_source.clone(); // 留一份副本给后面的完成通知与埋点用    // The same `AgentControl` is sent to spawn the thread. // 注释原文：同一个 AgentControl 会被注入新线程——子代理共享同一套控制面    let new_thread = match (session_source, options.fork_mode.as_ref(), inheritance) {        (Some(session_source), Some(_), inheritance) => {            Box::pin(self.spawn_forked_thread(&state, config, session_source, &options, inheritance, multi_agent_version)).await? // fork 模式：复制父级历史再过滤（第四节）        }        (Some(session_source), None, inheritance) => {            let history_mode = if let Some(parent_thread_id) = options.parent_thread_id // 父线程是分页历史模式时，子线程也沿用——保证 resume 路径一致                && let Ok(parent_thread) = state.get_thread(parent_thread_id).await            {                matches!(parent_thread.config_snapshot().await.history_mode, ThreadHistoryMode::Paginated).then_some(ThreadHistoryMode::Paginated)            } else {                None            };            Box::pin(state.spawn_new_thread_with_source( // 常规路径：新建线程并注入继承的环境与策略                config.clone(), self.clone(), session_source, history_mode, options.parent_thread_id,                /*forked_from_thread_id*/ None, /*thread_source*/ Some(ThreadSource::Subagent), /*metrics_service_name*/ None,                inheritance.environments, inheritance.exec_policy, options.environments.clone(),            )).await?        }        (None, _, _) => Box::pin(state.spawn_new_thread(config.clone(), self.clone())).await?, // 无来源：最普通的裸线程创建    };    agent_metadata.agent_id = Some(new_thread.thread_id); // 元数据回填真实 thread id，之后按 id 可反查角色/昵称    reservation.commit(agent_metadata.clone()); // 提交预留：active=false，Drop 不再回滚——资源正式归属新线程    if let Some(residency_slot) = residency_slot {        residency_slot.commit(new_thread.thread_id); // 常驻槽位转正：pending-1、进入 LRU 队列尾部（最近使用）    }

pub(crate) fn effective_agent_max_threads( // 按多智能体版本返回子代理并发上限    &self,    multi_agent_version: MultiAgentVersion,) -> Option<usize> {    match multi_agent_version {        MultiAgentVersion::V2 => Some(            self.multi_agent_v2.max_concurrent_threads_per_session.saturating_sub(1), // V2：配置值减 1——父会话自己占一个执行位，子代理池必须让出位置        ),        MultiAgentVersion::Disabled | MultiAgentVersion::V1 => {            self.agent_max_threads.or(DEFAULT_AGENT_MAX_THREADS) // V1/关闭：用户显式配置优先，否则用默认常量兜底        }    }}

pub(crate) fn reserve_spawn_slot( // 预留一个派生槽位：先查上限，再返回 RAII 守卫    self: &Arc<Self>,    max_threads: Option<usize>, // None 表示不限（V2 常驻路径走这里）) -> Result<SpawnReservation> {    if let Some(max_threads) = max_threads {        if !self.try_increment_spawned(max_threads) { // CAS 自增：当前计数已达上限则失败，不排队等待——快速失败            return Err(CodexErr::new(CodexErrorDetails::AgentLimitReached { max_threads })); // 超限直接报错，调用方（模型）能看到明确原因并调整策略        }    } else {        self.total_count.fetch_add(1, Ordering::AcqRel); // 无上限路径：只计数不检查，用于 V2 常驻槽位之外的场景    }    Ok(SpawnReservation { state: Arc::clone(self), active: true, reserved_agent_nickname: None, reserved_agent_path: None }) // active=true 标记"未提交"，Drop 时据此决定是否回滚}impl Drop for SpawnReservation { // RAII：预留对象离开作用域且未 commit → 自动归还资源    fn drop(&mut self) {        if self.active { // 已 commit 的守卫 active=false，跳过——避免二次释放            if let Some(agent_path) = self.reserved_agent_path.take() {                self.state.release_reserved_agent_path(&agent_path); // 归还预留的 agent path（如 /root/worker-1），防止路径泄漏被永久占用            }            self.state.total_count.fetch_sub(1, Ordering::AcqRel); // 计数回退，槽位重新可用        }    }}

async fn reserve_slot( // V2 常驻槽位预留：占不到就驱逐最闲的常驻线程，循环重试    self: Arc<Self>,    manager: &Arc<ThreadManagerState>,    capacity: usize, // 容量上限（配置值减 1）    protected_thread_id: Option<ThreadId>, // 受保护线程：本轮驱逐扫描中跳过它（resume 自己的场景）) -> CodexResult<V2ResidencySlot> {    loop {        if self.try_reserve_pending_slot(capacity) { // 先尝试占一个 pending 槽位：residents+pending < capacity 才成功            return Ok(V2ResidencySlot { residency: self, active: true }); // 拿到守卫，等 commit 转正；失败路径 Drop 自动释放        }        if !self.try_unload_one_resident(manager, protected_thread_id).await { // 满了：尝试驱逐一个可卸载的常驻线程            return Err(CodexErr::new(CodexErrorDetails::AgentLimitReached { max_threads: capacity })); // 没有可驱逐对象（都在干活）才真正报错        }    }}async fn try_unload_one_resident( // LRU 扫描：从最久未使用的常驻线程开始找"可卸载者"    &self, manager: &Arc<ThreadManagerState>, protected_thread_id: Option<ThreadId>,) -> bool {    let candidates_to_scan = self.resident_count(); // 最多扫一轮，防止死循环    for _ in 0..candidates_to_scan {        let Some(candidate_thread_id) = self.pop_lru_candidate(protected_thread_id) else { return false }; // 从队头（最久未用）弹出一个候选；受保护线程会被放回队尾跳过        let Some(candidate_thread) = manager.get_thread(candidate_thread_id).await.ok() // 线程可能已消失，取不到就继续下一个            .filter(|thread| is_resident_candidate(thread)) else { continue }; // 只处理 V2 + SubAgent 来源的常驻线程        if !is_unloadable(candidate_thread.as_ref()).await { // 终态 + 无活跃 turn + 邮箱空，三者缺一不可（第 233-239 行）            self.touch(candidate_thread_id); // 还在干活：刷新 LRU 位置到队尾，本轮不碰它            continue;        }        candidate_thread.ensure_rollout_materialized().await; // 先把未落盘的 rollout 刷出来——驱逐前保证历史完整持久化        if let Err(err) = candidate_thread.shutdown_and_wait().await { // 优雅关闭并等待退出；失败则刷新 LRU、换下一个候选            warn!("failed to shut down v2 resident thread before unloading {candidate_thread_id}: {err}");            self.touch(candidate_thread_id);            continue;        }        let environments = candidate_thread.environment_selections().await; // 记下它的环境选择，驱逐后 resume 时能原样恢复        candidate_thread.session.services.agent_control.state.save_evicted_environments(candidate_thread_id, environments); // 环境快照存入 registry（evicted_environments）        let _ = manager.remove_thread(&candidate_thread_id).await; // 从线程管理器移除，槽位释放        return true;    }    false // 扫完一轮没有可卸载者：调用方据此返回 AgentLimitReached}

if let Some(features) = role_config.features { // 角色 toml 里声明的 feature 开关    for (key, enabled) in features.entries() {        if !enabled // 只处理"关闭"方向的声明——开启方向直接忽略，这是降权约束的第一道闸            && let Some(                feature @ (Feature::ShellTool | Feature::Apps | Feature::Personality // 白名单：只有这几个敏感能力允许被角色关掉                    | Feature::Plugins | Feature::MemoryTool | Feature::RequestPermissionsTool),            ) = feature_for_key(&key)        {            overrides.features.insert(feature.key().to_string(), false); // 记入覆盖表，后面统一 disable——白名单外的 key 不生效        }    }}if let Some(mut skills) = role_config.skills { // 技能同理：只保留"禁用"声明，启用声明被过滤掉    skills.config.retain(|skill| !skill.enabled); // 自定义技能列表里只留 enabled=false 的条目    skills.bundled = skills.bundled.filter(|bundled| !bundled.enabled); // 内置技能同理：只保留禁用项    skills.include_instructions = skills.include_instructions.filter(|enabled| !enabled); // include_instructions 为 false 才保留（即"不要注入说明"）    skills.max_context_tokens = None; // 角色不允许调整技能上下文预算——防止子代理用更大窗口绕过限制}let role_layer_toml = TomlValue::try_from(&overrides)?; // 把覆盖表序列化成 toml，作为新的配置层if role_layer_toml.as_table().is_some_and(toml::map::Map::is_empty) {    return Ok(()); // 没有任何有效覆盖：原样返回父配置，不产生多余配置层}*config = role_overrides::build_next_config(config, role_layer_toml, &overrides)?; // 生成新 Config：角色层插入配置栈 + 逐项应用覆盖

fn keep_forked_rollout_item(item: &RolloutItem, preserve_reference_context_item: bool) -> bool { // fork 过滤器：逐条决定父历史能否进入子线程    match item {        RolloutItem::ResponseItem(envelope) => match &envelope.item {            ResponseItem::Message { role, phase, .. } => match role.as_str() {                "system" | "developer" | "user" => true, // 指令与用户消息保留：子代理需要同样的任务上下文                "assistant" => *phase == Some(MessagePhase::FinalAnswer), // assistant 只留最终答案——中间推理对子代理是噪音且浪费 token                _ => false,            },            ResponseItem::FunctionCallOutput { call_id: None, .. } | ResponseItem::ConfigurationUpdate { .. } => true, // 无配对调用的输出与配置变更保留：它们承载环境事实            ResponseItem::AdditionalTools { .. } | ResponseItem::AgentMessage { .. } | ResponseItem::Reasoning { .. } // 工具清单、agent 间消息、推理过程——全部丢弃                | ResponseItem::LocalShellCall { .. } | ResponseItem::FunctionCall { .. } | ResponseItem::ToolSearchCall { .. }                | ResponseItem::FunctionCallOutput { call_id: Some(_), .. } | ResponseItem::CustomToolCall { .. } // 工具调用与配对输出成对丢弃：子代理会自己重新执行需要的操作                | ResponseItem::CustomToolCallOutput { .. } | ResponseItem::ToolSearchOutput { .. }                | ResponseItem::WebSearchCall { .. } | ResponseItem::ImageGenerationCall { .. }                | ResponseItem::Compaction { .. } | ResponseItem::CompactionTrigger { .. } | ResponseItem::ContextCompaction { .. } // 压缩痕迹不继承：子线程有自己的上下文预算与压缩策略                | ResponseItem::Other => false,        },        RolloutItem::RealtimeItem(_) | RolloutItem::InterAgentCommunication(_) // 实时流、agent 间通信——父级协作过程对子代理无意义            | RolloutItem::InterAgentCommunicationMetadata { .. } | RolloutItem::RetainedContext(_) | RolloutItem::SecurityRiskScore(_) => false,        // Full-history forks preserve the cached prompt prefix and can keep diffing from the parent's durable baseline. // 注释原文：全量 fork 保留缓存的 prompt 前缀，可继续基于父级基线做增量        // Truncated forks drop part of that prompt, so they must rebuild context on their first child turn. // 截断 fork 丢了一部分 prompt，必须在首个子轮次重建上下文        RolloutItem::TurnContext(_) | RolloutItem::WorldState(_) => preserve_reference_context_item, // 参考上下文条目：全量 fork 保留、LastNTurns 截断 fork 丢弃——由调用方按 fork 模式决定        // Child threads inherit model context, not the parent's cumulative usage state. // 注释原文：子线程继承模型上下文，不继承父级累计用量状态        RolloutItem::TokenUsageRecord(_) => false, // token 用量是父会话的账本，子代理从零开始计数        RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) => true, // 压缩检查点、事件流、会话元数据保留：resume 与审计依赖它们    }}

let control = self.clone(); // 克隆控制面句柄，watcher 是独立任务tokio::spawn(async move { // 脱离式任务：不阻塞 spawn 调用方，父线程无需等待子代理结束    let status = match control.subscribe_status(child_thread_id).await { // 订阅子线程状态流（broadcast）        Ok(mut status_rx) => {            let mut status = status_rx.borrow().clone();            while !is_final(&status) { // 轮询直到终态：Completed / Errored / Interrupted                if status_rx.changed().await.is_err() { // 订阅断开（线程被驱逐/销毁）→ 降级为主动查询一次                    status = control.get_status(child_thread_id).await;                    break;                }                status = status_rx.borrow().clone();            }            status        }        Err(_) => control.get_status(child_thread_id).await, // 订阅失败兜底：直接查当前状态    };    if !is_final(&status) { return; } // 非终态（如线程已消失）：放弃通知，避免注入脏数据    let Ok(state) = control.upgrade() else { return; }; // manager 已销毁则退出    let child_thread = state.get_thread(child_thread_id).await.ok();    let child_uses_multi_agent_v2 = match child_thread.as_ref() {        Some(child_thread) => child_thread.multi_agent_version() == Some(MultiAgentVersion::V2), // 子线程实际版本以运行时为准（配置可能中途变化）        None => true, // 查不到时按 V2 处理：走通信路径更安全    };    if child_agent_path.is_some() && child_uses_multi_agent_v2 { // V2 + 有 agent path → 构造 InterAgentCommunication 发给父级邮箱（trigger_turn=false，不强制开新轮次）        let Some(parent_agent_path) = child_agent_path.as_str().rsplit_once('/').and_then(|(parent, _)| AgentPath::try_from(parent).ok()) else { return; }; // 从子路径 /root/worker-1 反推父路径 /root        let Some(message) = format_inter_agent_completion_message(parent_agent_path.clone(), child_agent_path.clone(), &status) else { return; }; // 按终态格式化结果消息（成功/失败文案不同）        let communication = InterAgentCommunication::new(child_agent_path, parent_agent_path, Vec::new(), message, /*trigger_turn*/ false);        let context = AgentCommunicationContext::new(AgentCommunicationKind::Result, child_thread_id); // 标记为 Result 类型通信，父级可区分"结果"与"指令"        let _ = control.send_inter_agent_communication(parent_thread_id, communication, context, TurnStartOptions::default()).await;        return;    }    let Ok(parent_thread) = state.get_thread(parent_thread_id).await else { return; }; // V1 路径：父线程必须还活着才能注入    parent_thread.inject_fragment_without_turn(SubagentNotification::new(child_reference.as_str(), status)).await; // 以"无 turn 片段"形式注入子代理完成通知，父级下一轮自然看到});

#[derive(Default)]pub(super) struct AgentExecutionLimiter { // 执行限流器：只统计"正在跑 turn"的子代理，不是总线程数    active: AtomicUsize, // 当前活跃执行计数（原子操作，无锁）    max_threads: OnceLock<usize>, // 上限惰性初始化：第一次 spawn 时按配置写入，之后不可变}pub(crate) struct AgentExecutionGuard { // RAII guard：持有期间 active+1，drop 时自动 -1    limiter: Arc<AgentExecutionLimiter>,}impl Drop for AgentExecutionGuard {    fn drop(&mut self) { self.limiter.active.fetch_sub(1, Ordering::AcqRel); } // turn 结束（无论成败）归还执行位——作用域即租期}fn is_execution_limited(multi_agent_version: MultiAgentVersion, session_source: &SessionSource) -> bool { // 限流只作用于 V2 + SubAgent：根会话与 V1 不受此闸约束    multi_agent_version == MultiAgentVersion::V2 && matches!(session_source, SessionSource::SubAgent(_))}impl AgentExecutionLimiter {    fn has_capacity(&self) -> bool { self.active.load(Ordering::Acquire) < self.max_threads() } // 容量判断：活跃数严格小于上限才放行    fn guard(self: Arc<Self>) -> AgentExecutionGuard { // 获取执行位：先自增再返回守卫（调用方需先 has_capacity 检查）        self.active.fetch_add(1, Ordering::AcqRel);        AgentExecutionGuard { limiter: self }    }}

spawn_agent_internal 流水线（V2 ThreadSpawn）

▼

▼

▼

▼

▼

▼