Boucle principale de l'Agent

源码版本rust-v0.145.0

Pour codex, faire tourner le modèle n'est pas un appel unique mais une boucle : envoyer la requête → recevoir le SSE → parser le tool call → exécuter dans le bac à sable (sandbox) → réinjecter le résultat → renvoyer une requête, jusqu'à ce que le modèle lui-même fasse end_turn. Cette boucle est pilotée par Session::submission_loop, dont l'entrée est CodexThread::submit(Op::UserInput{...}). Les Op entrent dans la boucle via une mpsc channel, les handlers les traitent en asynchrone et produisent un flux d'Event renvoyé au client. Dans les scénarios multi-agent, AgentControl greffe au-dessus de cette boucle une couche de dérivation de sub-agent et de communication inter-agent.

Responsabilités

1. File d'Op unique : CodexThread::submit envoie l'Op sur SessionIo::tx_sub ; dans submission_loop, le handler répartit selon la variante de l'Op vers user_input_or_turn, interrupt, exec_approval, etc. codex-rs/core/src/session/handlers.rs:710-777
2. Cycle de vie : SessionIo::shutdown_and_wait envoie Op::Shutdown puis attend la fin du future session_loop_termination pour un arrêt propre. codex-rs/core/src/session/mod.rs:808-817
3. Registry multi-agent : AgentRegistry maintient, via agent_tree: HashMap<AgentPath, AgentMetadata>, le root et tous les sub-agent spawned ; reserve_spawn_slot limite la concurrence, next_thread_spawn_depth limite la profondeur de récursion. codex-rs/core/src/agent/registry.rs:23-76
4. Communication inter-agent : AgentControl::send_inter_agent_communication emballe un InterAgentCommunication dans un Op::InterAgentCommunication qui passe par la même file submission ; quand un sous-agent termine, maybe_start_completion_watcher notifie automatiquement le parent. codex-rs/core/src/agent/control.rs:167-225

Motivations de conception

Au début, codex était un thread unique avec une seule boucle : la TUI envoie un Op::UserInput, submission_loop tourne un tour, renvoie le flux d'Event à la TUI. Toutes les mutations d'état passent en série dans la même file pour éviter les races. Le scénario multi-agent met ce modèle à l'épreuve : chaque sub-agent est aussi un CodexThread, avec son propre submission_loop et son historique de session, mais partage le même handle AgentControl. AgentControl utilise un Weak<ThreadManagerState> pour référencer en retour le thread manager global, afin d'éviter les références circulaires ; AgentRegistry enveloppe agent_tree d'un Mutex parce que plusieurs sub-agent peuvent le lire/modifier concurremment depuis différentes tokio task.

L'énum Op est le seul type de message dans la file submission. Mettre user input, interrupt, approval, inter-agent communication et settings dans le même enum a cet avantage : l'ordre vu par le handler est l'ordre d'intention de l'utilisateur — l'utilisateur tape d'abord « change de model » puis envoie un message, le handler voit ThreadSettings avant UserInput, sans réordonnancement. maybe_start_completion_watcher est un point de détail intéressant : à la fin d'un sub-agent, il faut notifier le parent ; on tokio::spawn une task indépendante, on récupère un watch channel via subscribe_status, on attend is_final(&status) puis on choisit entre le chemin v2 multi-agent (envoi d'InterAgentCommunication) ou v1 (injection d'un user message), pour ne pas bloquer le submission_loop du parent.

Fichiers clés

codex-rs/core/src/agent/mod.rs:1-11 — entrée du module agent, re-exporte AgentControl, exceeds_thread_spawn_depth_limit.codex-rs/core/src/agent/control.rs:95-108 — structure AgentControl, porte Weak<ThreadManagerState> + Arc<AgentRegistry> + Arc<RolloutBudget>.codex-rs/core/src/agent/control.rs:435-518 — maybe_start_completion_watcher, notifie le parent de façon asynchrone à la fin d'un sub-agent.codex-rs/core/src/codex_thread.rs:162-203 — structure CodexThread et new, porte Arc<Session> + SessionIo + SessionSource.codex-rs/core/src/session/mod.rs:756-806 — SessionIo::submit / submit_with_id, qui pousse réellement la Submission dans le channel.codex-rs/protocol/src/protocol.rs:528-583 — pub enum Op, tous les types de message submission.

CodexThread est en fait une facade : toutes ses méthodes délèguent à SessionIo ; la vraie boucle est dans Session::submission_loop. Cette stratification permet à l'état interne de Session (Arc<Session>) d'être partagé entre plusieurs handles de thread, tandis que SessionIo n'expose que les bouts émetteur/récepteur.

// 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 est la vraie implémentation de l'envoi dans le channel. Submission porte id / op / client_user_message_id / trace ; si trace n'est pas défini, on extrait le W3C trace context du span courant, pour que le traçage distribué s'enchaîne. Si le channel est fermé, on renvoie CodexErr::InternalAgentDied ; en amont, handle_thread_request_result nettoie le registre des threads en conséquence.

// 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 est l'entrée de toute la boucle principale de l'Agent. À l'arrivée d'un Op, match sub.op.clone() répartit vers le handler correspondant ; chaque handler retourne false (ne pas sortir), seul Op::Shutdown retourne true pour faire sortir la boucle. Ainsi, toutes les opérations asynchrones passent en série dans la même file, sans 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; }
    }
}

Flux de données

Limites et échecs

• Propagation de InternalAgentDied : une fois le channel fermé, tous les submit retournent InternalAgentDied ; handle_thread_request_result fait alors remove_thread + release_spawned_thread + nettoie la résidence v2, pour éviter que le registre ne garde des threads morts. codex-rs/core/src/agent/control.rs:238-250
• Limite de profondeur des sub-agent : exceeds_thread_spawn_depth_limit(depth, max_depth) intercepte au stade prepare_thread_spawn, pour empêcher qu'une récursion infinie de spawn ne fasse exploser le processus. codex-rs/core/src/agent/registry.rs:70-76
• Plafond d'agents concurrents : AgentExecutionLimiter initialise max_threads à with_session_id, et ensure_execution_capacity_for_turn_start vérifie avant chaque turn ; au-delà, refus. codex-rs/core/src/agent/control.rs:126-130
• Démarrage prudent d'un turn idle : try_start_turn_if_idle refuse quand un Review est en cours, en Plan mode, ou qu'un turn est déjà en file ; il retourne TryStartTurnIfIdleError en restituant les items tels quels à l'appelant pour qu'il décide. codex-rs/core/src/codex_thread.rs:313-331

Récapitulatif

La boucle principale de l'Agent se résume à « un mpsc channel + un gros match » : tous les Op entrent en série dans submission_loop, les handlers traitent en asynchrone et produisent un flux d'Event. CodexThread est une facade qui enrobe Session et SessionIo en une API d'envoi/réception. En multi-agent, AgentControl greffe au-dessus de la boucle le registry et la communication inter-agent ; la fin d'un sub-agent remonte au parent via maybe_start_completion_watcher. Le détail de l'appel Responses API dans la boucle est dans Client LLM et Responses API ; pour les tool calls et l'exécution sandbox, voir les chapitres sandbox à venir.