fix(app-server-transport): discard stale pairing responses

2026-10-21 17:48:44 +00:00 · 2026-05-29 07:32:36 -07:00
parent 53faa76492
commit 5b7077c8bb
6 changed files with 532 additions and 313 deletions
@@ -31,6 +31,8 @@ use std::io;
 use std::panic::AssertUnwindSafe;
 use std::sync::Arc;
 use std::sync::Mutex as StdMutex;
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::Ordering;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
 use tokio::sync::watch;
@@ -57,10 +59,25 @@ pub struct RemoteControlHandle {
    enabled_tx: Arc<watch::Sender<bool>>,
    status_tx: Arc<watch::Sender<RemoteControlStatusChangedNotification>>,
    state_db_available: bool,
-    pairing_client: Arc<StdMutex<Option<RemoteControlPairingClient>>>,
+    pairing: PairingClientState,
    auth_change_rx: Arc<StdMutex<watch::Receiver<u64>>>,
 }

+#[derive(Clone)]
+pub(super) struct PairingClientState {
+    client: Arc<StdMutex<Option<RemoteControlPairingClient>>>,
+    generation: Arc<AtomicU64>,
+}
+
+impl PairingClientState {
+    fn new() -> Self {
+        Self {
+            client: Arc::new(StdMutex::new(None)),
+            generation: Arc::new(AtomicU64::new(0)),
+        }
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct RemoteControlUnavailable;

@@ -115,7 +132,7 @@ impl RemoteControlHandle {
            *state = false;
            changed
        });
-        clear_pairing_client(&self.pairing_client);
+        clear_pairing_client(&self.pairing);

        let status = self.status();
        info!(
@@ -147,11 +164,15 @@ impl RemoteControlHandle {
                "remote control pairing requires remote control to be enabled",
            ));
        }
+        if self.status().status != RemoteControlConnectionStatus::Connected {
+            return Err(Self::pairing_unavailable_error());
+        }

        let auth_change_revision = self.auth_change_revision();
        let pairing_client = {
            let mut pairing_client = self
-                .pairing_client
+                .pairing
+                .client
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            let current_pairing_client = pairing_client
@@ -174,6 +195,9 @@ impl RemoteControlHandle {
        if self.auth_change_revision() != auth_change_revision {
            return Err(Self::pairing_unavailable_error());
        }
+        if pairing_response.is_ok() && !self.pairing_client_is_current(&pairing_client) {
+            return Err(Self::pairing_unavailable_error());
+        }
        pairing_response
    }

@@ -192,6 +216,20 @@ impl RemoteControlHandle {
        )
    }

+    fn pairing_client_is_current(&self, pairing_client: &RemoteControlPairingClient) -> bool {
+        *self.enabled_tx.borrow()
+            && self.status().status == RemoteControlConnectionStatus::Connected
+            && self
+                .pairing
+                .client
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .as_ref()
+                .is_some_and(|current_pairing_client| {
+                    current_pairing_client.generation() == pairing_client.generation()
+                })
+    }
+
    fn publish_status(
        &self,
        connection_status: RemoteControlConnectionStatus,
@@ -239,10 +277,12 @@ fn remote_control_status_with_connection_status(
    }
 }

