Exit the stdio app-server when its connection closes (#36035)

## Why

Closing stdin could leave the app-server running when a remote-control client
was still connected.

## What changed

Track each connection's origin and shut down a stdio app-server when its stdio
connection closes, regardless of whether other connections remain. Report the
shutdown reason as `stdio_connection_closed`.

## Testing

Added a regression test that closes stdio while a remote-control connection is
active and verifies that the app-server exits and disconnects the remote client.

GitOrigin-RevId: 51ab14d45dfea7f40a0657a8bb167b54efaf0e48

codex-rs/app-server/src/lib.rs

@@ -34,6 +34,7 @@ use crate::outgoing_message::OutgoingEnvelope;
 use crate::outgoing_message::OutgoingMessageSender;
 use crate::outgoing_message::QueuedOutgoingMessage;
 use crate::transport::CHANNEL_CAPACITY;
+use crate::transport::ConnectionOrigin;
 use crate::transport::ConnectionState;
 use crate::transport::OutboundConnectionState;
 use crate::transport::RemoteControlPolicy;
@@ -709,7 +710,6 @@ pub async fn run_main_with_transport_options(
     let mut transport_accept_handles = Vec::<JoinHandle<()>>::new();
 
     let single_client_mode = matches!(&transport, AppServerTransport::Stdio);
-    let shutdown_when_no_connections = single_client_mode;
     let graceful_signal_restart_enabled =
         runtime_options.install_shutdown_signal_handler && !single_client_mode;
     let mut app_server_client_name_rx = None;
@@ -991,6 +991,7 @@ pub async fn run_main_with_transport_options(
                                 let Some(connection_state) = connections.remove(&connection_id) else {
                                     continue;
                                 };
+                                let stdio_closed = connection_state.origin == ConnectionOrigin::Stdio;
                                 connection_state.session.rpc_gate.close().await;
                                 let outbound_closed = outbound_control_tx
                                     .send(OutboundControlEvent::Closed { connection_id })
@@ -1005,8 +1006,8 @@ pub async fn run_main_with_transport_options(
                                 if !outbound_closed {
                                     break "outbound_router_closed";
                                 }
-                                if shutdown_when_no_connections && connections.is_empty() {
-                                    break "last_connection_closed";
+                                if single_client_mode && stdio_closed {
+                                    break "stdio_connection_closed";
                                 }
                             }
                             TransportEvent::IncomingMessage { connection_id, message } => {

codex-rs/app-server/src/transport.rs

@@ -37,6 +37,7 @@ pub(crate) use codex_app_server_transport::start_websocket_acceptor;
 pub use codex_app_server_transport::take_remote_control_disabled_env;
 
 pub(crate) struct ConnectionState {
+    pub(crate) origin: ConnectionOrigin,
     pub(crate) outbound_initialized: Arc<AtomicBool>,
     pub(crate) outbound_experimental_api_enabled: Arc<AtomicBool>,
     pub(crate) outbound_opted_out_notification_methods: Arc<RwLock<HashSet<String>>>,
@@ -45,12 +46,13 @@ pub(crate) struct ConnectionState {
 
 impl ConnectionState {
     pub(crate) fn new(
-        _origin: ConnectionOrigin,
+        origin: ConnectionOrigin,
         outbound_initialized: Arc<AtomicBool>,
         outbound_experimental_api_enabled: Arc<AtomicBool>,
         outbound_opted_out_notification_methods: Arc<RwLock<HashSet<String>>>,
     ) -> Self {
         Self {
+            origin,
             outbound_initialized,
             outbound_experimental_api_enabled,
             outbound_opted_out_notification_methods,

codex-rs/app-server/tests/suite/logging.rs

@@ -33,7 +33,7 @@ fn standalone_app_server_emits_json_info_events() -> Result<()> {
             "level": "INFO",
             "fields": {
                 "message": "processor task exited",
-                "exit_reason": "last_connection_closed",
+                "exit_reason": "stdio_connection_closed",
                 "remaining_connection_count": 0,
                 "shutdown_forced": false,
             },

codex-rs/app-server/tests/suite/v2/remote_control.rs

@@ -44,6 +44,8 @@ use codex_protocol::protocol::SessionSource;
 use codex_state::RemoteControlEnrollmentRecord;
 use codex_state::StateRuntime;
 use codex_utils_cli::CliConfigOverrides;
+use futures::SinkExt;
+use futures::StreamExt;
 use pretty_assertions::assert_eq;
 use serial_test::serial;
 use tempfile::TempDir;
@@ -56,6 +58,8 @@ use tokio::net::TcpStream;
 use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
 use tokio::time::timeout;
+use tokio_tungstenite::accept_async;
+use tokio_tungstenite::tungstenite::Message;
 
 const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
 const STARTUP_TIMEOUT: Duration = Duration::from_secs(30);
@@ -465,6 +469,27 @@ async fn remote_control_enable_returns_connecting_status() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn stdio_eof_exits_with_remote_control_connection() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let mut backend = ConnectedRemoteControlBackend::start(codex_home.path()).await?;
+    let mut app_server = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .without_auto_env()
+        .build_initialized()
+        .await?;
+
+    let request_id = app_server.send_remote_control_enable_request().await?;
+    let _: RemoteControlEnableResponse =
+        timeout(DEFAULT_TIMEOUT, app_server.read_response(request_id)).await??;
+    timeout(DEFAULT_TIMEOUT, backend.wait_until_initialized()).await??;
+
+    let status = timeout(DEFAULT_TIMEOUT, app_server.shutdown_gracefully()).await??;
+    assert!(status.success());
+    timeout(DEFAULT_TIMEOUT, backend.wait_for_disconnect()).await??;
+    Ok(())
+}
+
 #[tokio::test]
 async fn disable_waits_for_in_flight_durable_enable() -> Result<()> {
     let codex_home = TempDir::new()?;
@@ -806,11 +831,131 @@ struct BlockingRemoteControlBackend {
     server_task: JoinHandle<()>,
 }
 
+struct ConnectedRemoteControlBackend {
+    initialized_rx: Option<oneshot::Receiver<std::result::Result<(), String>>>,
+    server_task: JoinHandle<Result<()>>,
+}
+
 struct ClientManagementRemoteControlBackend {
     requests_rx: Option<oneshot::Receiver<Result<Vec<String>>>>,
     server_task: JoinHandle<()>,
 }
 
+impl ConnectedRemoteControlBackend {
+    async fn start(codex_home: &std::path::Path) -> Result<Self> {
+        let listener = configured_remote_control_listener(codex_home).await?;
+        let (initialized_tx, initialized_rx) = oneshot::channel();
+        let server_task = tokio::spawn(async move {
+            let mut initialized_tx = Some(initialized_tx);
+            let result: Result<()> = async {
+                let (_request_line, reader) = read_enroll_request(&listener).await?;
+                respond_with_json(
+                    reader.into_inner(),
+                    serde_json::json!({
+                        "server_id": "server-id",
+                        "environment_id": "environment-id",
+                        "remote_control_token": "remote-control-token",
+                        "expires_at": "3026-05-22T12:34:56Z",
+                    }),
+                )
+                .await?;
+
+                let (stream, _) = listener.accept().await?;
+                let mut websocket = accept_async(stream).await?;
+                websocket
+                    .send(Message::Text(
+                        serde_json::json!({
+                            "type": "client_message",
+                            "client_id": "client-id",
+                            "stream_id": "stream-id",
+                            "seq_id": 0,
+                            "message": {
+                                "id": 1,
+                                "method": "initialize",
+                                "params": {
+                                    "clientInfo": {
+                                        "name": "remote-test-client",
+                                        "version": "0.1.0",
+                                    },
+                                },
+                            },
+                        })
+                        .to_string()
+                        .into(),
+                    ))
+                    .await?;
+
+                loop {
+                    let message = websocket
+                        .next()
+                        .await
+                        .context("remote control disconnected before initialize response")??;
+                    let Message::Text(message) = message else {
+                        continue;
+                    };
+                    let message: serde_json::Value = serde_json::from_str(&message)?;
+                    if message["type"] == "server_message" && message["message"]["id"] == 1 {
+                        break;
+                    }
+                }
+
+                websocket
+                    .send(Message::Text(
+                        serde_json::json!({
+                            "type": "client_message",
+                            "client_id": "client-id",
+                            "stream_id": "stream-id",
+                            "seq_id": 1,
+                            "message": {
+                                "method": "initialized",
+                            },
+                        })
+                        .to_string()
+                        .into(),
+                    ))
+                    .await?;
+                if let Some(initialized_tx) = initialized_tx.take() {
+                    let _ = initialized_tx.send(Ok(()));
+                }
+
+                while let Some(message) = websocket.next().await {
+                    match message {
+                        Ok(Message::Close(_)) | Err(_) => break,
+                        Ok(_) => {}
+                    }
+                }
+                Ok(())
+            }
+            .await;
+
+            if let Err(err) = &result
+                && let Some(initialized_tx) = initialized_tx.take()
+            {
+                let _ = initialized_tx.send(Err(err.to_string()));
+            }
+            result
+        });
+
+        Ok(Self {
+            initialized_rx: Some(initialized_rx),
+            server_task,
+        })
+    }
+
+    async fn wait_until_initialized(&mut self) -> Result<()> {
+        self.initialized_rx
+            .take()
+            .context("remote control initialization should only be awaited once")?
+            .await?
+            .map_err(anyhow::Error::msg)
+    }
+
+    async fn wait_for_disconnect(&mut self) -> Result<()> {
+        (&mut self.server_task).await??;
+        Ok(())
+    }
+}
+
 impl ClientManagementRemoteControlBackend {
     async fn start(codex_home: &std::path::Path) -> Result<Self> {
         let listener = configured_remote_control_listener(codex_home).await?;
@@ -1043,6 +1188,12 @@ impl Drop for BlockingRemoteControlBackend {
     }
 }
 
+impl Drop for ConnectedRemoteControlBackend {
+    fn drop(&mut self) {
+        self.server_task.abort();
+    }
+}
+
 impl Drop for ClientManagementRemoteControlBackend {
     fn drop(&mut self) {
         self.server_task.abort();

codex-rs/cli/tests/app_server.rs

@@ -43,7 +43,7 @@ fn app_server_emits_json_info_events() -> Result<()> {
             "level": "INFO",
             "fields": {
                 "message": "processor task exited",
-                "exit_reason": "last_connection_closed",
+                "exit_reason": "stdio_connection_closed",
                 "remaining_connection_count": 0,
                 "shutdown_forced": false,
             },