account_processor: Arc<AccountRequestProcessor>,          // 账户类 RPC：登录、限流、GCP setupapps_processor: AppsRequestProcessor,                     // App 市场 / 应用管理catalog_processor: CatalogRequestProcessor,               // 模型目录查询command_exec_processor: CommandExecRequestProcessor,      // 一次性命令执行（exec）process_exec_processor: ProcessExecRequestProcessor,      // 长驻进程句柄操作config_processor: ConfigRequestProcessor,                 // 全局配置读写environment_processor: EnvironmentRequestProcessor,       // 环境变量 / 环境管理external_agent_config_processor: ExternalAgentConfigRequestProcessor,  // 外部 agent 接入配置feedback_processor: FeedbackRequestProcessor,             // 用户反馈提交fs_processor: FsRequestProcessor,                         // 文件系统 RPC（读/写/watch）git_processor: GitRequestProcessor,                       // Git 相关查询initialize_processor: InitializeRequestProcessor,         // 连接建立后的首次握手marketplace_processor: MarketplaceRequestProcessor,       // 插件市场操作mcp_processor: McpRequestProcessor,                       // MCP server 管理 / OAuthplugin_processor: PluginRequestProcessor,                 // 插件安装/启用/禁用project_processor: ProjectRequestProcessor,               // 项目（工作区）管理remote_control_processor: RemoteControlRequestProcessor,  // 远程控制 / 配对search_processor: SearchRequestProcessor,                 // 模糊搜索会话thread_goal_processor: ThreadGoalRequestProcessor,        // 线程目标设定thread_queue_processor: ThreadQueueRequestProcessor,      // 线程队列操作thread_processor: ThreadRequestProcessor,                 // 线程生命周期（start/resume/archive/delete）turn_processor: TurnRequestProcessor,                     // turn 提交/steer/interrupt——本讲核心windows_sandbox_processor: WindowsSandboxRequestProcessor,// Windows 沙箱配置

