Use installed callable apps for TUI mentions

codex-rs/tui/src/app.rs

@@ -212,6 +212,7 @@ mod app_server_events;
 pub(crate) mod app_server_requests;
 mod background_requests;
 mod config_persistence;
+mod connector_mentions;
 mod event_dispatch;
 mod history_pagination;
 mod history_ui;

codex-rs/tui/src/app/connector_mentions.rs

@@ -0,0 +1,155 @@
+//! Installed connector metadata used by the TUI `$` mention picker.
+
+use anyhow::Context;
+use anyhow::Result;
+use codex_app_server_client::AppServerRequestHandle;
+use codex_app_server_client::TypedRequestError;
+use codex_app_server_protocol::AppsInstalledParams;
+use codex_app_server_protocol::AppsInstalledResponse;
+use codex_app_server_protocol::AppsReadParams;
+use codex_app_server_protocol::AppsReadResponse;
+use codex_app_server_protocol::ClientRequest;
+use codex_app_server_protocol::RequestId;
+use codex_connectors::AppInfo;
+use std::collections::HashMap;
+use uuid::Uuid;
+
+use super::App;
+use super::background_requests::fetch_connectors_list;
+use crate::app_event::AppEvent;
+use crate::app_event::ConnectorsSnapshot;
+use crate::app_server_session::AppServerSession;
+use crate::chatwidget::ConnectorScopeGeneration;
+
+const APP_READ_BATCH_SIZE: usize = 100;
+
+impl App {
+    /// Fetch installed app mentions in the background and preserve their owning session scope.
+    pub(super) fn fetch_installed_connector_mentions(
+        &mut self,
+        app_server: &AppServerSession,
+        force_refresh: bool,
+        generation: ConnectorScopeGeneration,
+    ) {
+        let request_handle = app_server.request_handle();
+        let app_event_tx = self.app_event_tx.clone();
+        let thread_id = self.current_displayed_thread_id();
+        let cwd = self.chat_widget.config_ref().cwd.to_path_buf();
+        tokio::spawn(async move {
+            let result = fetch_installed_connector_mentions(
+                request_handle,
+                thread_id.map(|thread_id| thread_id.to_string()),
+                force_refresh,
+            )
+            .await
+            .map_err(|err| err.to_string());
+            app_event_tx.send(AppEvent::InstalledConnectorMentionsLoaded {
+                thread_id,
+                cwd,
+                generation,
+                result,
+            });
+        });
+    }
+}
+
+/// Load installed app identities and enrich callable apps with bounded metadata lookups.
+///
+/// Older app servers without `app/installed` fall back to their thread-scoped app directory.
+pub(super) async fn fetch_installed_connector_mentions(
+    request_handle: AppServerRequestHandle,
+    thread_id: Option<String>,
+    force_refresh: bool,
+) -> Result<ConnectorsSnapshot> {
+    let response: AppsInstalledResponse = match request_handle
+        .request_typed(ClientRequest::AppsInstalled {
+            request_id: RequestId::String(format!("apps-installed-{}", Uuid::new_v4())),
+            params: AppsInstalledParams {
+                thread_id: thread_id.clone(),
+                force_refresh,
+            },
+        })
+        .await
+    {
+        Ok(response) => response,
+        Err(TypedRequestError::Server { source, .. })
+            if source.code == -32601
+                || source.code == -32600
+                    && source.message.contains("app/installed")
+                    && (source.message.contains("unknown variant")
+                        || source.message.contains("unknown method")) =>
+        {
+            let mut snapshot = fetch_connectors_list(request_handle, force_refresh, thread_id)
+                .await
+                .map_err(|err| anyhow::anyhow!("{err:#}"))?;
+            snapshot.connectors.retain(|app| app.is_accessible);
+            return Ok(snapshot);
+        }
+        Err(err) => return Err(err).context("app/installed failed while loading app mentions"),
+    };
+    let callable_apps: Vec<_> = response.apps.iter().filter(|app| app.callable).collect();
+
+    let mut metadata_by_id = HashMap::new();
+    for apps in callable_apps.chunks(APP_READ_BATCH_SIZE) {
+        match request_handle
+            .request_typed::<AppsReadResponse>(ClientRequest::AppsRead {
+                request_id: RequestId::String(format!("apps-read-{}", Uuid::new_v4())),
+                params: AppsReadParams {
+                    app_ids: apps.iter().map(|app| app.id.clone()).collect(),
+                    thread_id: thread_id.clone(),
+                    include_tools: false,
+                },
+            })
+            .await
+        {
+            Ok(response) => {
+                metadata_by_id.extend(response.apps.into_iter().map(|app| (app.id.clone(), app)))
+            }
+            Err(err) => tracing::warn!(
+                error = %err,
+                "app/read failed while loading TUI connector mentions; using runtime metadata"
+            ),
+        }
+    }
+
+    let mut connectors = response
+        .apps
+        .into_iter()
+        .map(|app| {
+            let mut connector = AppInfo {
+                name: app.runtime_name.unwrap_or_else(|| app.id.clone()),
+                id: app.id,
+                description: None,
+                logo_url: None,
+                logo_url_dark: None,
+                icon_assets: None,
+                icon_dark_assets: None,
+                distribution_channel: None,
+                branding: None,
+                app_metadata: None,
+                labels: None,
+                install_url: None,
+                is_accessible: true,
+                is_enabled: app.callable,
+                plugin_display_names: Vec::new(),
+            };
+            if let Some(metadata) = metadata_by_id.remove(connector.id.as_str()) {
+                connector.name = metadata.name;
+                connector.description = metadata.description;
+                connector.logo_url = metadata.icon_url;
+                connector.logo_url_dark = metadata.icon_url_dark;
+                connector.distribution_channel = metadata.distribution_channel;
+                connector.install_url = metadata.install_url;
+                connector.plugin_display_names = metadata.plugin_display_names;
+            }
+            connector
+        })
+        .collect::<Vec<_>>();
+    connectors.sort_by(|left, right| (&left.name, &left.id).cmp(&(&right.name, &right.id)));
+
+    Ok(ConnectorsSnapshot { connectors })
+}
+
+#[cfg(test)]
+#[path = "connector_mentions_tests.rs"]
+mod tests;

