Arquitectura de App-server

源码版本rust-v0.145.0

app-server es el server JSON-RPC de codex que envuelve el ThreadManager, AuthManager y ConfigManager de codex-core en un proceso compartible por IDE, Web y TUI. Hacia fuera expone tres transportes — stdio / unix socket / websocket —; hacia dentro usa MessageProcessor para enrutar peticiones, y todos los RPC pasan por el mismo protocolo ClientRequest / ServerNotification.

Responsabilidades

1. La entrada run_main_with_transport_options carga config, auth, otel y state_db, arranca el transport acceptor y el outbound router, y al final hace spawn de MessageProcessor para entrar al bucle select principal (codex-rs/app-server/src/lib.rs:449-460).
2. La capa de transport unifica stdio:// / unix:// / ws://IP:PORT / off en TransportEvent (ConnectionOpened / ConnectionClosed / IncomingMessage); todas las conexiones comparten un único canal de eventos (codex-rs/app-server-transport/src/transport/mod.rs:73-78).
3. MessageProcessor::process_request deserializa el JSONRPCRequest a ClientRequest, y lo dispatcha por Initialize / ya inicializado; lo ya inicializado entra al processor correspondiente por dispatch_initialized_client_request (codex-rs/app-server/src/message_processor.rs:518-570).
4. Cada petición ya inicializada se encola o se spawnea según serialization_scope: RequestSerializationQueues garantiza que las peticiones del mismo scope se ejecuten en serie (por ejemplo, operaciones de turn del mismo thread) y solo entre scopes distintos hay concurrencia (codex-rs/app-server/src/message_processor.rs:846-861).
5. La salida va por OutgoingEnvelope::ToConnection / Broadcast; el broadcast solo se envía a conexiones ya initialized; las notificaciones experimentales solo se envían a conexiones con experimental_api activado (codex-rs/app-server/src/transport.rs:198-237).

Motivación de diseño

app-server es un crate aparte porque IDE (VSCode, JetBrains), TUI y la app de emparejamiento remoto comparten el mismo protocolo pero corren en entornos distintos: IDE en stdio, emparejamiento remoto por websocket, daemon en modo unix socket + pid file. Sacar el transport al crate app-server-transport y el protocolo al crate app-server-protocol deja al crate server ocupándose solo de «cómo enrutar JSON-RPC al ThreadManager»; añadir un transport nuevo no toca la lógica principal.

ConnectionSessionState usa OnceLock<InitializedConnectionSessionState> en vez de Mutex, porque initialize es una operación única: una vez fijado, no se cambia, y las peticiones posteriores solo lo leen. Eso evita el coste de lock para «conexión ya inicializada»; experimental_api_enabled / opted_out_notification_methods se leen directo con atomic / RwLock.

El enrutado de peticiones usa una colección de structs processor y no un único match: cada processor (ThreadRequestProcessor, TurnRequestProcessor, ConfigRequestProcessor, etc.) sostiene su propio estado, y handle_initialized_client_request solo hace un match y dispatch. Añadir un método nuevo no toca el código central, y los tests internos de cada processor son más fáciles de escribir.

serialization_scope es un diseño interesante: las operaciones de turn del mismo thread tienen que ser en serie (si no, dos Op::UserInput se desordenarían), pero las peticiones de threads distintos pueden ir concurrentes. RequestSerializationQueues separa por clave (connection_id, scope); las peticiones con la misma clave se encolan, y entre claves distintas se hace tokio::spawn directo.

El try_send del enrutado outbound, con desconexión si la cola se llena, evita que un cliente lento mate el server: la cola del websocket de una conexión lenta, al llenarse, se desconecta directamente; stdio/stdin, que no tiene protección de corte, sí usa send().await.

Archivos clave