-fn clear_pairing_client(pairing_client: &Arc<StdMutex<Option<RemoteControlPairingClient>>>) {
-    *pairing_client
+fn clear_pairing_client(pairing: &PairingClientState) {
+    *pairing
+        .client
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
+    pairing.generation.fetch_add(1, Ordering::Relaxed);
 }

 pub async fn start_remote_control(
@@ -267,8 +307,8 @@ pub async fn start_remote_control(
    };

    let (enabled_tx, enabled_rx) = watch::channel(initial_enabled);
-    let pairing_client = Arc::new(StdMutex::new(None));
-    let websocket_pairing_client = Arc::clone(&pairing_client);
+    let pairing = PairingClientState::new();
+    let websocket_pairing = pairing.clone();
    let auth_change_rx = Arc::new(StdMutex::new(auth_manager.auth_change_receiver()));
    let server_name = gethostname().to_string_lossy().trim().to_string();
    let remote_control_url = config.remote_control_url;
@@ -317,7 +357,7 @@ pub async fn start_remote_control(
            RemoteControlChannels {
                transport_event_tx,
                status_publisher,
-                pairing_client: websocket_pairing_client,
+                pairing: websocket_pairing,
            },
            shutdown_token,
            enabled_rx,
@@ -362,7 +402,7 @@ pub async fn start_remote_control(
            enabled_tx: Arc::new(enabled_tx),
            status_tx: Arc::new(status_tx),
            state_db_available,
-            pairing_client,
+            pairing,
            auth_change_rx,
        },
    ))
@@ -20,6 +20,7 @@ pub(super) struct RemoteControlPairingClient {
    environment_id: String,
    expires_at: OffsetDateTime,
    auth_change_revision: u64,
+    generation: u64,
 }

 impl RemoteControlPairingClient {
@@ -30,6 +31,7 @@ impl RemoteControlPairingClient {
        environment_id: String,
        expires_at: OffsetDateTime,
        auth_change_revision: u64,
+        generation: u64,
    ) -> Self {
        Self {
            pairing_url: remote_control_target.pair_url.clone(),
@@ -38,6 +40,7 @@ impl RemoteControlPairingClient {
            environment_id,
            expires_at,
            auth_change_revision,
+            generation,
        }
    }

@@ -45,6 +48,10 @@ impl RemoteControlPairingClient {
        self.auth_change_revision == auth_change_revision
    }

+    pub(super) fn generation(&self) -> u64 {
+        self.generation
+    }
+
    pub(super) async fn start(
        &self,
        request: StartRemoteControlPairingRequest,
@@ -0,0 +1,396 @@
+use super::*;
+use pretty_assertions::assert_eq;
+
+#[tokio::test]
+async fn remote_control_handle_disable_clears_stale_pairing_client() {
+    let remote_handle = remote_control_handle_with_pairing_client(
+        TEST_REMOTE_CONTROL_URL,
+        watch::channel(/*init*/ 0u64).1,
+    );
+
+    assert_eq!(
+        remote_handle.disable(),
+        RemoteControlStatusChangedNotification {
+            status: RemoteControlConnectionStatus::Disabled,
+            server_name: test_server_name(),
+            installation_id: TEST_INSTALLATION_ID.to_string(),
+            environment_id: None,
+        }
+    );
+    remote_handle.enable().expect("enable should succeed");
+    assert_eq!(
+        remote_handle
+            .start_pairing(RemoteControlPairingStartParams { manual_code: false })
+            .await
+            .expect_err("re-enabled remote control should wait for refreshed pairing auth")
+            .to_string(),
+        "remote control pairing is unavailable until enrollment completes"
+    );
+}
+
+#[tokio::test]
+async fn remote_control_disable_during_websocket_handshake_clears_pairing_client() {
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("listener should bind");
+    let remote_control_url = remote_control_url_for_listener(&listener);
+    let codex_home = TempDir::new().expect("temp dir should create");
+    let (transport_event_tx, _transport_event_rx) =
+        mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
+    let shutdown_token = CancellationToken::new();
+    let (remote_task, remote_handle) = start_remote_control(
+        RemoteControlStartConfig {
+            remote_control_url,
+            installation_id: TEST_INSTALLATION_ID.to_string(),
+        },
+        Some(remote_control_state_runtime(&codex_home).await),
+        remote_control_auth_manager(),
+        transport_event_tx,
+        shutdown_token.clone(),
+        /*app_server_client_name_rx*/ None,
+        /*initial_enabled*/ true,
+    )
+    .await
+    .expect("remote control should start");
+
+    let enroll_request = accept_http_request(&listener).await;
+    respond_with_json(
+        enroll_request.stream,
+        remote_control_server_token_response(
+            "srv_e_test",
+            "env_test",
+            TEST_REMOTE_CONTROL_SERVER_TOKEN,
+        ),
+    )
+    .await;
+
+    let (stream, _) = timeout(Duration::from_secs(5), listener.accept())
+        .await
+        .expect("remote control should connect in time")
+        .expect("listener accept should succeed");
+    let remote_handle_for_callback = remote_handle.clone();
+    let mut websocket = accept_hdr_async(
+        stream,
+        move |_request: &tungstenite::handshake::server::Request,
+              response: tungstenite::handshake::server::Response| {
+            remote_handle_for_callback.disable();
+            Ok(response)
+        },
+    )
+    .await
+    .expect("websocket handshake should succeed");
+    expect_remote_control_connection_closed(
+        &mut websocket,
+        "disabled remote control should close the websocket handshake",
+    )
+    .await;
+    assert!(
+        remote_handle
+            .pairing
+            .client
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .is_none(),
+        "disabling during websocket handshake should clear pairing auth"
+    );
+
+    shutdown_token.cancel();
+    let _ = remote_task.await;
+}
+
+#[tokio::test]
+async fn remote_control_handle_rejects_pairing_client_after_auth_change() {
+    let (auth_change_tx, auth_change_rx) = watch::channel(/*init*/ 0u64);
+    let remote_handle =
+        remote_control_handle_with_pairing_client(TEST_REMOTE_CONTROL_URL, auth_change_rx);
+    auth_change_tx.send_modify(|revision| *revision += 1);
+
+    assert_eq!(
+        remote_handle
+            .start_pairing(RemoteControlPairingStartParams::default())
+            .await
+            .expect_err("pairing should wait for current-account enrollment")
+            .to_string(),
+        "remote control pairing is unavailable until enrollment completes"
+    );
+}
+
+#[tokio::test]
+async fn remote_control_handle_discards_pairing_response_after_auth_change() {
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("listener should bind");
+    let remote_control_url = remote_control_url_for_listener(&listener);
+    let codex_home = TempDir::new().expect("temp dir should create");
+    save_auth(
+        codex_home.path(),
+        &remote_control_auth_dot_json(Some("account_id")),
+        AuthCredentialsStoreMode::File,
+    )
+    .expect("initial auth should save");
+    let auth_manager = AuthManager::shared(
+        codex_home.path().to_path_buf(),
+        /*enable_codex_api_key_env*/ false,
+        AuthCredentialsStoreMode::File,
+        /*chatgpt_base_url*/ None,
+    )
+    .await;
+    let remote_handle = remote_control_handle_with_pairing_client(
+        &remote_control_url,
+        auth_manager.auth_change_receiver(),
+    );
+    let pairing_task = tokio::spawn({
+        let remote_handle = remote_handle.clone();
+        async move {
+            remote_handle
+                .start_pairing(RemoteControlPairingStartParams::default())
+                .await
+        }
+    });
+
+    let pairing_request = accept_http_request(&listener).await;
+    assert_eq!(
+        pairing_request.request_line,
+        "POST /backend-api/wham/remote/control/server/pair HTTP/1.1"
+    );
+    assert_eq!(
+        pairing_request.headers.get("authorization"),
+        Some(&format!("Bearer {TEST_REMOTE_CONTROL_SERVER_TOKEN}"))
+    );
+    assert_eq!(
+        serde_json::from_str::<serde_json::Value>(&pairing_request.body)
+            .expect("pairing request body should deserialize"),
+        json!({ "manual_code": false })
+    );
+
+    save_auth(
+        codex_home.path(),
+        &remote_control_auth_dot_json(Some("next_account_id")),
+        AuthCredentialsStoreMode::File,
+    )
+    .expect("next auth should save");
+    auth_manager.reload().await;
+    respond_with_json(
+        pairing_request.stream,
+        json!({
+            "pairing_code": "stale-pairing-code",
+            "manual_pairing_code": "ABCD-EFGH",
+            "server_id": "srv_e_test",
+            "environment_id": "env_test",
+            "expires_at": "3026-05-22T12:34:56Z",
+        }),
+    )
+    .await;
+
+    assert_eq!(
+        pairing_task
+            .await
+            .expect("pairing task should join")
+            .expect_err("stale pairing response should be discarded")
+            .to_string(),
+        "remote control pairing is unavailable until enrollment completes"
+    );
+}
+
+#[tokio::test]
+async fn remote_control_handle_discards_pairing_response_after_disable() {
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("listener should bind");
+    let remote_control_url = remote_control_url_for_listener(&listener);
+    let remote_handle = remote_control_handle_with_pairing_client(
+        &remote_control_url,
+        watch::channel(/*init*/ 0u64).1,
+    );
+    let pairing_task = tokio::spawn({
+        let remote_handle = remote_handle.clone();
+        async move {
+            remote_handle
+                .start_pairing(RemoteControlPairingStartParams::default())
+                .await
+        }
+    });
+
+    let pairing_request = accept_http_request(&listener).await;
+    remote_handle.disable();
+    respond_with_json(
+        pairing_request.stream,
+        json!({
+            "pairing_code": "stale-pairing-code",
+            "manual_pairing_code": "ABCD-EFGH",
+            "server_id": "srv_e_test",
+            "environment_id": "env_test",
+            "expires_at": "3026-05-22T12:34:56Z",
+        }),
+    )
+    .await;
+
+    assert_eq!(
+        pairing_task
+            .await
+            .expect("pairing task should join")
+            .expect_err("disabled remote control should discard pairing response")
+            .to_string(),
+        "remote control pairing is unavailable until enrollment completes"
+    );
+}
+
+#[tokio::test]
+async fn remote_control_handle_keeps_pairing_response_after_pairing_auth_refresh() {
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("listener should bind");
+    let remote_control_url = remote_control_url_for_listener(&listener);
+    let remote_handle = remote_control_handle_with_pairing_client(
+        &remote_control_url,
+        watch::channel(/*init*/ 0u64).1,
+    );
+    let pairing_task = tokio::spawn({
+        let remote_handle = remote_handle.clone();
+        async move {
+            remote_handle
+                .start_pairing(RemoteControlPairingStartParams::default())
+                .await
+        }
+    });
+
+    let pairing_request = accept_http_request(&listener).await;
+    let generation = remote_handle
+        .pairing
+        .generation
+        .load(std::sync::atomic::Ordering::Relaxed);
+    *remote_handle
+        .pairing
+        .client
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner) =
+        Some(RemoteControlPairingClient::new(
+            &normalize_remote_control_url(&remote_control_url)
+                .expect("remote control url should normalize"),
+            TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN.to_string(),
+            "srv_e_test".to_string(),
+            "env_test".to_string(),
+            OffsetDateTime::parse(
+                TEST_REMOTE_CONTROL_SERVER_TOKEN_EXPIRES_AT,
+                &time::format_description::well_known::Rfc3339,
+            )
+            .expect("server token expiry should parse"),
+            /*auth_change_revision*/ 0,
+            generation,
+        ));
+    respond_with_json(
+        pairing_request.stream,
+        json!({
+            "pairing_code": "fresh-pairing-code",
+            "manual_pairing_code": "ABCD-EFGH",
+            "server_id": "srv_e_test",
+            "environment_id": "env_test",
+            "expires_at": "3026-05-22T12:34:56Z",
+        }),
+    )
+    .await;
+
+    assert_eq!(
+        pairing_task
+            .await
+            .expect("pairing task should join")
+            .expect("pairing response should be kept")
+            .pairing_code,
+        "fresh-pairing-code"
+    );
+}
+
+#[tokio::test]
+async fn remote_control_handle_clears_pairing_client_after_auth_change() {
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("listener should bind");
+    let remote_control_url = remote_control_url_for_listener(&listener);
+    let codex_home = TempDir::new().expect("temp dir should create");
+    save_auth(
+        codex_home.path(),
+        &remote_control_auth_dot_json(Some("account_id")),
+        AuthCredentialsStoreMode::File,
+    )
+    .expect("initial auth should save");
+    let auth_manager = AuthManager::shared(
+        codex_home.path().to_path_buf(),
+        /*enable_codex_api_key_env*/ false,
+        AuthCredentialsStoreMode::File,
+        /*chatgpt_base_url*/ None,
+    )
+    .await;
+    let (transport_event_tx, _transport_event_rx) =
+        mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
+    let shutdown_token = CancellationToken::new();
+    let (remote_task, remote_handle) = start_remote_control(
+        RemoteControlStartConfig {
+            remote_control_url,
+            installation_id: TEST_INSTALLATION_ID.to_string(),
+        },
+        Some(remote_control_state_runtime(&codex_home).await),
+        auth_manager.clone(),
+        transport_event_tx,
+        shutdown_token.clone(),
+        /*app_server_client_name_rx*/ None,
+        /*initial_enabled*/ true,
+    )
+    .await
+    .expect("remote control should start");
+
+    let enroll_request = accept_http_request(&listener).await;
+    respond_with_json(
+        enroll_request.stream,
+        remote_control_server_token_response(
+            "srv_e_initial",
+            "env_initial",
+            TEST_REMOTE_CONTROL_SERVER_TOKEN,
+        ),
+    )
+    .await;
+    let mut first_websocket = accept_remote_control_connection(&listener).await;
+
+    save_auth(
+        codex_home.path(),
+        &remote_control_auth_dot_json(Some("next_account_id")),
+        AuthCredentialsStoreMode::File,
+    )
+    .expect("next auth should save");
+    auth_manager.reload().await;
+    expect_remote_control_connection_closed(
+        &mut first_websocket,
+        "auth change should close the stale websocket",
+    )
+    .await;
+    assert_eq!(
+        remote_handle
+            .start_pairing(RemoteControlPairingStartParams::default())
+            .await
+            .expect_err("pairing should wait for current-account enrollment")
+            .to_string(),
+        "remote control pairing is unavailable until enrollment completes"
+    );
+
+    let enroll_request = accept_http_request(&listener).await;
+    assert_eq!(
+        enroll_request.request_line,
+        "POST /backend-api/wham/remote/control/server/enroll HTTP/1.1"
+    );
+    respond_with_json(
+        enroll_request.stream,
+        remote_control_server_token_response(
+            "srv_e_next",
+            "env_next",
+            TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN,
+        ),
+    )
+    .await;
+    let mut second_websocket = accept_remote_control_connection(&listener).await;
+    second_websocket
+        .close(None)
+        .await
+        .expect("second websocket should close");
+
+    shutdown_token.cancel();
+    let _ = remote_task.await;
+}
@@ -94,19 +94,7 @@ async fn start_remote_control_pairing_uses_server_token_and_maps_response() {
            .await
            .expect("response should write");
    });
-    let client = RemoteControlPairingClient::new(
-        &RemoteControlTarget {
-            websocket_url: "ws://unused".to_string(),
-            enroll_url: "http://unused".to_string(),
-            refresh_url: "http://unused".to_string(),
-            pair_url,
-        },
-        "remote-control-token".to_string(),
-        "server-id".to_string(),
-        "environment-id".to_string(),
-        OffsetDateTime::from_unix_timestamp(33_336_362_096).expect("future timestamp should parse"),
-        /*auth_change_revision*/ 0,
-    );
+    let client = pairing_client(pair_url);

    let response = client
        .start(StartRemoteControlPairingRequest { manual_code: true })
@@ -163,19 +151,7 @@ async fn start_remote_control_pairing_preserves_backend_error_context() {
            .await
            .expect("response should write");
    });
-    let client = RemoteControlPairingClient::new(
-        &RemoteControlTarget {
-            websocket_url: "ws://unused".to_string(),
-            enroll_url: "http://unused".to_string(),
-            refresh_url: "http://unused".to_string(),
-            pair_url,
-        },
-        "remote-control-token".to_string(),
-        "server-id".to_string(),
-        "environment-id".to_string(),
-        OffsetDateTime::from_unix_timestamp(33_336_362_096).expect("future timestamp should parse"),
-        /*auth_change_revision*/ 0,
-    );
+    let client = pairing_client(pair_url);

    let err = client
        .start(StartRemoteControlPairingRequest { manual_code: false })
@@ -205,6 +181,7 @@ async fn start_remote_control_pairing_rejects_expired_server_token() {
        "environment-id".to_string(),
        OffsetDateTime::from_unix_timestamp(0).expect("expired timestamp should parse"),
        /*auth_change_revision*/ 0,
+        /*generation*/ 0,
    );

    let err = client
@@ -263,19 +240,7 @@ async fn start_remote_control_pairing_rejects_mismatched_enrollment() {
            .await
            .expect("response should write");
    });
