Architecture App-server

源码版本rust-v0.145.0

app-server est le server JSON-RPC de codex : il emballe ThreadManager, AuthManager, ConfigManager de codex-core en un processus réutilisable par IDE, Web et TUI. Il expose trois transports — stdio / unix socket / websocket — et utilise à l'interne un MessageProcessor pour router les requêtes ; toutes les RPC passent par le même protocole ClientRequest / ServerNotification.

Responsabilités

1. L'entrée run_main_with_transport_options charge config, auth, otel, state_db, démarre le transport acceptor et l'outbound router, puis spawn MessageProcessor qui entre dans la boucle principale select (codex-rs/app-server/src/lib.rs:449-460).
2. La couche Transport unifie stdio:// / unix:// / ws://IP:PORT / off en TransportEvent (ConnectionOpened / ConnectionClosed / IncomingMessage) ; toutes les connexions partagent un event channel (codex-rs/app-server-transport/src/transport/mod.rs:73-78).
3. MessageProcessor::process_request désérialise un JSONRPCRequest en ClientRequest, puis répartit entre Initialize et requête déjà initialisée ; les requêtes déjà initialisées passent par dispatch_initialized_client_request vers le processor correspondant (codex-rs/app-server/src/message_processor.rs:518-570).
4. Chaque requête déjà initialisée est mise en file ou spawnée selon serialization_scope : RequestSerializationQueues garantit la sérialisation des requêtes de même scope (par ex. les opérations de turn d'un même thread), la concurrence n'a lieu qu'entre scopes différents (codex-rs/app-server/src/message_processor.rs:846-861).
5. La sortie passe par OutgoingEnvelope::ToConnection / Broadcast ; le broadcast n'envoie qu'aux connexions déjà initialized ; les notifications expérimentales ne vont qu'aux connexions ayant activé experimental_api (codex-rs/app-server/src/transport.rs:198-237).

Motivations de conception

app-server est un crate séparé parce que les IDE (VSCode, JetBrains), la TUI et les apps de pairage distant partagent le même protocole mais dans des environnements différents : l'IDE tourne en stdio, le pairage distant en websocket, le mode daemon nécessite unix socket + fichier pid. Extraire le transport dans le crate app-server-transport et le protocole dans app-server-protocol, le crate server ne s'occupe que de « comment router JSON-RPC vers ThreadManager » — ajouter un transport ne touche pas à la logique principale.

ConnectionSessionState utilise OnceLock<InitializedConnectionSessionState> plutôt qu'un Mutex, parce que initialize est une opération unique — une fois posée, elle est immutable, et les requêtes suivantes ne font que lire. Cela évite le coût d'un lock pour « connexion déjà initialisée » ; experimental_api_enabled / opted_out_notification_methods se lisent en atomic / RwLock.

Le routing des requêtes utilise une collection de processor struct plutôt qu'un seul match : chaque processor (ThreadRequestProcessor, TurnRequestProcessor, ConfigRequestProcessor, etc.) détient son propre état ; handle_initialized_client_request ne fait que le match de dispatch. Ainsi, ajouter une nouvelle méthode ne touche pas au code central, et le test interne de chaque processor est plus simple.

serialization_scope est un design intéressant : les opérations de turn d'un même thread doivent être sérialisées (sinon deux Op::UserInput s'entrelacent), mais des requêtes de threads différents peuvent être concurrentes. RequestSerializationQueues key par (connection_id, scope) ; les requêtes de même key se mettent en file, les cross-key partent en tokio::spawn.

Le routing outbound en try_send + kick-when-full protège le server contre un client lent : si la file outbound websocket est pleine, on disconnect_connection ; stdio/stdin n'ayant pas de protection de coupure, il utilise send().await.

Fichiers clés

codex-rs/app-server/src/main.rs:19-60 — AppServerArgs CLI, --listen parse le transport, --session-source distingue vscode / cli / mcp.codex-rs/app-server/src/lib.rs:449-540 — run_main_with_transport_options, entrée qui charge config, auth, otel.codex-rs/app-server/src/lib.rs:849-990 — spawn de MessageProcessor et boucle principale select, qui traite transport event et outbound envelope.codex-rs/app-server/src/message_processor.rs:224-302 — MessageProcessor::new, crée ThreadManager, ThreadStateManager, SkillsWatcher, les processor.codex-rs/app-server/src/message_processor.rs:799-862 — dispatch_initialized_client_request, logique de dispatch par serialization scope.codex-rs/app-server/src/message_processor.rs:864-1000 — le gros match de handle_initialized_client_request, route un ClientRequest vers les processor.codex-rs/app-server/src/transport.rs:134-172 — send_message_to_connection, disconnect en cas de file lente pleine.codex-rs/app-server-transport/src/transport/mod.rs:73-158 — l'énum AppServerTransport et from_listen_url.codex-rs/app-server-protocol/src/rpc.rs:34-88 — JSONRPCMessage / JSONRPCRequest / JSONRPCError, types de base du protocole (la note précise que ce n'est pas un vrai JSON-RPC 2.0, le champ jsonrpc n'est pas exigé).codex-rs/app-server/src/request_processors/initialize_processor.rs:44-101 — InitializeRequestProcessor::initialize, pose experimental_api, attestation, originator et autres identités globales de la connexion.

