Fetch workspace connectors concurrently (#33421)

## Why

Workspace accounts fetched the paginated connector directory before starting the independent workspace connector request, adding the latency of both request chains.

## What changed

Start the workspace connector request alongside the public directory lookup. Continue to ignore workspace lookup failures and filter hidden workspace apps before merging the results.

## Testing

Add a regression test that blocks the directory response until the workspace request starts, verifying that both lookups overlap and their connectors are returned.

GitOrigin-RevId: fa29023a4154b555a31756054109e706e61c32c0

codex-rs/connectors/Cargo.toml

@@ -20,12 +20,12 @@ serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 sha1 = { workspace = true }
 tempfile = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
 tracing = { workspace = true }
 urlencoding = { workspace = true }
 
 [dev-dependencies]
 pretty_assertions = { workspace = true }
-tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
 
 [lib]
 doctest = false

codex-rs/connectors/src/lib.rs

@@ -172,10 +172,28 @@ where
         return Ok(cached_connectors);
     }
 
-    let mut apps = list_directory_connectors(&mut fetch_page).await?;
-    if is_workspace_account {
-        apps.extend(list_workspace_connectors(&mut fetch_page).await?);
-    }
+    let apps = if is_workspace_account {
+        // The workspace directory is independent from the paginated public directory.
+        // Start both before awaiting either so workspace accounts do not pay for the
+        // two request chains back-to-back.
+        let workspace_connectors =
+            fetch_page("/connectors/directory/list_workspace?external_logos=true".to_string());
+        let directory_connectors = list_directory_connectors(&mut fetch_page);
+        let (directory_connectors, workspace_connectors) =
+            tokio::join!(directory_connectors, workspace_connectors);
+        let mut apps = directory_connectors?;
+        if let Ok(response) = workspace_connectors {
+            apps.extend(
+                response
+                    .apps
+                    .into_iter()
+                    .filter(|app| !is_hidden_directory_app(app)),
+            );
+        }
+        apps
+    } else {
+        list_directory_connectors(&mut fetch_page).await?
+    };
 
     let mut connectors = merge_directory_apps(apps)
         .into_iter()
@@ -260,23 +278,6 @@ where
     Ok(apps)
 }
 
-async fn list_workspace_connectors<F, Fut>(fetch_page: &mut F) -> anyhow::Result<Vec<DirectoryApp>>
-where
-    F: FnMut(String) -> Fut,
-    Fut: Future<Output = anyhow::Result<DirectoryListResponse>>,
-{
-    let response =
-        fetch_page("/connectors/directory/list_workspace?external_logos=true".to_string()).await;
-    match response {
-        Ok(response) => Ok(response
-            .apps
-            .into_iter()
-            .filter(|app| !is_hidden_directory_app(app))
-            .collect()),
-        Err(_) => Ok(Vec::new()),
-    }
-}
-
 fn merge_directory_apps(apps: Vec<DirectoryApp>) -> Vec<DirectoryApp> {
     let mut merged: HashMap<String, DirectoryApp> = HashMap::new();
     for app in apps {
@@ -518,7 +519,9 @@ mod tests {
     use std::sync::Mutex;
     use std::sync::atomic::AtomicUsize;
     use std::sync::atomic::Ordering;
+    use std::time::Duration;
     use tempfile::TempDir;
+    use tokio::sync::Notify;
 
     static CONNECTOR_DIRECTORY_CACHE_TEST_LOCK: LazyLock<tokio::sync::Mutex<()>> =
         LazyLock::new(|| tokio::sync::Mutex::new(()));
@@ -733,6 +736,60 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "test serializes access to the shared connector cache for its full duration"
+    )]
+    async fn list_all_connectors_overlaps_workspace_and_directory_requests() -> anyhow::Result<()> {
+        let _cache_guard = CONNECTOR_DIRECTORY_CACHE_TEST_LOCK.lock().await;
+
+        let codex_home = TempDir::new()?;
+        let cache_context = cache_context(&codex_home, "overlap");
+        let workspace_started = Arc::new(Notify::new());
+
+        // The public directory response waits until the workspace request is polled.
+        // Without overlap this future cannot complete; the timeout only bounds a
+        // regression instead of supplying the ordering.
+        let connectors = tokio::time::timeout(
+            Duration::from_secs(1),
+            list_all_connectors_with_options(
+                cache_context,
+                /*is_workspace_account*/ true,
+                /*force_refetch*/ true,
+                move |path| {
+                    let workspace_started = Arc::clone(&workspace_started);
+                    async move {
+                        if path.starts_with("/connectors/directory/list_workspace") {
+                            workspace_started.notify_one();
+                            Ok(DirectoryListResponse {
+                                apps: vec![app("workspace", "Workspace")],
+                                next_token: None,
+                            })
+                        } else {
+                            workspace_started.notified().await;
+                            Ok(DirectoryListResponse {
+                                apps: vec![app("directory", "Directory")],
+                                next_token: None,
+                            })
+                        }
+                    }
+                },
+            ),
+        )
+        .await
+        .expect("workspace request should start while directory request is pending")?;
+
+        assert_eq!(
+            connectors
+                .into_iter()
+                .map(|connector| connector.id)
+                .collect::<Vec<_>>(),
+            vec!["directory".to_string(), "workspace".to_string()]
+        );
+        Ok(())
+    }
+
     #[tokio::test]
     #[expect(
         clippy::await_holding_invalid_type,