-    let client = RemoteControlPairingClient::new(
-        &RemoteControlTarget {
-            websocket_url: "ws://unused".to_string(),
-            enroll_url: "http://unused".to_string(),
-            refresh_url: "http://unused".to_string(),
-            pair_url,
-        },
-        "remote-control-token".to_string(),
-        "server-id".to_string(),
-        "environment-id".to_string(),
-        OffsetDateTime::from_unix_timestamp(33_336_362_096).expect("future timestamp should parse"),
-        /*auth_change_revision*/ 0,
-    );
+    let client = pairing_client(pair_url);

    let err = client
        .start(StartRemoteControlPairingRequest { manual_code: false })
@@ -289,3 +254,20 @@ async fn start_remote_control_pairing_rejects_mismatched_enrollment() {
        "remote control pairing returned mismatched enrollment: expected server_id=server-id, environment_id=environment-id; got server_id=other-server-id, environment_id=other-environment-id"
    );
 }
+
+fn pairing_client(pair_url: String) -> RemoteControlPairingClient {
+    RemoteControlPairingClient::new(
+        &RemoteControlTarget {
+            websocket_url: "ws://unused".to_string(),
+            enroll_url: "http://unused".to_string(),
+            refresh_url: "http://unused".to_string(),
+            pair_url,
+        },
+        "remote-control-token".to_string(),
+        "server-id".to_string(),
+        "environment-id".to_string(),
+        OffsetDateTime::from_unix_timestamp(33_336_362_096).expect("future timestamp should parse"),
+        /*auth_change_revision*/ 0,
+        /*generation*/ 0,
+    )
+}
@@ -65,6 +65,7 @@ const TEST_REMOTE_CONTROL_URL: &str = "http://127.0.0.1:1/backend-api/wham/remot
 const TEST_REMOTE_CONTROL_SERVER_TOKEN: &str = "Remote Control Token";
 const TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN: &str = "Refreshed Remote Control Token";
 const TEST_REMOTE_CONTROL_SERVER_TOKEN_EXPIRES_AT: &str = "2999-01-01T00:00:00Z";