codex-rs/app-server/src/main.rs:19-60 — AppServerArgs de línea de comandos; --listen parsea el transport, --session-source distingue vscode / cli / mcp.codex-rs/app-server/src/lib.rs:449-540 — run_main_with_transport_options, entrada que carga config, auth y otel.codex-rs/app-server/src/lib.rs:849-990 — spawn de MessageProcessor y bucle select principal, gestionando transport event y outbound envelope.codex-rs/app-server/src/message_processor.rs:224-302 — MessageProcessor::new, crea ThreadManager, ThreadStateManager, SkillsWatcher y los processors.codex-rs/app-server/src/message_processor.rs:799-862 — dispatch_initialized_client_request, lógica de dispatch por serialization scope.codex-rs/app-server/src/message_processor.rs:864-1000 — el gran match de handle_initialized_client_request, enruta ClientRequest a cada processor.codex-rs/app-server/src/transport.rs:134-172 — send_message_to_connection, desconecta si la cola de la conexión lenta se llena.codex-rs/app-server-transport/src/transport/mod.rs:73-158 — enum AppServerTransport y from_listen_url para parsear.codex-rs/app-server-protocol/src/rpc.rs:34-88 — JSONRPCMessage / JSONRPCRequest / JSONRPCError, tipos básicos de protocolo (el comentario deja claro que no es JSON-RPC 2.0 estricto, no exige el campo jsonrpc).codex-rs/app-server/src/request_processors/initialize_processor.rs:44-101 — InitializeRequestProcessor::initialize, fija la identidad global de la conexión: experimental_api, attestation, originator, etc.

El select principal elige entre transport event y outbound envelope, con guard de señal de shutdown. La rama transport pasa el IncomingMessage a processor.process_request; la rama outbound llama a route_outgoing_envelope:

// app-server/src/lib.rs:877-892 — decisión de shutdown del bucle principal
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 convierte la petición en QueuedInitializedRequest y decide si encola o spawnea. El mismo scope tiene que ir en serie; entre scopes distintos, concurrencia — por ejemplo, dos conexiones enviando cada una su turn no se bloquean entre sí:

// message_processor.rs:851-860 — encolar por scope vs spawn directo
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 envío dirigido de broadcast: al hacer broadcast recorre todas las conexiones initialized, pero las notificaciones experimentales solo se mandan a las conexiones con experimental_api activado; opted_out_notification_methods deja al cliente desuscribirse por nombre de método:

// app-server/src/transport.rs:212-236 — ruta de 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();
    // ...
}

Flujo de datos

Bordes y fallos

• Conexión lenta desconectada: si la cola outbound del websocket se llena, try_send devuelve Full y se ejecuta disconnect_connection directo, evitando que el server muera por un cliente lento; stdio, sin disconnect_sender, usa send().await bloqueante (codex-rs/app-server/src/transport.rs:154-172).
• Conexión no inicializada solo acepta Initialize: la primera línea de dispatch_initialized_client_request comprueba session.initialized(); si no está inicializada, devuelve invalid_request("Not initialized") (codex-rs/app-server/src/message_processor.rs:806-808).
• Peticiones experimentales bajo gate: las peticiones con experimental_reason() se rechazan en conexiones sin experimental_api, devolviendo experimental_required_message (codex-rs/app-server/src/message_processor.rs:810-814).
• Shutdown espera al cierre de los turns: ShutdownState::update mira running_turn_count y connections.len(); solo hace Finish cuando ambos están a 0; con timeout para no esperar indefinidamente (codex-rs/app-server/src/lib.rs:877-892).
• transport off también es válido: no arranca ningún acceptor, solo corre remote control; es la ruta para controlar un codex remoto con daemon en un escenario headless (codex-rs/app-server/src/lib.rs:712-740).

Resumen

app-server es la capa de protocolo de codex: transportes múltiples, cada processor gestiona lo suyo, y serialization scope garantiza la serialización dentro del mismo thread. ConnectionSessionState usa OnceLock para expresar la semántica de «inicializado una vez, inmutable». Para seguir bajando: el emparejamiento remoto y el ciclo de vida del daemon están en el crate app-server-daemon; el wrapper de cliente, en app-server-client. Cómo las llamadas a herramientas MCP entran por aquí al core se ve en Integración MCP.