fix(remote-control): honor retries while connecting

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

@@ -52,6 +52,7 @@ use std::sync::atomic::Ordering;
 use tokio::sync::Semaphore;
 use tokio::sync::SemaphorePermit;
 use tokio::sync::mpsc;
+use tokio::sync::mpsc::error::TrySendError;
 use tokio::sync::oneshot;
 use tokio::sync::watch;
 use tokio::task::JoinHandle;
@@ -65,6 +66,8 @@ pub struct RemoteControlStartConfig {
     pub installation_id: String,
 }
 
+const RECONNECT_CHANNEL_CAPACITY: usize = 1;
+
 pub(super) struct QueuedServerEnvelope {
     pub(super) event: ServerEvent,
     pub(super) client_id: ClientId,
@@ -75,7 +78,7 @@ pub(super) struct QueuedServerEnvelope {
 #[derive(Clone)]
 pub struct RemoteControlHandle {
     enabled_tx: Arc<watch::Sender<bool>>,
-    reconnect_tx: mpsc::UnboundedSender<u64>,
+    reconnect_tx: mpsc::Sender<u64>,
     next_reconnect_generation: Arc<AtomicU64>,
     status_tx: Arc<watch::Sender<RemoteControlStatusChangedNotification>>,
     state_db_available: bool,
@@ -263,33 +266,33 @@ impl RemoteControlHandle {
                 response = Some(Err(RemoteControlReconnectUnavailable::Disabled));
                 return false;
             }
-            if self.reconnect_tx.is_closed() {
-                response = Some(Err(RemoteControlReconnectUnavailable::WorkerUnavailable));
-                return false;
-            }
-            if status.status == RemoteControlConnectionStatus::Connecting {
-                response = Some(Ok(status.clone()));
-                return false;
-            }
-
+            let reconnect_permit = match self.reconnect_tx.try_reserve() {
+                Ok(reconnect_permit) => reconnect_permit,
+                Err(TrySendError::Full(())) => {
+                    response = Some(Ok(status.clone()));
+                    return false;
+                }
+                Err(TrySendError::Closed(())) => {
+                    response = Some(Err(RemoteControlReconnectUnavailable::WorkerUnavailable));
+                    return false;
+                }
+            };
             let generation = self
                 .next_reconnect_generation
                 .fetch_add(1, Ordering::Relaxed)
                 .wrapping_add(1);
-            if self.reconnect_tx.send(generation).is_err() {
-                response = Some(Err(RemoteControlReconnectUnavailable::WorkerUnavailable));
-                return false;
-            }
+            reconnect_permit.send(generation);
 
             let next_status = remote_control_status_with_connection_status(
                 status,
                 RemoteControlConnectionStatus::Connecting,
             );
+            let status_changed = next_status != *status;
             reconnect_generation = Some(generation);
             previous_status = Some(status.status);
             *status = next_status.clone();
             response = Some(Ok(next_status));
-            true
+            status_changed
         });
 
         let response = response.expect("remote control reconnect must produce a response");
@@ -321,7 +324,7 @@ impl RemoteControlHandle {
                 environment_id = ?status.environment_id,
                 installation_id = %status.installation_id,
                 server_name = %status.server_name,
-                "remote control reconnect coalesced with an existing connection attempt"
+                "remote control reconnect coalesced with a pending request"
             );
         }
         response
@@ -887,7 +890,7 @@ pub async fn start_remote_control(
     };
 
     let (enabled_tx, enabled_rx) = watch::channel(initial_enabled);
-    let (reconnect_tx, reconnect_rx) = mpsc::unbounded_channel();
+    let (reconnect_tx, reconnect_rx) = mpsc::channel(RECONNECT_CHANNEL_CAPACITY);
     let current_enrollment = Arc::new(RemoteControlEnrollmentState::new(/*enrollment*/ None));
     let websocket_current_enrollment = current_enrollment.clone();
     let pairing_persistence_key_required = app_server_client_name_rx.is_some();

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