+const TEST_HTTP_ACCEPT_TIMEOUT: Duration = Duration::from_secs(10);

 fn remote_control_auth_manager() -> Arc<AuthManager> {
    auth_manager_from_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
@@ -151,12 +152,16 @@ fn remote_control_handle_with_pairing_client(
        "env_test".to_string(),
        OffsetDateTime::from_unix_timestamp(33_336_362_096).expect("future timestamp should parse"),
        /*auth_change_revision*/ 0,
+        /*generation*/ 0,
    ))));
    RemoteControlHandle {
        enabled_tx: Arc::new(enabled_tx),
        status_tx: Arc::new(status_tx),
        state_db_available: true,
-        pairing_client,
+        pairing: PairingClientState {
+            client: pairing_client,
+            generation: Arc::new(std::sync::atomic::AtomicU64::new(0)),
+        },
        auth_change_rx: Arc::new(StdMutex::new(auth_change_rx)),
    }
 }
@@ -928,221 +933,8 @@ async fn remote_control_handle_enable_disable_stops_and_restarts_connections() {
    let _ = remote_task.await;
 }

-#[tokio::test]
-async fn remote_control_handle_disable_clears_stale_pairing_client() {
-    let remote_handle = remote_control_handle_with_pairing_client(
-        TEST_REMOTE_CONTROL_URL,
-        watch::channel(/*init*/ 0u64).1,
-    );
-
-    assert_eq!(
-        remote_handle.disable(),
-        RemoteControlStatusChangedNotification {
-            status: RemoteControlConnectionStatus::Disabled,
-            server_name: test_server_name(),
-            installation_id: TEST_INSTALLATION_ID.to_string(),
-            environment_id: None,
-        }
-    );
-    remote_handle.enable().expect("enable should succeed");
-    assert_eq!(
-        remote_handle
-            .start_pairing(RemoteControlPairingStartParams { manual_code: false })
-            .await
-            .expect_err("re-enabled remote control should wait for refreshed pairing auth")
-            .to_string(),
-        "remote control pairing is unavailable until enrollment completes"
-    );
-}
-
-#[tokio::test]
-async fn remote_control_handle_rejects_pairing_client_after_auth_change() {
-    let (auth_change_tx, auth_change_rx) = watch::channel(/*init*/ 0u64);
-    let remote_handle =
-        remote_control_handle_with_pairing_client(TEST_REMOTE_CONTROL_URL, auth_change_rx);
-    auth_change_tx.send_modify(|revision| *revision += 1);
-
-    assert_eq!(
-        remote_handle
-            .start_pairing(RemoteControlPairingStartParams::default())
-            .await
-            .expect_err("pairing should wait for current-account enrollment")
-            .to_string(),
-        "remote control pairing is unavailable until enrollment completes"
-    );
-}
-
-#[tokio::test]
-async fn remote_control_handle_discards_pairing_response_after_auth_change() {
-    let listener = TcpListener::bind("127.0.0.1:0")
-        .await
-        .expect("listener should bind");
-    let remote_control_url = remote_control_url_for_listener(&listener);
-    let codex_home = TempDir::new().expect("temp dir should create");
-    save_auth(
-        codex_home.path(),
-        &remote_control_auth_dot_json(Some("account_id")),
-        AuthCredentialsStoreMode::File,
-    )
-    .expect("initial auth should save");
-    let auth_manager = AuthManager::shared(
-        codex_home.path().to_path_buf(),
-        /*enable_codex_api_key_env*/ false,
-        AuthCredentialsStoreMode::File,
-        /*chatgpt_base_url*/ None,
-    )
-    .await;
-    let remote_handle = remote_control_handle_with_pairing_client(
-        &remote_control_url,
-        auth_manager.auth_change_receiver(),
-    );
-    let pairing_task = tokio::spawn({
-        let remote_handle = remote_handle.clone();
-        async move {
-            remote_handle
-                .start_pairing(RemoteControlPairingStartParams::default())
-                .await
-        }
-    });
-
-    let pairing_request = accept_http_request(&listener).await;
-    assert_eq!(
-        pairing_request.request_line,
-        "POST /backend-api/wham/remote/control/server/pair HTTP/1.1"
-    );
-    assert_eq!(
-        pairing_request.headers.get("authorization"),
-        Some(&format!("Bearer {TEST_REMOTE_CONTROL_SERVER_TOKEN}"))
-    );
-    assert_eq!(
-        serde_json::from_str::<serde_json::Value>(&pairing_request.body)
-            .expect("pairing request body should deserialize"),
-        json!({ "manual_code": false })
-    );
-
-    save_auth(
-        codex_home.path(),
-        &remote_control_auth_dot_json(Some("next_account_id")),
-        AuthCredentialsStoreMode::File,
-    )
-    .expect("next auth should save");
-    auth_manager.reload().await;
-    respond_with_json(
-        pairing_request.stream,
-        json!({
-            "pairing_code": "stale-pairing-code",
-            "manual_pairing_code": "ABCD-EFGH",
-            "server_id": "srv_e_test",
-            "environment_id": "env_test",
-            "expires_at": "3026-05-22T12:34:56Z",
-        }),
-    )
-    .await;
-
-    assert_eq!(
-        pairing_task
-            .await
-            .expect("pairing task should join")
-            .expect_err("stale pairing response should be discarded")
-            .to_string(),
-        "remote control pairing is unavailable until enrollment completes"
-    );
-}
-
-#[tokio::test]
-async fn remote_control_handle_clears_pairing_client_after_auth_change() {
-    let listener = TcpListener::bind("127.0.0.1:0")
-        .await
-        .expect("listener should bind");
-    let remote_control_url = remote_control_url_for_listener(&listener);
-    let codex_home = TempDir::new().expect("temp dir should create");
-    save_auth(
-        codex_home.path(),
-        &remote_control_auth_dot_json(Some("account_id")),
-        AuthCredentialsStoreMode::File,
-    )
-    .expect("initial auth should save");
-    let auth_manager = AuthManager::shared(
-        codex_home.path().to_path_buf(),
-        /*enable_codex_api_key_env*/ false,
-        AuthCredentialsStoreMode::File,
-        /*chatgpt_base_url*/ None,
-    )
-    .await;
-    let (transport_event_tx, _transport_event_rx) =
-        mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
-    let shutdown_token = CancellationToken::new();
-    let (remote_task, remote_handle) = start_remote_control(
-        RemoteControlStartConfig {
-            remote_control_url,
-            installation_id: TEST_INSTALLATION_ID.to_string(),
-        },
-        Some(remote_control_state_runtime(&codex_home).await),
-        auth_manager.clone(),
-        transport_event_tx,
-        shutdown_token.clone(),
-        /*app_server_client_name_rx*/ None,
-        /*initial_enabled*/ true,
-    )
-    .await
-    .expect("remote control should start");
-
-    let enroll_request = accept_http_request(&listener).await;
-    respond_with_json(
-        enroll_request.stream,
-        remote_control_server_token_response(
-            "srv_e_initial",
-            "env_initial",
-            TEST_REMOTE_CONTROL_SERVER_TOKEN,
-        ),
-    )
-    .await;
-    let mut first_websocket = accept_remote_control_connection(&listener).await;
-
-    save_auth(
-        codex_home.path(),
-        &remote_control_auth_dot_json(Some("next_account_id")),
-        AuthCredentialsStoreMode::File,
-    )
-    .expect("next auth should save");
-    auth_manager.reload().await;
-    expect_remote_control_connection_closed(
-        &mut first_websocket,
-        "auth change should close the stale websocket",
-    )
-    .await;
-    assert_eq!(
-        remote_handle
-            .start_pairing(RemoteControlPairingStartParams::default())
-            .await
-            .expect_err("pairing should wait for current-account enrollment")
-            .to_string(),
-        "remote control pairing is unavailable until enrollment completes"
-    );
-
-    let enroll_request = accept_http_request(&listener).await;
-    assert_eq!(
-        enroll_request.request_line,
-        "POST /backend-api/wham/remote/control/server/enroll HTTP/1.1"
-    );
-    respond_with_json(
-        enroll_request.stream,
-        remote_control_server_token_response(
-            "srv_e_next",
-            "env_next",
-            TEST_REFRESHED_REMOTE_CONTROL_SERVER_TOKEN,
-        ),
-    )
-    .await;
-    let mut second_websocket = accept_remote_control_connection(&listener).await;
-    second_websocket
-        .close(None)
-        .await
-        .expect("second websocket should close");
-
-    shutdown_token.cancel();
-    let _ = remote_task.await;
-}
+#[path = "pairing_integration_tests.rs"]
+mod pairing_integration_tests;

 #[tokio::test]
 async fn remote_control_transport_clears_outgoing_buffer_when_backend_acks() {
@@ -2231,7 +2023,7 @@ async fn expect_remote_control_connection_closed(
 }

 async fn accept_http_request(listener: &TcpListener) -> CapturedHttpRequest {
-    let (stream, _) = timeout(Duration::from_secs(5), listener.accept())
+    let (stream, _) = timeout(TEST_HTTP_ACCEPT_TIMEOUT, listener.accept())
        .await
        .expect("HTTP request should arrive in time")
        .expect("listener accept should succeed");
@@ -1,3 +1,4 @@
+use super::PairingClientState;
 use crate::transport::TransportEvent;
 use crate::transport::remote_control::client_tracker::ClientTracker;
 use crate::transport::remote_control::client_tracker::REMOTE_CONTROL_IDLE_SWEEP_INTERVAL;
@@ -41,7 +42,7 @@ use std::collections::VecDeque;
 use std::io;
 use std::io::ErrorKind;
 use std::sync::Arc;
-use std::sync::Mutex as StdMutex;
+use std::sync::atomic::Ordering;
 use tokio::net::TcpStream;
 use tokio::sync::Mutex;
 use tokio::sync::mpsc;
@@ -254,7 +255,7 @@ pub(crate) struct RemoteControlWebsocket {
    enrollment: Option<RemoteControlEnrollment>,
    auth_recovery: UnauthorizedRecovery,
    auth_change_rx: watch::Receiver<u64>,
-    pairing_client: Arc<StdMutex<Option<RemoteControlPairingClient>>>,
+    pairing: PairingClientState,
    client_tracker: Arc<Mutex<ClientTracker>>,
    state: Arc<Mutex<WebsocketState>>,
    server_event_rx: Arc<Mutex<mpsc::Receiver<super::QueuedServerEnvelope>>>,
@@ -294,7 +295,7 @@ enum ConnectionEndReason {
 pub(super) struct RemoteControlChannels {
    pub(super) transport_event_tx: mpsc::Sender<TransportEvent>,
    pub(super) status_publisher: RemoteControlStatusPublisher,
-    pub(super) pairing_client: Arc<StdMutex<Option<RemoteControlPairingClient>>>,
+    pub(super) pairing: PairingClientState,
 }

 #[derive(Clone)]
@@ -411,7 +412,7 @@ impl RemoteControlWebsocket {
            enrollment: None,
            auth_recovery,
            auth_change_rx,
-            pairing_client: channels.pairing_client,
+            pairing: channels.pairing,
            client_tracker: Arc::new(Mutex::new(client_tracker)),
            state: Arc::new(Mutex::new(WebsocketState {
                outbound_buffer,
@@ -619,13 +620,14 @@ impl RemoteControlWebsocket {
                    &mut self.enrollment,
                    connect_options,
                    &self.status_publisher,
-                    &self.pairing_client,
+                    &self.pairing,
                ) => connect_result,
            };

            match connect_result {
                Ok((websocket_connection, response)) => {
                    if !*self.enabled_rx.borrow() {
+                        clear_pairing_client(&self.pairing);
                        return ConnectOutcome::Disabled;
                    }
                    self.reconnect_attempt = 0;
@@ -750,7 +752,7 @@ impl RemoteControlWebsocket {
            }
            _ = join_set.join_next() => ConnectionEndReason::ConnectionWorkerStopped,
        };
-        clear_pairing_client(&self.pairing_client);
+        clear_pairing_client(&self.pairing);
        shutdown_token.cancel();

        Self::join_connection_workers(&mut join_set, REMOTE_CONTROL_CONNECTION_SHUTDOWN_TIMEOUT)
@@ -1247,7 +1249,7 @@ pub(super) async fn connect_remote_control_websocket(
    enrollment: &mut Option<RemoteControlEnrollment>,
    connect_options: RemoteControlConnectOptions<'_>,
    status_publisher: &RemoteControlStatusPublisher,
-    pairing_client: &Arc<StdMutex<Option<RemoteControlPairingClient>>>,
+    pairing: &PairingClientState,
 ) -> io::Result<(
    WebSocketStream<MaybeTlsStream<TcpStream>>,
    tungstenite::http::Response<()>,
@@ -1256,33 +1258,28 @@ pub(super) async fn connect_remote_control_websocket(

    let Some(state_db) = state_db else {
        *enrollment = None;
-        clear_pairing_client(pairing_client);
+        clear_pairing_client(pairing);
        return Err(io::Error::new(
            ErrorKind::NotFound,
            "remote control requires sqlite state db",
        ));
    };

-    let (auth, auth_change_revision) = loop {
-        let auth_change_revision = *auth_context.auth_change_rx.borrow_and_update();
-        let auth = match load_remote_control_auth(auth_context.auth_manager).await {
-            Ok(auth) => auth,
-            Err(err) => {
-                if err.kind() == ErrorKind::PermissionDenied {
-                    *enrollment = None;
-                    status_publisher.publish_environment_id(/*environment_id*/ None);
-                    clear_pairing_client(pairing_client);
-                }
-                return Err(err);
+    let auth = match load_remote_control_auth(auth_context.auth_manager).await {
+        Ok(auth) => auth,
+        Err(err) => {
+            if err.kind() == ErrorKind::PermissionDenied {
+                *enrollment = None;
+                status_publisher.publish_environment_id(/*environment_id*/ None);
+                clear_pairing_client(pairing);
            }
-        };
-        if *auth_context.auth_change_rx.borrow() == auth_change_revision {
-            break (auth, auth_change_revision);
+            return Err(err);
        }
-        info!(
-            "retrying app-server remote control websocket auth load after auth changed while loading"
-        );
    };
+    // Loading auth may reload or proactively refresh through the same watch
+    // receiver. Treat the auth used for this connection as the current
+    // revision before publishing pairing auth for it.
+    let auth_change_revision = *auth_context.auth_change_rx.borrow_and_update();
    let enrollment_account_id = enrollment.as_ref().map(|enrollment| &enrollment.account_id);
    if enrollment_account_id.is_some_and(|account_id| account_id != &auth.account_id) {
        info!(
@@ -1295,7 +1292,7 @@ pub(super) async fn connect_remote_control_websocket(
        );
        *enrollment = None;
        status_publisher.publish_environment_id(/*environment_id*/ None);
-        clear_pairing_client(pairing_client);
+        clear_pairing_client(pairing);
    }

    if let Some(enrollment) = enrollment.as_ref() {
@@ -1373,7 +1370,7 @@ pub(super) async fn connect_remote_control_websocket(
                    connect_options.app_server_client_name,
                    enrollment,
                    status_publisher,
-                    pairing_client,
+                    pairing,
                )
                .await;
                enroll_remote_control_server_if_missing(
@@ -1431,7 +1428,7 @@ pub(super) async fn connect_remote_control_websocket(
    match websocket_connect_result {
        Ok((websocket_stream, response)) => {
            set_pairing_client(
-                pairing_client,
+                pairing,
                remote_control_target,
                enrollment_ref,
                auth_change_revision,
@@ -1439,7 +1436,7 @@ pub(super) async fn connect_remote_control_websocket(
            Ok((websocket_stream, response.map(|_| ())))
        }
        Err(err) => {
-            clear_pairing_client(pairing_client);
+            clear_pairing_client(pairing);
            match &err {
                tungstenite::Error::Http(response) if response.status().as_u16() == 404 => {
                    info!(
@@ -1456,7 +1453,7 @@ pub(super) async fn connect_remote_control_websocket(
                        connect_options.app_server_client_name,
                        enrollment,
                        status_publisher,
-                        pairing_client,
+                        pairing,
                    )
                    .await;
                }
@@ -1471,7 +1468,7 @@ pub(super) async fn connect_remote_control_websocket(
                            )
                        })?
                        .clear_server_token();
-                    clear_pairing_client(pairing_client);
+                    clear_pairing_client(pairing);
                    return Err(io::Error::other(format!(
                        "remote control websocket auth failed with HTTP {}; refreshing server token before reconnect",
                        response.status()
@@ -1496,7 +1493,7 @@ async fn clear_remote_control_enrollment(
    app_server_client_name: Option<&str>,
    enrollment: &mut Option<RemoteControlEnrollment>,
    status_publisher: &RemoteControlStatusPublisher,
-    pairing_client: &Arc<StdMutex<Option<RemoteControlPairingClient>>>,
+    pairing: &PairingClientState,
 ) {
    if let Err(clear_err) = update_persisted_remote_control_enrollment(
        Some(state_db),
@@ -1511,11 +1508,11 @@ async fn clear_remote_control_enrollment(
    }
    *enrollment = None;
    status_publisher.publish_environment_id(/*environment_id*/ None);
-    clear_pairing_client(pairing_client);
+    clear_pairing_client(pairing);
 }

 fn set_pairing_client(
-    pairing_client: &Arc<StdMutex<Option<RemoteControlPairingClient>>>,
+    pairing: &PairingClientState,
    remote_control_target: &RemoteControlTarget,
    enrollment: &RemoteControlEnrollment,
    auth_change_revision: u64,
@@ -1532,7 +1529,9 @@ fn set_pairing_client(
            "remote control pairing is unavailable until enrollment completes",
        )
    })?;
-    *pairing_client
+    let generation = pairing.generation.load(Ordering::Relaxed);
+    *pairing
+        .client
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner) =
        Some(RemoteControlPairingClient::new(
@@ -1542,6 +1541,7 @@ fn set_pairing_client(
            enrollment.environment_id.clone(),
            expires_at,
            auth_change_revision,
+            generation,
        ));
    Ok(())
 }
@@ -1734,8 +1734,8 @@ mod tests {
        }
    }

-    fn test_pairing_client() -> Arc<StdMutex<Option<RemoteControlPairingClient>>> {
-        Arc::new(StdMutex::new(None))
+    fn test_pairing() -> PairingClientState {
+        PairingClientState::new()
    }

    #[test]
@@ -1891,7 +1891,7 @@ mod tests {
        let mut enrollment = Some(remote_control_enrollment(Some(
            TEST_REMOTE_CONTROL_SERVER_TOKEN,
        )));
-        let pairing_client = test_pairing_client();
+        let pairing = test_pairing();
        let (status_publisher, status_rx) = remote_control_status_channel();

        let err = match connect_remote_control_websocket(
@@ -1910,7 +1910,7 @@ mod tests {
                app_server_client_name: None,
            },
            &status_publisher,
-            &pairing_client,
+            &pairing,
        )
        .await
        {
@@ -1921,7 +1921,8 @@ mod tests {
        server_task.await.expect("server task should succeed");
        assert_eq!(err.to_string(), expected_error);
        assert!(
-            pairing_client
+            pairing
+                .client
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .is_none()
@@ -1953,7 +1954,7 @@ mod tests {
        let mut enrollment = Some(remote_control_enrollment(Some(
            TEST_REMOTE_CONTROL_SERVER_TOKEN,
        )));
-        let pairing_client = test_pairing_client();
+        let pairing = test_pairing();
        let (status_publisher, status_rx) = remote_control_status_channel();

        let server_task = tokio::spawn(async move {
@@ -1981,7 +1982,7 @@ mod tests {
                app_server_client_name: None,
            },
            &status_publisher,
-            &pairing_client,
+            &pairing,
        )
        .await
        .expect_err("unauthorized response should fail the websocket connect");
@@ -2007,7 +2008,8 @@ mod tests {
            ))
        );
        assert_eq!(
-            pairing_client
+            pairing
+                .client
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .is_none(),
@@ -2074,7 +2076,7 @@ mod tests {
                app_server_client_name: None,
            },
            &status_publisher,
-            &test_pairing_client(),
+            &test_pairing(),
        )
        .await
        .expect_err("unauthorized enrollment should fail the websocket connect");
@@ -2169,7 +2171,7 @@ mod tests {
                app_server_client_name: None,
            },
            &status_publisher,
-            &test_pairing_client(),
+            &test_pairing(),
        )
        .await
        .expect_err("unauthorized refresh should fail the websocket connect");
@@ -2235,7 +2237,7 @@ mod tests {
                app_server_client_name: None,
            },
            &status_publisher,
-            &test_pairing_client(),
+            &test_pairing(),
        )
        .await
        .expect_err("missing sqlite state db should fail remote control");
@@ -2286,7 +2288,7 @@ mod tests {
                app_server_client_name: None,
            },
            &status_publisher,
-            &test_pairing_client(),
+            &test_pairing(),
        )
        .await
        .expect_err("missing auth should fail remote control");
@@ -2338,7 +2340,7 @@ mod tests {
                    RemoteControlChannels {
                        transport_event_tx,
                        status_publisher,
-                        pairing_client: test_pairing_client(),
+                        pairing: test_pairing(),
                    },
                    shutdown_token,
                    enabled_rx,