openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use installed callable apps for TUI mentions (#39051)

Browse Source 
## Why

The app directory is a discovery catalog and can include apps that are not
callable in the active thread. Using it for `$` suggestions can therefore
offer mentions that cannot be used.

## What changed

- Load mention candidates from `app/installed`, enrich callable apps through
  bounded `app/read` batches, and retain compatibility with app servers that
  do not support `app/installed`.
- Keep the mention catalog separate from the `/apps` directory cache and scope
  results to the current thread, workspace, and account.
- Refresh mentions after app notifications, enablement changes, configuration
  updates, and apps MCP readiness, while immediately removing disabled apps.

## Testing

Add coverage for batching and fallback behavior, stale scoped results,
notification refreshes, enablement changes, MCP startup races, and filtering
non-callable apps from mentions.

GitOrigin-RevId: 4d9e20794aed30cc97120170792fb555b86fe1e7
This commit is contained in:
Charlie Marsh
2026-08-17 19:11:34 +00:00
committed by copyberry
parent d7d526b81d
commit fe5889928c
17 changed files with 759 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app.rs
+1
View File
@@ -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
+155
View File
@@ -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
+170
View File
@@ -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
+24
View File
@@ -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) {
codex-rs/tui/src/app/tests/connector_policy.rs
+69 -12
View File
@@ -5,7 +5,7 @@ use codex_app_server_protocol::AppListUpdatedNotification;
use pretty_assertions::assert_eq;
#[tokio::test]
async fn connector_directory_results_ignore_stale_thread_workspace_and_account() -> Result<()> {
async fn installed_connector_mentions_ignore_stale_thread_workspace_and_account() -> Result<()> {
let (mut app, _app_event_rx, _op_rx) = make_test_app_with_channels().await;
app.chat_widget
.set_feature_enabled(Feature::Apps, /*enabled*/ true);
@@ -44,12 +44,11 @@ async fn connector_directory_results_ignore_stale_thread_workspace_and_account()
app.handle_event(
&mut tui,
&mut app_server,
AppEvent::ConnectorsLoaded {
AppEvent::InstalledConnectorMentionsLoaded {
thread_id: candidate_thread,
cwd: candidate_cwd,
generation: candidate_generation,
result: Ok(snapshot.clone()),
is_final: true,
},
)
.await?;
@@ -100,7 +99,8 @@ async fn queued_connector_fetches_are_bound_to_their_original_account() -> Resul
}
#[tokio::test]
async fn unscoped_app_list_notifications_only_revalidate_the_current_directory() -> Result<()> {
async fn app_list_notifications_revalidate_installed_mentions_and_the_current_directory()
-> Result<()> {
let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await;
let thread_id = ThreadId::new();
let cwd = app.chat_widget.config_ref().cwd.to_path_buf();
@@ -117,6 +117,10 @@ async fn unscoped_app_list_notifications_only_revalidate_the_current_directory()
};
app.chat_widget
.on_connectors_loaded(Ok(scoped_snapshot.clone()), /*is_final*/ true);
app.chat_widget.on_connector_mentions_loaded(
app.chat_widget.connector_scope_generation(),
Ok(scoped_snapshot.clone()),
);
app.chat_widget.insert_str("$");
while app_event_rx.try_recv().is_ok() {}
@@ -138,19 +142,66 @@ async fn unscoped_app_list_notifications_only_revalidate_the_current_directory()
)
.await;
let refresh_requested = matches!(
app_event_rx.try_recv(),
Ok(AppEvent::FetchConnectorsList {
force_refetch: false,
..
})
if should_refresh {
assert!(matches!(
app_event_rx.try_recv(),
Ok(AppEvent::FetchInstalledConnectorMentions {
force_refresh: true,
..
})
));
}
assert_eq!(
matches!(
app_event_rx.try_recv(),
Ok(AppEvent::FetchConnectorsList {
force_refetch: false,
..
})
),
should_refresh
);
assert_eq!(refresh_requested, should_refresh);
let popup = render_bottom_popup(&app.chat_widget, /*width*/ 80);
assert!(popup.contains("Scoped App"));
assert!(!popup.contains("Old Workspace"));
}
let generation = app.chat_widget.connector_scope_generation();
app.chat_widget.on_connector_mentions_loaded(
generation,
Ok(ConnectorsSnapshot {
connectors: vec![serde_json::from_str(
r#"{"id":"old-workspace","name":"Old Workspace","isAccessible":true}"#,
)?],
}),
);
assert!(render_bottom_popup(&app.chat_widget, /*width*/ 80).contains("Old Workspace"));
assert!(app_event_rx.try_recv().is_err());
// Revocations refresh mentions even if discovery already has this notification.
let revoked_notification = ServerNotification::AppListUpdated(AppListUpdatedNotification {
data: vec![serde_json::from_str(
r#"{"id":"scoped-app","name":"Scoped App","isAccessible":true}"#,
)?],
});
app.handle_app_server_event(
&app_server,
AppServerEvent::ServerNotification(Box::new(revoked_notification)),
)
.await;
assert!(matches!(
app_event_rx.try_recv(),
Ok(AppEvent::FetchInstalledConnectorMentions {
force_refresh: true,
generation: requested_generation,
}) if requested_generation == generation
));
assert!(app_event_rx.try_recv().is_err());
assert!(render_bottom_popup(&app.chat_widget, /*width*/ 80).contains("Old Workspace"));
app.chat_widget
.on_connector_mentions_loaded(generation, Ok(scoped_snapshot.clone()));
assert!(!render_bottom_popup(&app.chat_widget, /*width*/ 80).contains("Old Workspace"));
let ready_app = r#"{"id":"newly-ready","name":"Newly Ready","isAccessible":true}"#;
let ready_notification = ServerNotification::AppListUpdated(AppListUpdatedNotification {
data: vec![serde_json::from_str(ready_app)?],
@@ -160,11 +211,17 @@ async fn unscoped_app_list_notifications_only_revalidate_the_current_directory()
AppServerEvent::ServerNotification(Box::new(ready_notification.clone())),
)
.await;
assert!(matches!(
app_event_rx.try_recv(),
Ok(AppEvent::FetchInstalledConnectorMentions {
force_refresh: true,
..
})
));
assert!(app_event_rx.try_recv().is_err());
assert!(!render_bottom_popup(&app.chat_widget, /*width*/ 80).contains("Newly Ready"));
let mut tui = crate::tui::test_support::make_test_tui()?;
let generation = app.chat_widget.connector_scope_generation();
for should_retry in [true, false, true, false] {
app.handle_event(
&mut tui,
codex-rs/tui/src/app_event.rs
+14
View File
@@ -510,6 +510,14 @@ pub(crate) enum AppEvent {
is_final: bool,
},
/// Thread-scoped installed applications that may actually be mentioned.
InstalledConnectorMentionsLoaded {
thread_id: Option<ThreadId>,
cwd: PathBuf,
generation: ConnectorScopeGeneration,
result: Result<ConnectorsSnapshot, String>,
},
/// Result of computing a `/diff` command.
DiffResult(PathBuf, String),
@@ -577,6 +585,12 @@ pub(crate) enum AppEvent {
generation: ConnectorScopeGeneration,
},
/// Refresh callable installed applications without loading the app directory.
FetchInstalledConnectorMentions {
force_refresh: bool,
generation: ConnectorScopeGeneration,
},
/// Fetch plugin marketplace state for the provided working directory.
FetchPluginsList {
cwd: PathBuf,
codex-rs/tui/src/chatwidget.rs
+1
View File
@@ -334,6 +334,7 @@ use crate::status_indicator_widget::StatusDetailsCapitalization;
use crate::text_formatting::truncate_text;
use crate::tui::FrameRequester;
mod command_lifecycle;
mod connector_mentions;
mod connectors;
mod constructor;
pub(crate) use self::connectors::ConnectorScopeGeneration;
codex-rs/tui/src/chatwidget/connector_mentions.rs
+62
View File
@@ -0,0 +1,62 @@
//! Thread-authorized installed-app mentions, independent from the discovery directory.
use super::*;
use crate::app_event::ConnectorsSnapshot;
impl ChatWidget {
/// Refresh thread-authorized installed apps, coalescing pending readiness retries.
pub(crate) fn refresh_connector_mentions(&mut self, force_refresh: bool) {
if !self.connectors_enabled() || self.thread_id.is_none() {
return;
}
if self.connectors.mention_refresh_in_flight {
self.connectors.mention_refresh_pending =
Some(self.connectors.mention_refresh_pending.unwrap_or_default() || force_refresh);
return;
}
self.connectors.locally_disabled.clear();
self.connectors.mention_refresh_in_flight = true;
self.app_event_tx
.send(AppEvent::FetchInstalledConnectorMentions {
force_refresh,
generation: self.connectors.generation,
});
}
/// Apply authorized installed apps while preserving only newer local revocations.
pub(crate) fn on_connector_mentions_loaded(
&mut self,
generation: ConnectorScopeGeneration,
result: Result<ConnectorsSnapshot, String>,
) {
if generation != self.connectors.generation {
return;
}
self.connectors.mention_refresh_in_flight = false;
self.connectors.notified_installed_app_ids = None;
match result {
Ok(mut snapshot) => {
self.connectors.installed_app_ids = snapshot
.connectors
.iter()
.filter(|connector| connector.is_accessible)
.map(|connector| connector.id.clone())
.collect();
snapshot.connectors.retain(|connector| {
connector.is_accessible
&& connector.is_enabled
&& !self.connectors.locally_disabled.contains(&connector.id)
});
self.connectors.mention_snapshot = Some(snapshot.clone());
self.bottom_pane.set_connectors_snapshot(Some(snapshot));
}
Err(err) => warn!("failed to refresh installed app mentions: {err}"),
}
if let Some(force_refresh) = self.connectors.mention_refresh_pending.take() {
self.refresh_connector_mentions(force_refresh);
}
}
}
codex-rs/tui/src/chatwidget/connectors.rs
+74 -48
View File
@@ -4,7 +4,7 @@ use super::*;
use crate::app_event::ConnectorsSnapshot;
use std::sync::atomic::AtomicU64;
/// Prevents stale directory requests from matching a replacement widget for the same thread.
/// Prevents stale requests from matching a replacement widget for the same thread.
static NEXT_CONNECTOR_SCOPE_GENERATION: AtomicU64 = AtomicU64::new(1);
/// Identifies one account-, workspace-, and thread-scoped connector state.
@@ -22,6 +22,7 @@ pub(super) enum ConnectorsCacheState {
#[derive(Debug, Default)]
pub(super) struct ConnectorsState {
/// Discovery-only catalog; its entries are not authorized composer mentions.
pub(super) cache: ConnectorsCacheState,
pub(super) partial_snapshot: Option<ConnectorsSnapshot>,
pub(super) prefetch_in_flight: bool,
@@ -30,31 +31,54 @@ pub(super) struct ConnectorsState {
pub(super) pending_notification: Option<Vec<AppInfo>>,
/// Prevents a recovery request from scheduling another retry if it also fails.
pub(super) notification_error_retry_in_flight: bool,
/// Installed apps that are callable in the current thread and account.
pub(super) mention_snapshot: Option<ConnectorsSnapshot>,
/// Installed identities include disabled apps omitted from composer mentions.
pub(super) installed_app_ids: HashSet<String>,
/// Suppress repeated notifications while their installed-app refresh is in flight.
pub(super) notified_installed_app_ids: Option<HashSet<String>>,
pub(super) mention_refresh_in_flight: bool,
/// `Some(false)` retains a cache-only retry when MCP readiness races another lookup.
pub(super) mention_refresh_pending: Option<bool>,
/// Revocations override only installed-app requests that were already in flight.
pub(super) locally_disabled: HashSet<String>,
/// Process-unique identity for this widget's account, workspace, and thread.
pub(super) generation: ConnectorScopeGeneration,
}
impl ChatWidget {
/// Refresh both discovery data and the independently authorized mention catalog.
pub(crate) fn refresh_connectors(&mut self, force_refetch: bool) {
self.queue_connectors_refresh(force_refetch);
self.refresh_connector_mentions(force_refetch);
}
pub(super) fn prefetch_connectors(&mut self) {
self.queue_connectors_refresh(/*force_refetch*/ false);
}
/// Revalidate directory notifications without granting their unscoped contents to this thread.
/// Revalidate app notifications without granting their unscoped contents to this thread.
pub(crate) fn refresh_connector_directory_after_notification(
&mut self,
connectors: Vec<AppInfo>,
) {
if self.thread_id.is_none()
|| !self.connectors_enabled()
|| self
.connectors
.partial_snapshot
.as_ref()
.is_some_and(|snapshot| snapshot.connectors == connectors)
if self.thread_id.is_none() || !self.connectors_enabled() {
return;
}
let installed_app_ids = connectors
.iter()
.filter(|connector| connector.is_accessible)
.map(|connector| connector.id.clone())
.collect::<HashSet<_>>();
if installed_app_ids != self.connectors.installed_app_ids
&& self.connectors.notified_installed_app_ids.as_ref() != Some(&installed_app_ids)
{
self.connectors.notified_installed_app_ids = Some(installed_app_ids);
self.refresh_connector_mentions(/*force_refresh*/ true);
}
if self
.connectors
.partial_snapshot
.as_ref()
.is_some_and(|snapshot| snapshot.connectors == connectors)
|| matches!(
&self.connectors.cache,
ConnectorsCacheState::Ready(snapshot) if snapshot.connectors == connectors
@@ -76,7 +100,7 @@ impl ChatWidget {
self.connectors.generation
}
/// Revoke cached apps and pending requests before their account or workspace changes.
/// Revoke cached app data and pending work before its account or thread changes.
pub(crate) fn invalidate_connector_scope(&mut self) {
let generation = ConnectorScopeGeneration(
NEXT_CONNECTOR_SCOPE_GENERATION.fetch_add(1, Ordering::Relaxed),
@@ -121,19 +145,16 @@ impl ChatWidget {
self.config.features.enabled(Feature::Apps) && self.has_chatgpt_account
}
/// Return only authorized installed apps, never entries from the discovery directory.
pub(super) fn connectors_for_mentions(&self) -> Option<&[AppInfo]> {
if !self.connectors_enabled() {
return None;
}
if let Some(snapshot) = &self.connectors.partial_snapshot {
return Some(snapshot.connectors.as_slice());
}
match &self.connectors.cache {
ConnectorsCacheState::Ready(snapshot) => Some(snapshot.connectors.as_slice()),
_ => None,
}
self.connectors
.mention_snapshot
.as_ref()
.map(|snapshot| snapshot.connectors.as_slice())
}
pub(crate) fn add_connectors_output(&mut self) {
@@ -149,6 +170,7 @@ impl ChatWidget {
let should_force_refetch = !self.connectors.prefetch_in_flight
|| matches!(connectors_cache, ConnectorsCacheState::Ready(_));
self.queue_connectors_refresh(should_force_refetch);
self.refresh_connector_mentions(/*force_refresh*/ true);
match connectors_cache {
ConnectorsCacheState::Ready(snapshot) => {
@@ -350,6 +372,7 @@ impl ChatWidget {
.map(str::to_string)
}
/// Update the `/apps` discovery catalog without changing composer authorization.
pub(crate) fn on_connectors_loaded(
&mut self,
result: Result<ConnectorsSnapshot, String>,
@@ -396,28 +419,23 @@ impl ChatWidget {
if is_final {
self.connectors.partial_snapshot = None;
self.refresh_connectors_popup_if_open(&snapshot.connectors);
self.connectors.cache = ConnectorsCacheState::Ready(snapshot.clone());
self.connectors.cache = ConnectorsCacheState::Ready(snapshot);
} else {
self.connectors.partial_snapshot = Some(snapshot.clone());
self.connectors.partial_snapshot = Some(snapshot);
}
self.bottom_pane.set_connectors_snapshot(Some(snapshot));
}
Err(err) => {
let partial_snapshot = self.connectors.partial_snapshot.take();
if let ConnectorsCacheState::Ready(snapshot) = &self.connectors.cache {
if matches!(self.connectors.cache, ConnectorsCacheState::Ready(_)) {
warn!("failed to refresh apps list; retaining current apps snapshot: {err}");
self.bottom_pane
.set_connectors_snapshot(Some(snapshot.clone()));
} else if let Some(snapshot) = partial_snapshot {
warn!(
"failed to load full apps list; falling back to installed apps snapshot: {err}"
);
self.refresh_connectors_popup_if_open(&snapshot.connectors);
self.connectors.cache = ConnectorsCacheState::Ready(snapshot.clone());
self.bottom_pane.set_connectors_snapshot(Some(snapshot));
self.connectors.cache = ConnectorsCacheState::Ready(snapshot);
} else {
self.connectors.cache = ConnectorsCacheState::Failed(err);
self.bottom_pane.set_connectors_snapshot(/*snapshot*/ None);
}
}
}
@@ -427,26 +445,34 @@ impl ChatWidget {
}
}
/// Revoke disabled apps immediately and revalidate permissions before enabling them.
pub(crate) fn update_connector_enabled(&mut self, connector_id: &str, enabled: bool) {
let ConnectorsCacheState::Ready(mut snapshot) = self.connectors.cache.clone() else {
return;
};
let mut changed = false;
for connector in &mut snapshot.connectors {
if connector.id == connector_id {
changed = connector.is_enabled != enabled;
connector.is_enabled = enabled;
break;
}
if let ConnectorsCacheState::Ready(mut snapshot) = self.connectors.cache.clone()
&& let Some(connector) = snapshot
.connectors
.iter_mut()
.find(|connector| connector.id == connector_id)
&& connector.is_enabled != enabled
{
connector.is_enabled = enabled;
self.refresh_connectors_popup_if_open(&snapshot.connectors);
self.connectors.cache = ConnectorsCacheState::Ready(snapshot);
}
if !changed {
return;
if enabled {
self.connectors.locally_disabled.remove(connector_id);
} else {
self.connectors
.locally_disabled
.insert(connector_id.to_string());
}
self.refresh_connectors_popup_if_open(&snapshot.connectors);
self.connectors.cache = ConnectorsCacheState::Ready(snapshot.clone());
self.bottom_pane.set_connectors_snapshot(Some(snapshot));
if !enabled && let Some(mentions) = self.connectors.mention_snapshot.as_mut() {
mentions
.connectors
.retain(|connector| connector.id != connector_id);
self.bottom_pane
.set_connectors_snapshot(Some(mentions.clone()));
}
self.refresh_connector_mentions(/*force_refresh*/ enabled);
}
}
codex-rs/tui/src/chatwidget/mcp_startup.rs
+6
View File
@@ -262,10 +262,13 @@ impl ChatWidget {
.starts_with(MCP_STARTUP_MULTI_HEADER_PREFIX)
}
/// Update startup state and retry installed-app discovery when its MCP server becomes ready.
pub(super) fn on_mcp_server_status_updated(
&mut self,
notification: McpServerStatusUpdatedNotification,
) {
let refresh_connector_mentions = notification.name == "codex_apps"
&& notification.status == McpServerStartupState::Ready;
let status = match notification.status {
McpServerStartupState::Starting => McpStartupStatus::Starting,
McpServerStartupState::Ready => McpStartupStatus::Ready,
@@ -281,5 +284,8 @@ impl ChatWidget {
status,
/*complete_when_settled*/ true,
);
if refresh_connector_mentions {
self.refresh_connector_mentions(/*force_refresh*/ false);
}
}
}
codex-rs/tui/src/chatwidget/plugins.rs
+13 -3
View File
@@ -953,11 +953,21 @@ impl ChatWidget {
}
fn plugin_install_auth_app_is_installed(&self, app_id: &str) -> bool {
self.connectors_for_mentions().is_some_and(|connectors| {
connectors
if !self.connectors_enabled() {
return false;
}
let connectors = &self.connectors;
connectors.installed_app_ids.contains(app_id)
|| connectors
.partial_snapshot
.iter()
.chain(match &connectors.cache {
super::connectors::ConnectorsCacheState::Ready(snapshot) => Some(snapshot),
_ => None,
})
.flat_map(|snapshot| &snapshot.connectors)
.any(|connector| connector.id == app_id && connector.is_accessible)
})
}
fn finish_plugin_install_auth_flow(&mut self, abandoned: bool) {
codex-rs/tui/src/chatwidget/session_flow.rs
+1 -3
View File
@@ -148,9 +148,7 @@ impl ChatWidget {
}
self.transcript.saw_copy_source_this_turn = false;
self.refresh_skills_for_current_cwd(/*force_reload*/ true);
if self.connectors_enabled() {
self.prefetch_connectors();
}
self.refresh_connector_mentions(/*force_refresh*/ false);
self.submit_initial_user_message_if_pending();
if display == SessionConfiguredDisplay::Normal
&& let Some(forked_from_id) = forked_from_id
codex-rs/tui/src/chatwidget/settings.rs
+2 -4
View File
@@ -244,9 +244,7 @@ impl ChatWidget {
self.has_codex_backend_auth = has_codex_backend_auth;
self.bottom_pane
.set_connectors_enabled(self.connectors_enabled());
if self.thread_id.is_some() {
self.prefetch_connectors();
}
self.refresh_connector_mentions(/*force_refresh*/ false);
self.bottom_pane
.set_token_activity_command_enabled(has_codex_backend_auth);
self.refresh_status_surfaces();
@@ -538,7 +536,7 @@ impl ChatWidget {
if cwd_changed {
self.invalidate_connector_scope();
self.refresh_skills_for_current_cwd(/*force_reload*/ true);
self.prefetch_connectors();
self.refresh_connector_mentions(/*force_refresh*/ false);
}
self.refresh_plugin_mentions();
self.request_redraw();
codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__apps_mentions_only_callable_installed.snap
+22
View File
@@ -0,0 +1,22 @@
---
source: tui/src/chatwidget/tests/popups_and_settings.rs
expression: snapshot
---
› $
Google Drive [App] Connected files
Press enter to insert or esc to close
--- plugin authentication ---
Plugins
Plugin plugin installed.
App setup 1/1: % Arabica UAE
Already installed in this session.
ChatGPT apps link unavailable This app did not provide an install/manage
URL.
› 1. Continue Advance to the next app.
2. Skip remaining app setup Stop this follow-up flow for this plugin.
Press esc to close.
codex-rs/tui/src/chatwidget/tests/app_server.rs
+9 -13
View File
@@ -344,21 +344,17 @@ async fn thread_settings_updated_updates_visible_state_without_transcript() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.2")).await;
set_chatgpt_auth(&mut chat);
set_fast_mode_test_catalog(&mut chat);
chat.set_feature_enabled(Feature::Apps, /*enabled*/ true);
let thread_id = ThreadId::new();
let mut session = configured_thread_session(thread_id);
session.cwd = test_path_buf("/tmp/original-workspace").abs();
chat.handle_thread_session(session);
let generation = chat.connector_scope_generation();
chat.on_connectors_loaded(
Ok(crate::app_event::ConnectorsSnapshot {
connectors: vec![
serde_json::from_str(r#"{"id":"old","name":"Old","isAccessible":true}"#)
.expect("valid app"),
],
}),
/*is_final*/ true,
);
chat.prefetch_connectors();
let previous_generation = chat.connector_scope_generation();
let old_app = serde_json::from_str(r#"{"id":"old","name":"Old","isAccessible":true}"#)
.expect("valid app");
chat.connectors.mention_snapshot = Some(crate::app_event::ConnectorsSnapshot {
connectors: vec![old_app],
});
let _ = drain_insert_history(&mut rx);
chat.handle_server_notification(
@@ -366,9 +362,9 @@ async fn thread_settings_updated_updates_visible_state_without_transcript() {
/*replay_kind*/ None,
);
assert_ne!(chat.connector_scope_generation(), generation);
assert_ne!(chat.connector_scope_generation(), previous_generation);
assert!(chat.connectors_for_mentions().is_none());
assert!(chat.connectors.prefetch_in_flight);
assert!(chat.connectors.mention_refresh_in_flight);
assert_eq!(chat.current_model(), "gpt-5.4");
assert_eq!(
chat.current_reasoning_effort(),
codex-rs/tui/src/chatwidget/tests/mcp_startup.rs
+22
View File
@@ -69,6 +69,28 @@ async fn mcp_startup_ignores_status_for_other_thread() {
);
}
#[tokio::test]
async fn mcp_apps_readiness_retries_an_in_flight_installed_mention_lookup() {
let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
set_chatgpt_auth(&mut chat);
chat.set_feature_enabled(Feature::Apps, /*enabled*/ true);
chat.refresh_connector_mentions(/*force_refresh*/ false);
chat.on_mcp_server_status_updated(McpServerStatusUpdatedNotification {
thread_id: None,
name: "codex_apps".to_string(),
status: McpServerStartupState::Ready,
error: None,
failure_reason: None,
});
assert_eq!(chat.connectors.mention_refresh_pending, Some(false));
let snapshot = crate::app_event::ConnectorsSnapshot { connectors: vec![] };
chat.on_connector_mentions_loaded(chat.connector_scope_generation(), Ok(snapshot));
assert!(chat.connectors.mention_refresh_in_flight);
}
#[tokio::test]
async fn mcp_startup_dedupes_same_round_duplicate_failure_warning() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
codex-rs/tui/src/chatwidget/tests/popups_and_settings.rs
+114 -33
View File
@@ -2087,7 +2087,8 @@ async fn plugins_popup_search_no_matches_and_backspace_restores_results() {
#[tokio::test]
async fn apps_popup_stays_loading_until_final_snapshot_updates() {
let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
set_chatgpt_auth(&mut chat);
chat.config
.features
@@ -2124,6 +2125,14 @@ async fn apps_popup_stays_loading_until_final_snapshot_updates() {
chat.connectors.prefetch_in_flight,
"expected /apps to trigger a forced connectors refresh"
);
assert_matches!(
rx.try_recv(),
Ok(AppEvent::FetchConnectorsList { force_refetch, .. }) if force_refetch
);
assert_matches!(
rx.try_recv(),
Ok(AppEvent::FetchInstalledConnectorMentions { force_refresh, .. }) if force_refresh
);
let before = render_bottom_popup(&chat, /*width*/ 80);
assert!(
@@ -2187,7 +2196,8 @@ async fn apps_popup_stays_loading_until_final_snapshot_updates() {
#[tokio::test]
async fn apps_notification_update_excludes_inaccessible_apps_from_mentions() {
let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
set_chatgpt_auth(&mut chat);
chat.config
.features
@@ -2230,7 +2240,7 @@ async fn apps_notification_update_excludes_inaccessible_apps_from_mentions() {
app_metadata: None,
labels: None,
install_url: Some("https://example.test/arabica".to_string()),
is_accessible: false,
is_accessible: true,
is_enabled: true,
plugin_display_names: Vec::new(),
},
@@ -2239,44 +2249,116 @@ async fn apps_notification_update_excludes_inaccessible_apps_from_mentions() {
/*is_final*/ false,
);
assert_matches!(
&chat.connectors.partial_snapshot,
Some(snapshot)
if snapshot
.connectors
.iter()
.find(|connector| connector.id == "arabica_uae")
.is_some_and(|connector| !connector.is_accessible)
assert!(chat.connectors_for_mentions().is_none());
let mut installed = chat
.connectors
.partial_snapshot
.as_ref()
.expect("directory notification should remain available to /apps")
.connectors
.clone();
installed[1].is_enabled = false;
chat.on_connector_mentions_loaded(
chat.connector_scope_generation(),
Ok(ConnectorsSnapshot {
connectors: installed,
}),
);
let popup = render_bottom_popup(&chat, /*width*/ 80);
assert!(
popup.contains("Google Drive"),
"expected accessible apps to appear in the mention popup, got:\n{popup}"
"expected callable installed apps to appear in the mention popup, got:\n{popup}"
);
assert!(
!popup.contains("% Arabica UAE"),
"did not expect an inaccessible directory app in the mention popup, got:\n{popup}"
"directory accessibility must not make an app callable, got:\n{popup}"
);
chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
chat.insert_str("$arabica-uae ");
chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
assert_matches!(
next_submit_op(&mut op_rx),
Op::UserTurn { items, .. }
if matches!(items.as_slice(), [
UserInput::Text { .. },
UserInput::Mention { name, path },
] if name == "Google Drive" && path == "app://google_drive")
);
chat.connectors.partial_snapshot = None;
assert_matches!(&chat.connectors.cache, ConnectorsCacheState::Uninitialized);
for (app_id, app_name) in [
("arabica_uae", "% Arabica UAE"),
("google_drive", "Google Drive"),
] {
chat.on_plugin_install_loaded(
chat.config.cwd.to_path_buf(),
crate::app_event::PluginLocation::Remote {
marketplace_name: "marketplace".to_string(),
},
"plugin".to_string(),
"Plugin".to_string(),
Ok(serde_json::from_value(serde_json::json!({
"authPolicy": "ON_INSTALL",
"appsNeedingAuth": [{ "id": app_id, "name": app_name }],
}))
.expect("valid plugin installation response")),
);
let auth_popup = render_bottom_popup(&chat, /*width*/ 80);
assert!(auth_popup.contains("Already installed") && auth_popup.contains("Continue"));
if app_id == "arabica_uae" {
let snapshot = normalize_snapshot_paths(format!(
"{popup}\n\n--- plugin authentication ---\n{auth_popup}"
));
assert_chatwidget_snapshot!("apps_mentions_only_callable_installed", snapshot);
}
}
}
#[tokio::test]
async fn account_changes_revoke_app_directory_and_restart_scoped_requests() {
async fn apps_installed_mentions_revoke_access_and_reject_stale_account_results() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
set_chatgpt_auth(&mut chat);
chat.on_connectors_loaded(
Ok(ConnectorsSnapshot {
connectors: vec![
serde_json::from_str(
r#"{"id":"old-account","name":"Old Account","isAccessible":true}"#,
)
.expect("valid app"),
],
}),
/*is_final*/ true,
);
chat.set_feature_enabled(Feature::Apps, /*enabled*/ true);
let generation = chat.connector_scope_generation();
let connector_id = "account-app";
let connector =
serde_json::from_str(r#"{"id":"account-app","name":"Account app","isAccessible":true}"#)
.expect("valid installed app");
let snapshot = ConnectorsSnapshot {
connectors: vec![connector],
};
chat.on_connector_mentions_loaded(generation, Ok(snapshot.clone()));
assert!(chat.connectors.installed_app_ids.contains(connector_id));
chat.refresh_connector_mentions(/*force_refresh*/ false);
rx.try_recv().expect("pre-disable mention refresh");
chat.update_connector_enabled(connector_id, /*enabled*/ false);
chat.on_connector_mentions_loaded(generation, Ok(snapshot.clone()));
assert_eq!(chat.connectors_for_mentions(), Some([].as_slice()));
rx.try_recv().expect("post-disable mention refresh");
chat.on_connector_mentions_loaded(generation, Ok(snapshot.clone()));
assert_eq!(
chat.connectors_for_mentions(),
Some(snapshot.connectors.as_slice())
);
chat.on_connectors_loaded(Ok(snapshot.clone()), /*is_final*/ true);
for enabled in [false, true] {
chat.update_connector_enabled(connector_id, enabled);
assert_eq!(chat.connectors_for_mentions(), Some([].as_slice()));
assert_matches!(
rx.try_recv(),
Ok(AppEvent::FetchInstalledConnectorMentions { force_refresh, .. })
if force_refresh == enabled
);
let mut refreshed = snapshot.clone();
refreshed.connectors[0].is_enabled = enabled;
chat.on_connector_mentions_loaded(generation, Ok(refreshed));
}
chat.update_account_state(
/*status_account_display*/ None, /*plan_type*/ None,
@@ -2284,15 +2366,14 @@ async fn account_changes_revoke_app_directory_and_restart_scoped_requests() {
);
assert_ne!(chat.connector_scope_generation(), generation);
assert_matches!(&chat.connectors.cache, ConnectorsCacheState::Uninitialized);
assert!(chat.connectors_for_mentions().is_none());
assert!(chat.connectors.prefetch_in_flight);
assert_matches!(
rx.try_recv(),
Ok(AppEvent::FetchConnectorsList {
force_refetch: false,
generation
}) if generation == chat.connector_scope_generation()
);
assert!(chat.connectors.mention_refresh_in_flight);
chat.on_connector_mentions_loaded(generation, Ok(snapshot));
assert!(chat.connectors_for_mentions().is_none());
assert!(chat.connectors.mention_refresh_in_flight);
assert!(chat.connectors.installed_app_ids.is_empty());
let (mut replacement, _, _) = make_chatwidget_manual(/*model_override*/ None).await;
replacement.invalidate_connector_scope();