Allow config reads to join active app-server read batches (#39036)

## Why

Config reads arriving while a long-running shared read was active waited for that
request to finish, even though they could safely run concurrently.

## What changed

- Admit newly queued shared reads into the active batch until an exclusive write
  is queued, preserving the write as a fairness barrier.
- Treat `hooks/list` and `configRequirements/read` as shared config reads.
- Cover concurrent reads, read/write ordering, and config reads completing
  alongside a `skills/list` request.

GitOrigin-RevId: 8194301c2ba60be0fc71b8915923af9accc745dc

codex-rs/app-server-protocol/src/protocol/common.rs

@@ -812,7 +812,7 @@ client_request_definitions! {
     },
     HooksList => "hooks/list" {
         params: v2::HooksListParams,
-        serialization: global("config"),
+        serialization: global_shared_read("config"),
         response: v2::HooksListResponse,
     },
     MarketplaceAdd => "marketplace/add" {
@@ -1310,7 +1310,7 @@ client_request_definitions! {
 
     ConfigRequirementsRead => "configRequirements/read" {
         params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>,
-        serialization: global("config"),
+        serialization: global_shared_read("config"),
         response: v2::ConfigRequirementsReadResponse,
     },
 
@@ -2245,6 +2245,15 @@ mod tests {
             Some(ClientRequestSerializationScope::GlobalSharedRead("config"))
         );
 
+        let hooks_list = ClientRequest::HooksList {
+            request_id: request_id(),
+            params: v2::HooksListParams { cwds: Vec::new() },
+        };
+        assert_eq!(
+            hooks_list.serialization_scope(),
+            Some(ClientRequestSerializationScope::GlobalSharedRead("config"))
+        );
+
         let skills_extra_roots_set = ClientRequest::SkillsExtraRootsSet {
             request_id: request_id(),
             params: v2::SkillsExtraRootsSetParams {
@@ -2340,6 +2349,15 @@ mod tests {
             Some(ClientRequestSerializationScope::GlobalSharedRead("config"))
         );
 
+        let config_requirements_read = ClientRequest::ConfigRequirementsRead {
+            request_id: request_id(),
+            params: None,
+        };
+        assert_eq!(
+            config_requirements_read.serialization_scope(),
+            Some(ClientRequestSerializationScope::GlobalSharedRead("config"))
+        );
+
         let account_read = ClientRequest::GetAccount {
             request_id: request_id(),
             params: v2::GetAccountParams {

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

@@ -8,8 +8,10 @@ use std::sync::Arc;
 use codex_app_server_protocol::ClientRequestSerializationScope;
 use codex_diagnostics::Gauge;
 use codex_diagnostics::GaugeGuard;
-use futures::future::join_all;
+use futures::stream::FuturesUnordered;
+use futures::stream::StreamExt;
 use tokio::sync::Mutex;
+use tokio::sync::Notify;
 use tracing::Instrument;
 
 use crate::connection_rpc_gate::ConnectionRpcGate;
@@ -145,9 +147,14 @@ struct QueuedSerializedRequest {
     _diagnostics_guard: GaugeGuard,
 }
 
+struct RequestSerializationQueue {
+    requests: VecDeque<QueuedSerializedRequest>,
+    changed: Arc<Notify>,
+}
+
 #[derive(Clone, Default)]
 pub(crate) struct RequestSerializationQueues {
-    inner: Arc<Mutex<HashMap<RequestSerializationQueueKey, VecDeque<QueuedSerializedRequest>>>>,
+    inner: Arc<Mutex<HashMap<RequestSerializationQueueKey, RequestSerializationQueue>>>,
 }
 
 impl RequestSerializationQueues {
@@ -181,12 +188,17 @@ impl RequestSerializationQueues {
             let mut queues = self.inner.lock().await;
             match queues.get_mut(&key) {
                 Some(queue) => {
-                    queue.push_back(request);
+                    queue.requests.push_back(request);
+                    queue.changed.notify_one();
                     false
                 }
                 None => {
-                    let mut queue = VecDeque::new();
-                    queue.push_back(request);
+                    let mut requests = VecDeque::new();
+                    requests.push_back(request);
+                    let queue = RequestSerializationQueue {
+                        requests,
+                        changed: Arc::new(Notify::new()),
+                    };
                     queues.insert(key.clone(), queue);
                     true
                 }
@@ -202,26 +214,26 @@ impl RequestSerializationQueues {
 
     async fn drain(self, key: RequestSerializationQueueKey) {
         loop {
-            let requests = {
+            let (requests, changed) = {
                 let mut queues = self.inner.lock().await;
                 let Some(queue) = queues.get_mut(&key) else {
                     return;
                 };
-                match queue.pop_front() {
+                match queue.requests.pop_front() {
                     Some(request) => {
                         let access = request.access;
                         let mut requests = vec![request];
                         if access == RequestSerializationAccess::SharedRead {
-                            while queue.front().is_some_and(|request| {
+                            while queue.requests.front().is_some_and(|request| {
                                 request.access == RequestSerializationAccess::SharedRead
                             }) {
-                                let Some(request) = queue.pop_front() else {
+                                let Some(request) = queue.requests.pop_front() else {
                                     break;
                                 };
                                 requests.push(request);
                             }
                         }
-                        requests
+                        (requests, Arc::clone(&queue.changed))
                     }
                     None => {
                         queues.remove(&key);
@@ -230,7 +242,49 @@ impl RequestSerializationQueues {
                 }
             };
 
-            join_all(requests.into_iter().map(|request| request.request.run())).await;
+            if requests[0].access == RequestSerializationAccess::Exclusive {
+                for request in requests {
+                    request.request.run().await;
+                }
+                continue;
+            }
+
+            let mut running_reads = requests
+                .into_iter()
+                .map(|request| request.request.run())
+                .collect::<FuturesUnordered<_>>();
+
+            loop {
+                tokio::select! {
+                    Some(()) = running_reads.next() => {
+                        if running_reads.is_empty() {
+                            break;
+                        }
+                    }
+                    () = changed.notified() => {
+                        let requests = {
+                            let mut queues = self.inner.lock().await;
+                            let Some(queue) = queues.get_mut(&key) else {
+                                return;
+                            };
+                            let mut requests = Vec::new();
+                            while queue.requests.front().is_some_and(|request| {
+                                request.access == RequestSerializationAccess::SharedRead
+                            }) {
+                                let Some(request) = queue.requests.pop_front() else {
+                                    break;
+                                };
+                                requests.push(request);
+                            }
+                            requests
+                        };
+
+                        for request in requests {
+                            running_reads.push(request.request.run());
+                        }
+                    }
+                }
+            }
         }
     }
 }
@@ -238,6 +292,10 @@ impl RequestSerializationQueues {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use codex_app_server_protocol::ClientRequest;
+    use codex_app_server_protocol::ConfigBatchWriteParams;
+    use codex_app_server_protocol::HooksListParams;
+    use codex_app_server_protocol::RequestId;
     use pretty_assertions::assert_eq;
     use std::sync::Arc;
     use tokio::sync::broadcast;
@@ -539,9 +597,129 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn exclusive_write_waits_for_running_shared_reads() {
+    async fn later_shared_read_joins_running_shared_read() {
         let queues = RequestSerializationQueues::default();
         let key = RequestSerializationQueueKey::Global("test");
+        let (first_started_tx, first_started_rx) = oneshot::channel::<()>();
+        let (first_release_tx, first_release_rx) = oneshot::channel::<()>();
+        let (later_started_tx, later_started_rx) = oneshot::channel::<()>();
+
+        queues
+            .enqueue(
+                key.clone(),
+                RequestSerializationAccess::SharedRead,
+                QueuedInitializedRequest::new(gate(), async move {
+                    first_started_tx.send(()).expect("receiver should be open");
+                    let _ = first_release_rx.await;
+                }),
+            )
+            .await;
+        timeout(queue_drain_timeout(), first_started_rx)
+            .await
+            .expect("first read should start")
+            .expect("sender should be open");
+
+        queues
+            .enqueue(
+                key,
+                RequestSerializationAccess::SharedRead,
+                QueuedInitializedRequest::new(gate(), async move {
+                    later_started_tx.send(()).expect("receiver should be open");
+                }),
+            )
+            .await;
+
+        timeout(queue_drain_timeout(), later_started_rx)
+            .await
+            .expect("later read should join the running read")
+            .expect("sender should be open");
+        first_release_tx
+            .send(())
+            .expect("first read should still be waiting");
+    }
+
+    #[tokio::test]
+    async fn later_shared_read_waits_behind_writer_queued_during_running_read() {
+        let queues = RequestSerializationQueues::default();
+        let key = RequestSerializationQueueKey::Global("test");
+        let (first_read_started_tx, first_read_started_rx) = oneshot::channel::<()>();
+        let (first_read_release_tx, first_read_release_rx) = oneshot::channel::<()>();
+        let (write_started_tx, write_started_rx) = oneshot::channel::<()>();
+        let (write_release_tx, write_release_rx) = oneshot::channel::<()>();
+        let (later_read_started_tx, later_read_started_rx) = oneshot::channel::<()>();
+
+        queues
+            .enqueue(
+                key.clone(),
+                RequestSerializationAccess::SharedRead,
+                QueuedInitializedRequest::new(gate(), async move {
+                    first_read_started_tx
+                        .send(())
+                        .expect("receiver should be open");
+                    let _ = first_read_release_rx.await;
+                }),
+            )
+            .await;
+        timeout(queue_drain_timeout(), first_read_started_rx)
+            .await
+            .expect("first read should start")
+            .expect("sender should be open");
+
+        queues
+            .enqueue(
+                key.clone(),
+                RequestSerializationAccess::Exclusive,
+                QueuedInitializedRequest::new(gate(), async move {
+                    write_started_tx.send(()).expect("receiver should be open");
+                    let _ = write_release_rx.await;
+                }),
+            )
+            .await;
+        queues
+            .enqueue(
+                key,
+                RequestSerializationAccess::SharedRead,
+                QueuedInitializedRequest::new(gate(), async move {
+                    later_read_started_tx
+                        .send(())
+                        .expect("receiver should be open");
+                }),
+            )
+            .await;
+
+        let mut write_started_rx = Box::pin(write_started_rx);
+        timeout(shutdown_wait_timeout(), &mut write_started_rx)
+            .await
+            .expect_err("write should wait for the running read");
+        let mut later_read_started_rx = Box::pin(later_read_started_rx);
+        timeout(shutdown_wait_timeout(), &mut later_read_started_rx)
+            .await
+            .expect_err("later read should wait behind the queued write");
+
+        first_read_release_tx
+            .send(())
+            .expect("first read should still be waiting");
+        timeout(queue_drain_timeout(), &mut write_started_rx)
+            .await
+            .expect("write should start after the running read finishes")
+            .expect("sender should be open");
+        timeout(shutdown_wait_timeout(), &mut later_read_started_rx)
+            .await
+            .expect_err("later read should wait for the running write");
+
+        write_release_tx
+            .send(())
+            .expect("write should still be waiting");
+        timeout(queue_drain_timeout(), &mut later_read_started_rx)
+            .await
+            .expect("later read should start after the write finishes")
+            .expect("sender should be open");
+    }
+
+    #[tokio::test]
+    async fn startup_config_reads_run_concurrently_and_exclude_writes() {
+        let queues = RequestSerializationQueues::default();
+        let key = RequestSerializationQueueKey::Global("config");
         let (blocker_started_tx, blocker_started_rx) = oneshot::channel::<()>();
         let (blocker_release_tx, blocker_release_rx) = oneshot::channel::<()>();
         let (read_started_tx, mut read_started_rx) = mpsc::unbounded_channel();
@@ -565,13 +743,34 @@ mod tests {
             .expect("blocker should start")
             .expect("sender should be open");
 
-        for value in [FIRST_REQUEST_VALUE, SECOND_REQUEST_VALUE] {
+        for (value, request) in [
+            (
+                FIRST_REQUEST_VALUE,
+                ClientRequest::HooksList {
+                    request_id: RequestId::Integer(i64::from(FIRST_REQUEST_VALUE)),
+                    params: HooksListParams { cwds: Vec::new() },
+                },
+            ),
+            (
+                SECOND_REQUEST_VALUE,
+                ClientRequest::ConfigRequirementsRead {
+                    request_id: RequestId::Integer(i64::from(SECOND_REQUEST_VALUE)),
+                    params: None,
+                },
+            ),
+        ] {
             let read_started_tx = read_started_tx.clone();
             let mut read_release_rx = read_release_tx.subscribe();
+            let (request_key, access) = RequestSerializationQueueKey::from_scope(
+                ConnectionId(0),
+                request
+                    .serialization_scope()
+                    .expect("startup config reads should be serialized"),
+            );
             queues
                 .enqueue(
-                    key.clone(),
-                    RequestSerializationAccess::SharedRead,
+                    request_key,
+                    access,
                     QueuedInitializedRequest::new(gate(), async move {
                         read_started_tx
                             .send(value)
@@ -581,10 +780,26 @@ mod tests {
                 )
                 .await;
         }
+
+        let config_write = ClientRequest::ConfigBatchWrite {
+            request_id: RequestId::Integer(i64::from(THIRD_REQUEST_VALUE)),
+            params: ConfigBatchWriteParams {
+                edits: Vec::new(),
+                file_path: None,
+                expected_version: None,
+                reload_user_config: false,
+            },
+        };
+        let (write_key, write_access) = RequestSerializationQueueKey::from_scope(
+            ConnectionId(0),
+            config_write
+                .serialization_scope()
+                .expect("config writes should be serialized"),
+        );
         queues
             .enqueue(
-                key.clone(),
-                RequestSerializationAccess::Exclusive,
+                write_key,
+                write_access,
                 QueuedInitializedRequest::new(gate(), async move {
                     write_started_tx.send(()).expect("receiver should be open");
                 }),

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

@@ -9,10 +9,15 @@ use app_test_support::create_mock_responses_server_repeating_assistant;
 use app_test_support::write_chatgpt_auth;
 use codex_app_server_protocol::ConfigBatchWriteParams;
 use codex_app_server_protocol::ConfigEdit;
+use codex_app_server_protocol::ConfigReadParams;
+use codex_app_server_protocol::ConfigReadResponse;
+use codex_app_server_protocol::ConfigRequirementsReadResponse;
 use codex_app_server_protocol::ConfigWriteResponse;
 use codex_app_server_protocol::ExperimentalFeatureEnablementSetParams;
 use codex_app_server_protocol::ExperimentalFeatureEnablementSetResponse;
 use codex_app_server_protocol::MergeStrategy;
+use codex_app_server_protocol::PermissionProfileListParams;
+use codex_app_server_protocol::PermissionProfileListResponse;
 use codex_app_server_protocol::PluginListParams;
 use codex_app_server_protocol::PluginListResponse;
 use codex_app_server_protocol::SkillScope;
@@ -658,6 +663,61 @@ async fn skills_list_loads_remote_installed_plugin_skills_from_cache() -> Result
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn config_reads_complete_alongside_skills_list_request() -> Result<()> {
+    let codex_home = TempDir::new()?;
+    let cwd = TempDir::new()?;
+    let mut mcp = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .without_managed_config()
+        .build_initialized_with_timeout(DEFAULT_TIMEOUT)
+        .await?;
+
+    let skills_request_id = mcp
+        .send_skills_list_request(SkillsListParams {
+            cwds: vec![cwd.path().to_path_buf()],
+            force_reload: true,
+        })
+        .await?;
+
+    let config_request_id = mcp
+        .send_config_read_request(ConfigReadParams {
+            include_layers: false,
+            cwd: None,
+        })
+        .await?;
+    let requirements_request_id = mcp.send_config_requirements_read_request().await?;
+    let permission_profiles_request_id = mcp
+        .send_permission_profile_list_request(PermissionProfileListParams {
+            cursor: None,
+            limit: None,
+            cwd: None,
+        })
+        .await?;
+
+    let (config, requirements, permission_profiles) = timeout(Duration::from_secs(5), async {
+        let config: ConfigReadResponse = mcp.read_response(config_request_id).await?;
+        let requirements: ConfigRequirementsReadResponse =
+            mcp.read_response(requirements_request_id).await?;
+        let permission_profiles: PermissionProfileListResponse =
+            mcp.read_response(permission_profiles_request_id).await?;
+        anyhow::Ok((config, requirements, permission_profiles))
+    })
+    .await??;
+    assert!(config.layers.is_none());
+    assert_eq!(
+        requirements,
+        ConfigRequirementsReadResponse { requirements: None }
+    );
+    assert!(!permission_profiles.data.is_empty());
+
+    let SkillsListResponse { data } =
+        timeout(DEFAULT_TIMEOUT, mcp.read_response(skills_request_id)).await??;
+    assert_eq!(data.len(), 1);
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn skills_list_skips_cwd_roots_when_environment_disabled() -> Result<()> {
     let codex_home = TempDir::new()?;