codex-rs/tui/src/app/connector_mentions_tests.rs

@@ -0,0 +1,170 @@
+use super::*;
+use codex_app_server_protocol::AppsListParams;
+use codex_app_server_protocol::JSONRPCMessage;
+use futures::SinkExt;
+use futures::StreamExt;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use tokio::net::TcpListener;
+use tokio_tungstenite::accept_async;
+use tokio_tungstenite::tungstenite::Message;
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum MockResponses {
+    FirstBatchFails,
+    Unsupported,
+}
+
+async fn load_mock_mentions(responses: MockResponses) -> Result<(ConnectorsSnapshot, Vec<usize>)> {
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let websocket_url = format!("ws://{}", listener.local_addr()?);
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await?;
+        let mut websocket = accept_async(stream).await?;
+        let mut read_batches = Vec::new();
+
+        while let Some(frame) = websocket.next().await {
+            let Message::Text(text) = frame? else {
+                continue;
+            };
+            let JSONRPCMessage::Request(request) = serde_json::from_str(&text)? else {
+                continue;
+            };
+            let request_id = request.id.clone();
+            let response = match request.method.as_str() {
+                "initialize" => {
+                    json!({ "id": request_id, "result": { "userAgent": "codex-tui-test" } })
+                }
+                "app/installed" => {
+                    assert_eq!(
+                        request.params,
+                        Some(json!({ "threadId": "thread-1", "forceRefresh": true }))
+                    );
+                    if responses == MockResponses::Unsupported {
+                        let message = "Invalid request: unknown variant `app/installed`";
+                        json!({ "id": request_id, "error": { "code": -32600, "message": message } })
+                    } else {
+                        let mut apps = (0..103)
+                            .map(|index| json!({
+                                "id": format!("app-{index:03}"),
+                                "runtimeName": (index != 102).then(|| format!("Runtime {index:03}")),
+                                "enabled": true,
+                                "callable": true,
+                            }))
+                            .collect::<Vec<_>>();
+                        apps.extend([
+                            json!({ "id": "disabled", "enabled": false, "callable": false }),
+                            json!({ "id": "no-visible-tools", "enabled": true, "callable": false }),
+                        ]);
+                        json!({ "id": request_id, "result": { "apps": apps } })
+                    }
+                }
+                "app/list" => {
+                    let params = serde_json::from_value::<AppsListParams>(
+                        request.params.expect("app/list params"),
+                    )?;
+                    assert_eq!(
+                        (params.thread_id.as_deref(), params.force_refetch),
+                        (Some("thread-1"), true)
+                    );
+                    json!({
+                        "id": request_id,
+                        "result": {
+                            "data": [
+                                { "id": "installed", "name": "Installed", "isAccessible": true },
+                                { "id": "disabled", "name": "Disabled", "isAccessible": true, "isEnabled": false },
+                                { "id": "directory", "name": "Directory", "isAccessible": false }
+                            ],
+                            "nextCursor": null
+                        }
+                    })
+                }
+                "app/read" => {
+                    let params = serde_json::from_value::<AppsReadParams>(
+                        request.params.expect("app/read params"),
+                    )?;
+                    assert_eq!(params.thread_id.as_deref(), Some("thread-1"));
+                    assert!(!params.include_tools);
+                    assert!(params.app_ids.iter().all(|id| id.starts_with("app-")));
+                    read_batches.push(params.app_ids.len());
+
+                    if responses == MockResponses::FirstBatchFails && read_batches.len() == 1 {
+                        json!({
+                            "id": request_id,
+                            "error": { "code": -32603, "message": "metadata service unavailable" }
+                        })
+                    } else {
+                        let apps = params
+                            .app_ids
+                            .into_iter()
+                            .filter(|id| id != "app-102")
+                            .map(|id| json!({ "id": id, "name": format!("Connector {id}") }))
+                            .collect::<Vec<_>>();
+                        json!({
+                            "id": request_id,
+                            "result": { "apps": apps, "missingAppIds": ["app-102"] }
+                        })
+                    }
+                }
+                method => panic!("unexpected app-server request: {method}"),
+            };
+            websocket
+                .send(Message::Text(response.to_string().into()))
+                .await?;
+        }
+
+        Ok::<_, anyhow::Error>(read_batches)
+    });
+    let client = crate::connect_remote_app_server(
+        codex_app_server_client::RemoteAppServerEndpoint::WebSocket {
+            websocket_url,
+            auth_token: None,
+        },
+    )
+    .await
+    .expect("connect to mock app server");
+
+    let snapshot = fetch_installed_connector_mentions(
+        client.request_handle(),
+        Some("thread-1".to_string()),
+        /*force_refresh*/ true,
+    )
+    .await?;
+    client.shutdown().await?;
+    Ok((snapshot, server.await??))
+}
+
+#[tokio::test]
+async fn installed_connector_mentions_read_callable_apps_in_bounded_thread_scoped_batches()
+-> Result<()> {
+    let (snapshot, read_batches) = load_mock_mentions(MockResponses::FirstBatchFails).await?;
+    assert_eq!(read_batches, vec![100, 3]);
+    assert_eq!(snapshot.connectors.len(), 105);
+    assert!(snapshot.connectors[103..].iter().all(|app| !app.is_enabled));
+    let selected_apps = snapshot
+        .connectors
+        .iter()
+        .filter(|app| ["app-000", "app-100", "app-102"].contains(&app.id.as_str()))
+        .map(|app| app.name.as_str())
+        .collect::<Vec<_>>();
+    assert_eq!(
+        selected_apps,
+        ["Connector app-100", "Runtime 000", "app-102"]
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn installed_connector_mentions_fall_back_for_legacy_app_servers() -> Result<()> {
+    let (snapshot, read_batches) = load_mock_mentions(MockResponses::Unsupported).await?;
+
+    assert!(read_batches.is_empty());
+    assert_eq!(
+        snapshot.connectors,
+        serde_json::from_value::<Vec<AppInfo>>(json!([
+            { "id": "installed", "name": "Installed", "isAccessible": true },
+            { "id": "disabled", "name": "Disabled", "isAccessible": true, "isEnabled": false }
+        ]))?
+    );
+    Ok(())
+}

codex-rs/tui/src/app/event_dispatch.rs

@@ -822,6 +822,14 @@ impl App {
                     self.fetch_connectors_list(app_server, force_refetch);
                 }
             }
+            AppEvent::FetchInstalledConnectorMentions {
+                force_refresh,
+                generation,
+            } => {
+                if generation == self.chat_widget.connector_scope_generation() {
+                    self.fetch_installed_connector_mentions(app_server, force_refresh, generation);
+                }
+            }
             AppEvent::PluginInstallAuthAdvance { refresh_connectors } => {
                 if refresh_connectors {
                     self.chat_widget.refresh_connectors(/*force_refetch*/ true);
@@ -1353,6 +1361,20 @@ impl App {
                     self.chat_widget.on_connectors_loaded(result, is_final);
                 }
             }
+            AppEvent::InstalledConnectorMentionsLoaded {
+                thread_id,
+                cwd,
+                generation,
+                result,
+            } => {
+                if thread_id == self.current_displayed_thread_id()
+                    && cwd.as_path() == self.chat_widget.config_ref().cwd.as_path()
+                    && generation == self.chat_widget.connector_scope_generation()
+                {
+                    self.chat_widget
+                        .on_connector_mentions_loaded(generation, result);
+                }
+            }
             AppEvent::UpdateReasoningEffort(effort) => {
                 self.on_update_reasoning_effort(effort.clone());
                 self.sync_active_thread_reasoning_setting(app_server, effort)
@@ -2719,6 +2741,8 @@ impl App {
     fn refresh_plugin_mentions_after_config_write(&mut self) {
         self.chat_widget.refresh_plugin_mentions();
         self.chat_widget.submit_op(AppCommand::reload_user_config());
+        self.chat_widget
+            .refresh_connector_mentions(/*force_refresh*/ true);
     }
 
     async fn apply_keymap_clear(&mut self, context: String, action: String) {