App-server-Architektur

源码版本rust-v0.145.0

app-server ist der JSON-RPC-Server von codex, der ThreadManager, AuthManager und ConfigManager von codex-core in einen Prozess verpackt, der von IDE, Web und TUI gemeinsam genutzt wird. Er exponiert nach außen drei Transporte — stdio / Unix-Socket / WebSocket — und nutzt intern einen MessageProcessor für das Request-Routing; alle RPCs laufen über dasselbe ClientRequest / ServerNotification-Protokoll.

Verantwortlichkeiten

1. Einstieg run_main_with_transport_options lädt config, auth, otel und state_db, startet Transport-Acceptor und Outbound-Router und spawnt zuletzt den MessageProcessor in die Haupt-select-Schleife (codex-rs/app-server/src/lib.rs:449-460).
2. Die Transport-Schicht vereinheitlicht stdio:// / unix:// / ws://IP:PORT / off zu TransportEvent (ConnectionOpened / ConnectionClosed / IncomingMessage); alle Verbindungen teilen sich einen Event-Channel (codex-rs/app-server-transport/src/transport/mod.rs:73-78).
3. MessageProcessor::process_request deserialisiert JSONRPCRequest in ClientRequest, dispatcht nach Initialize / bereits-initialisiert; bereits initialisierte Requests gehen über dispatch_initialized_client_request in den jeweiligen Processor (codex-rs/app-server/src/message_processor.rs:518-570).
4. Jeder initialisierte Request wird nach serialization_scope eingereiht oder gespawnt: RequestSerializationQueues stellt sicher, dass Requests mit gleichem Scope seriell laufen (z. B. Turn-Operationen desselben Threads); nur Scope-übergreifend wird nebenläufig gearbeitet (codex-rs/app-server/src/message_processor.rs:846-861).
5. Ausgehend läuft über OutgoingEnvelope::ToConnection / Broadcast; Broadcast geht nur an initialized-Verbindungen, und experimental-Benachrichtigungen nur an Verbindungen mit aktiviertem experimental_api (codex-rs/app-server/src/transport.rs:198-237).

Entwurfsbeweggründe

Dass app-server ein eigener Crate ist, liegt daran, dass IDE (VSCode, JetBrains), TUI und Remote-Pairing-App dasselbe Protokoll teilen, aber in unterschiedlichen Umgebungen laufen: IDE über stdio, Remote-Pairing über WebSocket, daemon-Modus über Unix-Socket + PID-Datei. Transport ist im app-server-transport-Crate untergebracht, das Protokoll im app-server-protocol-Crate; der Server-Crate kümmert sich nur um „wie JSON-RPC an den ThreadManager geroutet wird", sodass ein neuer Transport die Hauptlogik nicht anfasst.

ConnectionSessionState nutzt OnceLock<InitializedConnectionSessionState> statt Mutex, weil initialize ein einmaliger Vorgang ist — einmal gesetzt, unveränderlich; spätere Requests lesen nur. So entfällt der Lock-Overhead für „initialisierte Verbindung"; experimental_api_enabled / opted_out_notification_methods werden direkt über Atomic / RwLock gelesen.

Request-Routing nutzt eine Menge Processor-Structs statt eines einzelnen großen match: Jeder Processor (ThreadRequestProcessor, TurnRequestProcessor, ConfigRequestProcessor usw.) hält seinen eigenen Zustand; handle_initialized_client_request macht nur das Match-Dispatching. So kommt eine neue Methode ohne Änderung an zentralem Code aus, und Tests innerhalb des Processors sind leichter zu schreiben.

serialization_scope ist ein interessantes Design: Turn-Operationen desselben Threads müssen seriell laufen (sonst geraten zwei Op::UserInput in falsche Reihenfolge), aber Requests verschiedener Threads können nebenläufig sein. RequestSerializationQueues schlüsselt nach (connection_id, scope); gleichschlüsselige Requests reihen sich ein, unterschiedschlüssige werden direkt tokio::spawn-ed.

Beim Outbound-Routing verhindert try_send + bei vollem Queue disconnect, dass ein langsamer Client den Server blockiert: Eine langsame WebSocket-Verbindung mit vollem Queue wird direkt gekickt; stdio/stdin ohne disconnect_sender nutzt send().await.

Wichtige Dateien