@@ -139,9 +139,9 @@ fn test_server_name() -> String {
 fn remote_control_handle_with_reconnect_receiver(
     remote_control_url: &str,
     auth_manager: Arc<AuthManager>,
-) -> (RemoteControlHandle, mpsc::UnboundedReceiver<u64>) {
+) -> (RemoteControlHandle, mpsc::Receiver<u64>) {
     let (enabled_tx, _enabled_rx) = watch::channel(/*init*/ true);
-    let (reconnect_tx, reconnect_rx) = mpsc::unbounded_channel();
+    let (reconnect_tx, reconnect_rx) = mpsc::channel(RECONNECT_CHANNEL_CAPACITY);
     let (status_tx, _status_rx) = watch::channel(RemoteControlStatusChangedNotification {
         status: RemoteControlConnectionStatus::Connecting,
         server_name: test_server_name(),
@@ -203,7 +203,35 @@ fn remote_control_reconnect_rejects_unavailable_worker() {
 }
 
 #[test]
-fn remote_control_reconnect_coalesces_while_connecting() {
+fn remote_control_reconnect_rejects_unavailable_state_db() {
+    let (mut handle, _reconnect_rx) = remote_control_handle_with_reconnect_receiver(
+        "http://127.0.0.1:1/backend-api/",
+        remote_control_auth_manager(),
+    );
+    handle.state_db_available = false;
+
+    assert_eq!(
+        handle.reconnect(),
+        Err(RemoteControlReconnectUnavailable::StateDbUnavailable)
+    );
+}
+
+#[test]
+fn remote_control_reconnect_rejects_disabled_remote_control() {
+    let (handle, _reconnect_rx) = remote_control_handle_with_reconnect_receiver(
+        "http://127.0.0.1:1/backend-api/",
+        remote_control_auth_manager(),
+    );
+    handle.disable();
+
+    assert_eq!(
+        handle.reconnect(),
+        Err(RemoteControlReconnectUnavailable::Disabled)
+    );
+}
+
+#[test]
+fn remote_control_reconnect_coalesces_pending_requests() {
     let (handle, mut reconnect_rx) = remote_control_handle_with_reconnect_receiver(
         "http://127.0.0.1:1/backend-api/",
         remote_control_auth_manager(),
@@ -221,6 +249,26 @@ fn remote_control_reconnect_coalesces_while_connecting() {
     assert!(reconnect_rx.is_empty());
 }
 
+#[test]
+fn remote_control_reconnect_queues_retry_after_worker_receives_request() {
+    let (handle, mut reconnect_rx) = remote_control_handle_with_reconnect_receiver(
+        "http://127.0.0.1:1/backend-api/",
+        remote_control_auth_manager(),
+    );
+    handle.publish_status(RemoteControlConnectionStatus::Connected);
+
+    let first = handle.reconnect().expect("first reconnect should queue");
+    assert_eq!(reconnect_rx.try_recv(), Ok(1));
+    let second = handle
+        .reconnect()
+        .expect("retry should queue after the worker receives the first request");
+
+    assert_eq!(first.status, RemoteControlConnectionStatus::Connecting);
+    assert_eq!(second, first);
+    assert_eq!(reconnect_rx.try_recv(), Ok(2));
+    assert!(reconnect_rx.is_empty());
+}
+
 fn remote_control_server_token_response(
     server_id: &str,
     environment_id: &str,
@@ -718,6 +766,7 @@ async fn remote_control_handle_reconnects_without_disabling_or_reenrolling() {
 
     let client_id = ClientId("client-1".to_string());
     let stream_id = StreamId("stream-1".to_string());
+    let subscribe_cursor = "cursor-1".to_string();
     let initialize_message = JSONRPCMessage::Request(codex_app_server_protocol::JSONRPCRequest {
         id: codex_app_server_protocol::RequestId::Integer(1),
         method: "initialize".to_string(),
@@ -738,7 +787,7 @@ async fn remote_control_handle_reconnects_without_disabling_or_reenrolling() {
             client_id: client_id.clone(),
             stream_id: Some(stream_id.clone()),
             seq_id: Some(1),
-            cursor: None,
+            cursor: Some(subscribe_cursor.clone()),
         },
     )
     .await;
@@ -814,12 +863,6 @@ async fn remote_control_handle_reconnects_without_disabling_or_reenrolling() {
         remote_handle.reconnect().expect("reconnect should succeed"),
         connecting_status
     );
-    assert_eq!(
-        remote_handle
-            .reconnect()
-            .expect("duplicate reconnect should coalesce"),
-        connecting_status
-    );
     expect_remote_control_status_snapshot(&mut status_rx, connecting_status).await;
     timeout(Duration::from_secs(1), first_websocket.next())
         .await
@@ -831,6 +874,12 @@ async fn remote_control_handle_reconnects_without_disabling_or_reenrolling() {
         second_handshake_request.headers.get("authorization"),
         Some(&format!("Bearer {TEST_REMOTE_CONTROL_SERVER_TOKEN}"))
     );
+    assert_eq!(
+        second_handshake_request
+            .headers
+            .get("x-codex-subscribe-cursor"),
+        Some(&subscribe_cursor)
+    );
     expect_remote_control_status_snapshot(
         &mut status_rx,
         RemoteControlStatusChangedNotification {

codex-rs/app-server-transport/src/transport/remote_control/tests/clients_tests.rs

@@ -14,7 +14,7 @@ fn client_management_handle(
     auth_manager: Arc<AuthManager>,
 ) -> RemoteControlHandle {
     let (enabled_tx, _enabled_rx) = watch::channel(/*init*/ false);
-    let (reconnect_tx, _reconnect_rx) = mpsc::unbounded_channel();
+    let (reconnect_tx, _reconnect_rx) = mpsc::channel(/*buffer*/ 1);
     let (status_tx, _status_rx) = watch::channel(RemoteControlStatusChangedNotification {
         status: RemoteControlConnectionStatus::Disabled,
         server_name: test_server_name(),