Le select principal bascule entre transport event et outbound envelope, avec un garde shutdown. La branche transport confie IncomingMessage à processor.process_request ; la branche outbound appelle route_outgoing_envelope :

// app-server/src/lib.rs:877-892 — 主循环 shutdown 判定
let exit_reason = loop {
    let running_turn_count = { *running_turn_count_rx.borrow() };
    if matches!(
        shutdown_state.update(running_turn_count, connections.len()),
        ShutdownAction::Finish
    ) {
        transport_shutdown_token.cancel();
        let _ = outbound_control_tx
            .send(OutboundControlEvent::DisconnectAll)
            .await;
        break "shutdown_requested";
    }
    tokio::select! { /* ... */ }
};

dispatch_initialized_client_request convertit la requête en QueuedInitializedRequest puis décide file ou spawn. Même scope → obligatoirement sérialisé ; cross-scope seulement concurrent — deux connexions lançant chacune un turn ne se bloquent pas mutuellement :

// message_processor.rs:851-860 — scope 排队 vs 直接 spawn
if let Some(scope) = serialization_scope {
    let (key, access) = RequestSerializationQueueKey::from_scope(connection_id, scope);
    self.request_serialization_queues
        .enqueue(key, access, request)
        .await;
} else {
    tokio::spawn(async move { request.run().await; });
}

route_outgoing_envelope distingue envoi ciblé et broadcast : en broadcast, on parcourt les connexions déjà initialized, mais une notification expérimentale ne va qu'aux connexions ayant activé experimental_api ; opted_out_notification_methods permet à un client de se désabonner par nom de méthode :

// app-server/src/transport.rs:212-236 — Broadcast 路由
OutgoingEnvelope::Broadcast { message } => {
    let target_connections: Vec<ConnectionId> = connections
        .iter()
        .filter_map(|(connection_id, connection_state)| {
            if connection_state.initialized.load(Ordering::Acquire)
                && !should_skip_notification_for_connection(connection_state, &message)
            {
                Some(*connection_id)
            } else {
                None
            }
        })
        .collect();
    // ...
}

Flux de données

Limites et échecs

• Connexion lente éjectée : quand la file outbound d'une connexion websocket est pleine, try_send renvoie Full et on disconnect_connection, pour éviter qu'un client lent ne tue le server ; stdio ne send().await bloque que s'il n'a pas de disconnect_sender (codex-rs/app-server/src/transport.rs:154-172).
• Connexion non initialisée n'accepte qu'Initialize : la première ligne de dispatch_initialized_client_request vérifie session.initialized() ; si non initialisé, renvoie directement invalid_request("Not initialized") (codex-rs/app-server/src/message_processor.rs:806-808).
• Requêtes experimental gated : les requêtes avec experimental_reason() sont refusées sur une connexion sans experimental_api, et renvoient experimental_required_message (codex-rs/app-server/src/message_processor.rs:810-814).
• Shutdown attend la fin des turns : ShutdownState::update regarde running_turn_count et connections.len() ; Finish ne se déclenche que lorsque les deux sont à 0 ; un timeout évite d'attendre indéfiniment (codex-rs/app-server/src/lib.rs:877-892).
• Transport off est légal : aucun acceptor n'est démarré, seul tourne le remote control — c'est le chemin pour contrôler un codex distant en mode daemon headless (codex-rs/app-server/src/lib.rs:712-740).

Récapitulatif

app-server est la couche protocole de codex : transports multiples, processor chacun sa zone, serialization scope garantit la sérialisation au sein d'un même thread. ConnectionSessionState utilise OnceLock pour exprimer la sémantique « initialisé une fois, immuable ensuite ». Pour aller plus loin : pairage distant et cycle de vie daemon dans le crate app-server-daemon, encapsulation client dans app-server-client ; pour voir comment les tool calls MCP entrent dans core via cette couche, voir Intégration MCP.