codex-rs/app-server/src/main.rs:19-60 — CLI AppServerArgs; --listen parst den Transport, --session_source unterscheidet vscode / cli / mcp.codex-rs/app-server/src/lib.rs:449-540 — run_main_with_transport_options; Einstieg zum Laden von config, auth und otel.codex-rs/app-server/src/lib.rs:849-990 — MessageProcessor-Spawn und Haupt-select-Schleife; behandelt Transport-Events und Outbound-Envelopes.codex-rs/app-server/src/message_processor.rs:224-302 — MessageProcessor::new: erzeugt ThreadManager, ThreadStateManager, SkillsWatcher und die einzelnen Processor.codex-rs/app-server/src/message_processor.rs:799-862 — dispatch_initialized_client_request; Serialization-Scope-Dispatchlogik.codex-rs/app-server/src/message_processor.rs:864-1000 — handle_initialized_client_request als großes Match; routet ClientRequest an die Processor.codex-rs/app-server/src/transport.rs:134-172 — send_message_to_connection; bei vollem Queue der langsamen Verbindung disconnect.codex-rs/app-server-transport/src/transport/mod.rs:73-158 — AppServerTransport-Enum und from_listen_url-Parser.codex-rs/app-server-protocol/src/rpc.rs:34-88 — JSONRPCMessage / JSONRPCRequest / JSONRPCError; grundlegende Protokolltypen (Kommentar stellt klar: kein echtes JSON-RPC 2.0, jsonrpc-Feld nicht verlangt).codex-rs/app-server/src/request_processors/initialize_processor.rs:44-101 — InitializeRequestProcessor::initialize; setzt experimental_api, attestation, originator und andere globale Identitäten der Verbindung.

Das Haupt-select wählt zwischen Transport-Event und Outbound-Envelope, mit Shutdown-Signal-Guard. Der Transport-Zweig übergibt IncomingMessage an processor.process_request; der Outbound-Zweig ruft route_outgoing_envelope auf:

// 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 wandelt den Request in einen QueuedInitializedRequest um und entscheidet über Einreihen oder Spawnen. Gleichschlüsselige Scope müssen seriell laufen, unterschiedschlüssige nebenläufig — zwei Connections, die je einen Turn abschicken, blockieren sich nicht gegenseitig:

// 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 unterscheidet zwischen Direktversand und Broadcast: Bei Broadcast werden alle initialized-Verbindungen iteriert, aber experimental-Benachrichtigungen gehen nur an Verbindungen mit experimental_api; opted_out_notification_methods erlaubt dem Client, nach Methodennamen abzubestellen:

// 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();
    // ...
}

Datenfluss

Grenzen und Fehler

• Langsame Verbindung wird gekickt: Ist die Outbound-Queue einer WebSocket-Verbindung voll, liefert try_send Full, und es erfolgt direkt disconnect_connection, damit der Server nicht durch einen langsamen Client blockiert wird; nur stdio ohne disconnect_sender blockiert über send().await (codex-rs/app-server/src/transport.rs:154-172).
• Nicht-initialisierte Verbindung akzeptiert nur Initialize: dispatch_initialized_client_request prüft in der ersten Zeile session.initialized(); ohne Initialisierung gibt es direkt invalid_request("Not initialized") (codex-rs/app-server/src/message_processor.rs:806-808).
• experimental-Requests werden gegated: Requests mit experimental_reason() werden auf Verbindungen ohne experimental_api abgewiesen und liefern experimental_required_message (codex-rs/app-server/src/message_processor.rs:810-814).
• Shutdown wartet auf Turn-Abschluss: ShutdownState::update betrachtet running_turn_count und connections.len(); nur wenn beide 0 sind, Finish; mit Timeout gegen endloses Warten (codex-rs/app-server/src/lib.rs:877-892).
• Transport off ist ebenfalls gültig: Es startet keinen Acceptor, sondern läuft nur remote control — der Pfad für headless-Szenarien, in denen ein daemon Remote-codex steuert (codex-rs/app-server/src/lib.rs:712-740).

Zusammenfassung

app-server ist die Protokollschicht von codex: Transport in mehreren Formen, Processor je eigenverantwortlich, und serialization scope stellt sicher, dass gleiche Threads seriell laufen. ConnectionSessionState nutzt OnceLock, um die Semantik „einmal initialisiert, danach unveränderlich" auszudrücken. Für die nächsten Schichten: Remote-Pairing und daemon-Lifecycle liegen im app-server-daemon-Crate; die Client-Kapselung im app-server-client. Wie MCP-Werkzeugaufrufe hierüber in core gelangen, siehe MCP-Integration.