Agent 主循环
codex 跑模型不是一次调用,而是一个循环:发请求 → 收 SSE → 解析 tool call → 沙箱执行 → 把结果回灌 → 再发请求,直到模型自己 end_turn。这个循环由 Session::submission_loop 驱动,入口是 CodexThread::submit(Op::UserInput{...}),Op 通过 mpsc channel 进 loop,handler 异步处理并产生 Event 流回客户端。多 agent 场景下,AgentControl 在这条循环之上挂了 sub-agent 派生与 inter-agent 通信层。
职责
- 单一 Op 队列:
CodexThread::submit把Op发到SessionIo::tx_sub,handler 在submission_loop里按Op变体分发到user_input_or_turn、interrupt、exec_approval等处理函数。codex-rs/core/src/session/handlers.rs:710-777 - 生命周期:
SessionIo::shutdown_and_wait发Op::Shutdown后等session_loop_terminationfuture 完成,保证干净退出。codex-rs/core/src/session/mod.rs:808-817 - 多 agent registry:
AgentRegistry用agent_tree: HashMap<AgentPath, AgentMetadata>维护 root + 所有 spawned sub-agent,reserve_spawn_slot限并发,next_thread_spawn_depth限递归深度。codex-rs/core/src/agent/registry.rs:23-76 - Inter-agent 通信:
AgentControl::send_inter_agent_communication把InterAgentCommunication包成Op::InterAgentCommunication走同一个 submission 队列,子 agent 完成时maybe_start_completion_watcher自动派发完成通知给 parent。codex-rs/core/src/agent/control.rs:167-225
设计动机
早期 codex 就是单线程单循环:TUI 发一个 Op::UserInput,submission_loop 跑一轮 turn,把 Event 流回 TUI。所有状态变更串行进同一队列,避免并发竞争。多 agent 场景挑战了这套模型:每个 sub-agent 也是 CodexThread,有独立 submission_loop 和会话历史,但共享同一个 AgentControl 句柄。AgentControl 用 Weak<ThreadManagerState> 反向引用全局 thread manager,避免循环引用;AgentRegistry 用 Mutex 包住 agent_tree,因为多个 sub-agent 可能在不同 tokio task 里并发查/改。
Op 枚举是 submission 队列里唯一的消息类型。把 user input、interrupt、approval、inter-agent communication、settings 全塞进同一个 enum 的好处是 handler 顺序就是用户意图顺序——用户先敲"换 model"再发消息,handler 看到的就是先 ThreadSettings 后 UserInput,不会被打乱。maybe_start_completion_watcher 是个细节亮点:sub-agent 完成后要通知 parent,它 tokio::spawn 独立 task,subscribe_status 拿 watch channel,等到 is_final(&status) 再决定走 v2 multi-agent path(发 InterAgentCommunication)还是 v1 path(注入 user message),这样 parent 的 submission_loop 不会被阻塞。
关键文件
codex-rs/core/src/agent/mod.rs:1-11 — agent 模块入口,re-export AgentControl、exceeds_thread_spawn_depth_limit。codex-rs/core/src/agent/control.rs:95-108 — AgentControl 结构体,持 Weak<ThreadManagerState> + Arc<AgentRegistry> + Arc<RolloutBudget>。codex-rs/core/src/agent/control.rs:435-518 — maybe_start_completion_watcher,子 agent 完成后异步通知 parent。codex-rs/core/src/codex_thread.rs:162-203 — CodexThread 结构体与 new,持 Arc<Session> + SessionIo + SessionSource。codex-rs/core/src/session/mod.rs:756-806 — SessionIo::submit / submit_with_id,实际把 Submission 推 channel。codex-rs/protocol/src/protocol.rs:528-583 — pub enum Op,所有 submission 消息类型。CodexThread 实际是个 facade:所有方法都委派给 SessionIo,真正的循环在 Session::submission_loop。这样分层让 Session 的内部状态(Arc<Session>)可以被多个 thread 句柄共享,而 SessionIo 只暴露收发两端。
// core/src/codex_thread.rs:162-203 — CodexThread 是 Session + SessionIo 的 facade
pub struct CodexThread {
pub(crate) session: Arc<Session>,
pub(crate) io: SessionIo,
pub(crate) session_source: SessionSource,
session_configured: SessionConfiguredEvent,
rollout_path: Option<PathBuf>,
out_of_band_elicitations: Mutex<OutOfBandElicitations>,
}
impl CodexThread {
pub(crate) fn new(session: Arc<Session>, io: SessionIo, /* ... */) -> Self { /* ... */ }
pub async fn submit(&self, op: Op) -> CodexResult<String> {
self.io.submit(op).await
}
}SessionIo::submit_with_id 是 channel 投递的真正实现。Submission 持有 id / op / client_user_message_id / trace,trace 没设就从当前 span 提取 W3C trace context,让分布式追踪能串起来。channel 关闭时返回 CodexErr::InternalAgentDied,上层 handle_thread_request_result 据此清理 thread 注册表。
// core/src/session/mod.rs:797-817 — submit_with_id + shutdown_and_wait
pub(crate) async fn submit_with_id(&self, mut sub: Submission) -> CodexResult<()> {
if sub.trace.is_none() {
sub.trace = current_span_w3c_trace_context();
}
self.tx_sub
.send(sub)
.await
.map_err(|_| CodexErr::InternalAgentDied)?;
Ok(())
}
pub(crate) async fn shutdown_and_wait(&self) -> CodexResult<()> {
let session_loop_termination = self.session_loop_termination.clone();
match self.submit(Op::Shutdown).await {
Ok(_) => {}
Err(CodexErr::InternalAgentDied) => {}
Err(err) => return Err(err),
}
session_loop_termination.await;
Ok(())
}submission_loop 是整个 agent 主循环的入口。Op 进来后 match sub.op.clone() 分发到对应 handler,每个 handler 处理完返回 false(不退出),只有 Op::Shutdown 返回 true 让循环退出。这样所有异步操作都串行进同一队列,避免 race。
// core/src/session/handlers.rs:710-777 — submission_loop: Op 串行处理
pub(super) async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiver<Submission>) {
let mut shutdown_received = false;
while let Ok(sub) = rx_sub.recv().await {
debug!(?sub, "Submission");
let dispatch_span = submission_dispatch_span(&sub);
let should_exit = async {
match sub.op.clone() {
Op::Interrupt => { interrupt(&sess).await; false }
Op::UserInput { .. } => {
user_input_or_turn(&sess, sub.id.clone(), sub.op, sub.client_user_message_id).await;
false
}
Op::ThreadSettings { thread_settings } => {
update_thread_settings(&sess, sub.id.clone(), thread_settings).await;
false
}
Op::InterAgentCommunication { communication } => {
inter_agent_communication(&sess, sub.id.clone(), communication).await;
false
}
// ... 其他 Op 变体
}
}.await;
if should_exit { break; }
}
}数据流
边界与失败
InternalAgentDied传播:channel 关闭后所有submit返回InternalAgentDied,handle_thread_request_result据此remove_thread+release_spawned_thread+ 清 v2 residency,避免注册表残留死线程。codex-rs/core/src/agent/control.rs:238-250- sub-agent 深度限制:
exceeds_thread_spawn_depth_limit(depth, max_depth)在prepare_thread_spawn阶段拦截,防止无限递归 spawn 把进程撑爆。codex-rs/core/src/agent/registry.rs:70-76 - 并发 agent 上限:
AgentExecutionLimiter在with_session_id时初始化max_threads,ensure_execution_capacity_for_turn_start在每个 turn 开始前检查,超限拒绝。codex-rs/core/src/agent/control.rs:126-130 - idle turn 谨慎启动:
try_start_turn_if_idle拒绝 Review 进行中、Plan mode、已有排队 turn 的场景,返回TryStartTurnIfIdleError把 items 原样还回让调用方决定。codex-rs/core/src/codex_thread.rs:313-331
小结
agent 主循环的本质是"单一 mpsc channel + 大 match"——所有 Op 串行进 submission_loop,handler 异步处理产生 Event 流。CodexThread 是个 facade,把 Session 和 SessionIo 收发两端包装成对外 API。多 agent 场景下 AgentControl 在循环之上挂了 registry 和 inter-agent 通信,sub-agent 完成通过 maybe_start_completion_watcher 异步通知 parent。循环里调 Responses API 的细节见 LLM 客户端与 Responses API,工具调用与沙箱执行见后续 sandbox 章节。