if !session.initialized() {    return Err(invalid_request("Not initialized"));   // 连接还没握手 → 直接拒绝，不进业务逻辑}if let Some(reason) = codex_request.experimental_reason()    && !session.experimental_api_enabled()            // 方法被标记为实验性、且客户端没开实验开关{    return Err(invalid_request(experimental_required_message(reason)));  // 返回明确错误，告诉客户端怎么开启}

#[derive(Debug, Default)]struct AdmissionState {    closed: bool,      // 关停 drain 已开始 → 从此拒绝所有新 turn    paused: bool,      // 工作区切换进行中 → 暂停准入    active: usize,     // 当前持有的 permit 数（在途的 turn 提交）}#[derive(Clone, Debug)]pub(crate) struct TurnAdmission {    state: Arc<Mutex<AdmissionState>>,   // 共享状态，锁只持有很短临界区    active_tx: watch::Sender<usize>,     // 广播 active 计数，pause() 靠它等在途排空}

pub(crate) fn admit(&self) -> Result<TurnPermit, JSONRPCErrorError> {    let mut state = self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);    if state.closed {        return Err(server_draining_error());   // 服务器正在 drain → 让客户端去别处重试    }    if state.paused {        return Err(invalid_request("Workspace change in progress; retry when it completes"));  // 工作区切换没完成 → 拒绝并提示稍后重试    }    state.active += 1;                          // 把本次请求计入"在途"    self.active_tx.send_replace(state.active);  // 通知 pause() 的等待者：计数变了    Ok(TurnPermit(self.clone()))                // RAII permit：drop 时自动减一，无需手动释放}impl Drop for TurnPermit {    fn drop(&mut self) {        let mut state = self.0.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);        state.active -= 1;                      // 请求结束（或失败），释放计数        self.0.active_tx.send_replace(state.active);  // 若 pause() 正在等排空，这里把它唤醒    }}

pub(crate) async fn pause(&self) -> Result<TurnAdmissionPause, JSONRPCErrorError> {    let mut active = self.subscribe_active();   // 先订阅 active 计数，再置 paused（避免漏掉信号）    {        let mut state = self.state.lock().unwrap_or_else(std::sync::PoisonError::into_inner);        if state.closed || state.paused {            return Err(invalid_request("Runtime changes are already paused"));  // 重复 pause → 直接报错，不叠加        }        state.paused = true;                     // 关门：此后 admit() 一律被拒    }    let pause = TurnAdmissionPause(self.clone()); // RAII：drop 时自动 unpause（即使下面超时）    tokio::time::timeout(        std::time::Duration::from_secs(/*secs*/ 10),        active.wait_for(|active| *active == 0), // 等在途 turn 提交全部结束    )    .await    .map_err(|_| invalid_request("Work is still starting; retry the workspace switch"))?    .map_err(|_| invalid_request("Turn admission is unavailable"))?;    Ok(pause)}

#[derive(Clone, Debug, Eq, Hash, PartialEq)]pub(crate) enum RequestSerializationQueueKey {    Global(&'static str),                    // 全局资源：config / account-auth / memory ...    Thread { thread_id: String },            // 每线程一条独占队列（turn/start 等）    ThreadPath { path: PathBuf },            // resume/fork 按 rollout 文件路径排队    CommandExecProcess { connection_id, process_id },  // command exec 会话    Process { connection_id, process_handle },          // 长驻进程句柄    FuzzyFileSearchSession { session_id: String },      // 模糊搜索会话    FsWatch { connection_id, watch_id: String },        // fs watch    McpOauth { server_name: String },                 // 每个 MCP server 一条 OAuth 流}

async fn drain(self, key: RequestSerializationQueueKey) {    loop {        let (requests, changed) = {            let mut queues = self.inner.lock().await;            let Some(queue) = queues.get_mut(&key) else { return };   // 队列已被移除 → 直接退出            match queue.requests.pop_front() {                Some(request) => {                    let access = request.access;                    let mut requests = vec![request];                    if access == RequestSerializationAccess::SharedRead {                        // 队首是读请求 → 把后面连续的读全部拉出来，准备并发执行                        while queue.requests.front().is_some_and(|request| {                            request.access == RequestSerializationAccess::SharedRead                        }) {                            let Some(request) = queue.requests.pop_front() else { break; };                            requests.push(request);                        }                    }                    (requests, Arc::clone(&queue.changed))   // 顺手拿走变更信号，供后续 wait_for                }                None => { queues.remove(&key); return; }      // 队列空了 → 把 key 本身删掉，省内存            }        };        if requests[0].access == RequestSerializationAccess::Exclusive {            for request in requests {                request.request.run().await;   // 独占：严格串行，一个一个跑            }            continue;        }        let mut running_reads = requests            .into_iter()            .map(|request| request.request.run())            .collect::<<FuturesUnordered<_>>();  // 共享读：并发执行，谁先完成谁先出结果

let thread_start_task = async move {    if let Err(error) = Self::thread_start_task(        listener_task_context,        thread_store,        config_manager,        request_id,        /* …其余 20+ 参数省略… */    )    .await    {        outgoing.send_error(error_request_id, error).await;  // 失败 → 把 JSON-RPC error 发回客户端    }};self.background_tasks    .spawn(thread_start_task.instrument(request_context.span()))   // spawn 后台任务并挂上 tracing span    .await    .map_err(|_| internal_error("thread startup task stopped before completing"))?;Ok(())   // RPC 立即返回：thread/start 是"已受理"，不是"已完成"

let reserved_thread_id = if start_options.config.ephemeral {    None   // 临时线程不落盘 → 不需要提前预留 id} else {    stage_pending_thread_metadata(          // 先在存储里写一条"pending"元数据行        listener_task_context.thread_manager.as_ref(),        thread_store.as_ref(),        StoreThreadMetadataPatch {            project_id: project_id.clone().map(Some),   // 绑定工作区（如有）            daybreak_enabled,                            // 记录 Daybreak 开关状态            ..Default::default()        },        "thread/start",                                // 标记写入来源，便于审计追溯    )    .await?};let thread_id = reserved_thread_id    .unwrap_or_else(|| listener_task_context.thread_manager.reserve_thread_id());  // 临时线程 → 只在内存里预留start_options.reserved_thread_id = Some(thread_id);   // id 钉死：后续创建失败也能按这个 id 精确清理

let submission = thread    .start_or_steer_turn(        TurnInputRequest::new(input)            .with_thread_settings(thread_settings)      // 每 turn 的配置覆盖（模型/effort/沙箱…）            .on_start(TurnStartOptions {                turn_trigger: params.turn_trigger,       // 触发来源（用户 / 自动化）                parent_turn_id: params.parent_turn_id,   // 子 agent 场景的父 turn id                initiating_agent_path: None,                root_turn_id: params.root_turn_id,                final_output_json_schema: params.output_schema,  // 结构化输出 schema                service_tier: params.service_tier_for_turn,      // 服务档位（如有）                cyber_access_program: params.cyber_access_program.map(Into::into),            })            .with_additional_context(additional_context)       // 客户端注入的附加上下文            .with_responses_metadata(params.responsesapi_client_metadata)            .with_responses_metadata_patch(params.responsesapi_client_metadata_patch)            .with_trace(self.request_trace_context(&request_id).await),  // W3C trace context，跨进程追踪    )    .await    .map_err(|err| { /* 错误映射省略 */ })?;let (turn_id, root_turn_id, started) = match submission {    TurnInputSubmission::Accepted => {        return Err(internal_error(   // Accepted 表示"agent 收下但没启动"——对用户输入不该出现            "user input unexpectedly returned agent acceptance".to_string(),        ));    }    TurnInputSubmission::Started { turn_id, root_turn_id } => (turn_id, root_turn_id, true),   // 新 turn 已启动    TurnInputSubmission::Steered { turn_id, root_turn_id } => (turn_id, root_turn_id, false),  // 合并进了在途 turn    TurnInputSubmission::NotSubmitted { reason } => { /* 错误映射省略 */ }             // 被拒：draining / 无活动 turn 等

if !is_startup_interrupt {    let thread_state = self.thread_state_manager.thread_state(thread_uuid).await;    let is_running = matches!(thread.agent_status().await, AgentStatus::Running);    {        let mut thread_state = thread_state.lock().await;   // 锁的是每线程状态（不是全局锁）        if let Some(active_turn_id) = thread_state.active_turn_id() {            if active_turn_id != turn_id {                return Err(invalid_request(format!(                    "expected active turn id {turn_id} but found {active_turn_id}"   // 乐观校验：turn 不匹配直接拒绝                )));            }        } else if thread_state.last_terminal_turn_id.as_deref() == Some(turn_id.as_str())            || !is_running                                  // turn 已结束、或 agent 没在跑 → 没有可中断的东西        {            return Err(invalid_request("no active turn to interrupt"));        }        thread_state.pending_interrupts.push(request_id.clone());   // 先记账不回复——等 TurnAborted 事件再回响应    }    self.outgoing.record_request_turn_id(request_id, &turn_id).await;  // 绑定 request→turn，事件到达时才能找到它

请求路径（turn/start）

▼

▼

▼

▼

▼

事件回程（turn 运行中）

▼

▼

▼