Agent Main Loop

源码版本rust-v0.145.0

codex running a model is not a single call but a loop: send a request → receive SSE → parse tool calls → execute in sandbox → feed results back → send another request, until the model itself calls end_turn. This loop is driven by Session::submission_loop; the entry point is CodexThread::submit(Op::UserInput{...}). Ops enter the loop through an mpsc channel, the handler processes them asynchronously and produces a stream of Events back to the client. In multi-agent scenarios, AgentControl layers sub-agent spawning and inter-agent communication on top of this loop.

Responsibilities

1. Single Op queue: CodexThread::submit sends an Op to SessionIo::tx_sub; the handler inside submission_loop dispatches by Op variant to user_input_or_turn, interrupt, exec_approval, and other handlers. codex-rs/core/src/session/handlers.rs:710-777
2. Lifecycle: SessionIo::shutdown_and_wait sends Op::Shutdown and then waits for the session_loop_termination future to complete, ensuring a clean exit. codex-rs/core/src/session/mod.rs:808-817
3. Multi-agent registry: AgentRegistry uses agent_tree: HashMap<AgentPath, AgentMetadata> to track the root + all spawned sub-agents; reserve_spawn_slot bounds concurrency, next_thread_spawn_depth bounds recursion depth. codex-rs/core/src/agent/registry.rs:23-76
4. Inter-agent communication: AgentControl::send_inter_agent_communication wraps InterAgentCommunication as Op::InterAgentCommunication and goes through the same submission queue; when a child agent finishes, maybe_start_completion_watcher automatically dispatches a completion notice to the parent. codex-rs/core/src/agent/control.rs:167-225

Design motivation

Early codex was just a single thread, single loop: the TUI sent an Op::UserInput, the submission_loop ran one turn, and the Event stream flowed back to the TUI. All state changes go through the same queue serially to avoid concurrency races. Multi-agent scenarios challenge this model: each sub-agent is also a CodexThread, with its own submission_loop and conversation history, but they share the same AgentControl handle. AgentControl uses Weak<ThreadManagerState> to refer back to the global thread manager, avoiding reference cycles; AgentRegistry wraps agent_tree in a Mutex because multiple sub-agents may concurrently read / mutate from different tokio tasks.

The Op enum is the only message type in the submission queue. Putting user input, interrupt, approval, inter-agent communication, and settings all into the same enum means the handler order is exactly the user's intent order — the user types "switch model" then sends a message, and the handler sees ThreadSettings first then UserInput, never reordered. maybe_start_completion_watcher is a highlight: a sub-agent has to notify its parent on completion, and it tokio::spawns a separate task that subscribe_status reads a watch channel; only when is_final(&status) does it decide to go the v2 multi-agent path (sending InterAgentCommunication) or the v1 path (injecting a user message), so the parent's submission_loop is not blocked.

Key files

codex-rs/core/src/agent/mod.rs:1-11 — the agent module entry, re-exporting AgentControl, exceeds_thread_spawn_depth_limit.codex-rs/core/src/agent/control.rs:95-108 — the AgentControl struct, holding Weak<ThreadManagerState> + Arc<AgentRegistry> + Arc<RolloutBudget>.codex-rs/core/src/agent/control.rs:435-518 — maybe_start_completion_watcher, notifying the parent asynchronously when a child agent finishes.codex-rs/core/src/codex_thread.rs:162-203 — the CodexThread struct and new, holding Arc<Session> + SessionIo + SessionSource.codex-rs/core/src/session/mod.rs:756-806 — SessionIo::submit / submit_with_id, the actual channel push for a Submission.codex-rs/protocol/src/protocol.rs:528-583 — pub enum Op, all submission message types.

CodexThread is actually a facade: all methods delegate to SessionIo; the real loop is in Session::submission_loop. This layering lets the internal state of Session (Arc<Session>) be shared by multiple thread handles, while SessionIo only exposes the send / receive ends.

// 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 is the real channel push. Submission carries id / op / client_user_message_id / trace; if trace is not set, it is extracted from the current span's W3C trace context, so distributed tracing can be correlated. When the channel is closed it returns CodexErr::InternalAgentDied; the upstream handle_thread_request_result uses this to clean up the thread registry.

// 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 is the entry of the whole agent main loop. When an Op arrives, match sub.op.clone() dispatches to the corresponding handler; each handler returns false (do not exit) when done, and only Op::Shutdown returns true to break the loop. This way all async operations go through the same queue serially, avoiding races.

// 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; }
    }
}

Data flow

Boundaries and failures

• InternalAgentDied propagation: after the channel closes, all submit returns InternalAgentDied; handle_thread_request_result uses this to remove_thread + release_spawned_thread + clear v2 residency, avoiding dead threads lingering in the registry. codex-rs/core/src/agent/control.rs:238-250
• Sub-agent depth limit: exceeds_thread_spawn_depth_limit(depth, max_depth) intercepts at the prepare_thread_spawn stage, preventing infinite recursive spawns from blowing up the process. codex-rs/core/src/agent/registry.rs:70-76
• Concurrent agent cap: AgentExecutionLimiter initializes max_threads at with_session_id; ensure_execution_capacity_for_turn_start checks at the start of each turn and rejects when exceeded. codex-rs/core/src/agent/control.rs:126-130
• Cautious idle-turn start: try_start_turn_if_idle refuses when a Review is in progress, in Plan mode, or with a queued turn; it returns TryStartTurnIfIdleError and hands the items back untouched so the caller can decide. codex-rs/core/src/codex_thread.rs:313-331

Summary

The agent main loop is essentially "single mpsc channel + a big match" — all Ops enter submission_loop serially, handlers process asynchronously and produce a stream of Events. CodexThread is a facade wrapping the send/receive ends of Session and SessionIo into the external API. In multi-agent scenarios AgentControl layers a registry and inter-agent communication on top of the loop; sub-agent completion notifies the parent asynchronously via maybe_start_completion_watcher. For details on calling the Responses API inside the loop, see LLM client and Responses API; tool calls and sandbox execution are covered in the sandbox chapters.