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

Allow Apps catalog discovery before the initial tool inventory completes (#58352)

Browse Source 
## Why

Plugin and skill catalog reads currently wait for the Apps server's initial `tools/list` to finish, even after its MCP transport is initialized.

## What changed

Add the disabled-by-default `features.apps_early_resource_discovery` flag. When enabled, host-owned Apps requests for `mcp/plugin` and `mcp/skill` resources can use the initialized transport while the initial tool inventory is pending. Server readiness still waits for the inventory to complete, and retries use normal readiness.

Bound early client acquisition by the configured startup timeout, propagate startup failures and cancellation, and reject pages if the connection, authentication scope, or resource cache generation changes during discovery.

## Testing

Add coverage for gated discovery, concurrent plugin and skill reads, startup timeouts and failures, cancellation, cache invalidation, and account switches. Verify that early discovery reuses a single handshake and tool inventory, including with the newer MCP protocol and connection sharing enabled.

GitOrigin-RevId: 955b1f6d43b733bdf9763c1caf2ce6913c686c81
This commit is contained in:
bkotsopoulos
2026-10-30 14:47:56 +00:00
committed by copyberry
parent 80f469ce8a
commit 9630b2e019
11 changed files with 604 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/codex-mcp/src/connection_manager.rs
+12
View File
@@ -103,6 +103,9 @@ pub(crate) struct McpServerConnection {
// Startup-only budget; changing it must not replace a ready connection.
startup_timeout: Duration,
startup_trigger: Option<watch::Sender<bool>>,
// Only the initial host-owned Apps attempt publishes here. A retry uses normal readiness.
initialized_apps:
Option<watch::Receiver<Option<std::sync::Weak<codex_rmcp_client::RmcpClient>>>>,
_diagnostics_guard: GaugeGuard,
}
@@ -895,6 +898,13 @@ impl McpConnectionSet {
},
)
});
let (publish_apps, initialized_apps) =
if is_host_owned_codex_apps && config.apps_early_resource_discovery {
let (tx, rx) = watch::channel(None);
(Some(tx), Some(rx))
} else {
(None, None)
};
let mut async_managed_client = AsyncManagedClient::new(
ema_auth.clone(),
server_name.clone(),
@@ -925,6 +935,7 @@ impl McpConnectionSet {
protocol_mode,
catalog_item_limit,
connection_pool,
publish_apps,
);
// A forced reconnect must also wait for the previously published helper to exit.
if let Some(previous) = existing_ema.and_then(|set| set.servers.get(&server_name)) {
@@ -947,6 +958,7 @@ impl McpConnectionSet {
legacy_fallback,
startup_timeout,
startup_trigger,
initialized_apps,
_diagnostics_guard: LIVE_CONNECTIONS.track(),
}),
protocol_mode,
codex-rs/codex-mcp/src/connection_manager/resources.rs
+82
View File
@@ -1,5 +1,12 @@
use super::McpServerConnection;
use crate::CODEX_APPS_MCP_SERVER_NAME;
use crate::McpRuntime;
use codex_rmcp_client::RmcpClient;
use rmcp::model::ServerResult;
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::Weak;
use std::sync::atomic::Ordering;
use std::time::Duration;
use anyhow::Context;
@@ -183,6 +190,24 @@ impl McpConnectionSet {
.map(|view| view.metadata.environment_id.as_str())
}
/// Returns the first Apps startup attempt only while its inventory is pending.
pub(crate) fn early_apps_discovery(
&self,
) -> Option<(&Arc<McpServerConnection>, Option<Duration>)> {
if !self.config.as_ref()?.apps_early_resource_discovery {
return None;
}
let view = self.servers.get(CODEX_APPS_MCP_SERVER_NAME)?;
let connection = &view.connection;
if connection.initialized_apps.is_some()
&& !connection.client.startup_complete.load(Ordering::Acquire)
{
Some((connection, view.tool_timeout))
} else {
None
}
}
pub(crate) async fn client_by_name(
&self,
name: &str,
@@ -199,3 +224,60 @@ impl McpConnectionSet {
Ok((client, view.tool_timeout))
}
}
impl McpServerConnection {
/// Borrows the authenticated transport while startup owns its inventory. Discard the page
/// if startup fails or the runtime replaces its connection, auth scope, or resource generation.
pub(crate) async fn with_early_apps_client(
self: &Arc<Self>,
runtime: &McpRuntime,
timeout: Option<Duration>,
request: impl AsyncFnOnce(Arc<RmcpClient>, Option<Duration>) -> Result<ServerResult>,
) -> Result<ServerResult> {
let auth = runtime.auth_cache_key_for_server(CODEX_APPS_MCP_SERVER_NAME);
let key = runtime
.resource_cache_key(CODEX_APPS_MCP_SERVER_NAME)
.context("Apps registration was removed")?;
anyhow::ensure!(
key.connection.ptr_eq(&Arc::downgrade(self)),
"Apps connection was replaced"
);
let client = self.initialized_apps_client().await?;
let response = tokio::select! {
biased;
_ = self.client.cancel_token.cancelled() => anyhow::bail!("Apps startup cancelled"),
Err(error) = self.client.client.clone() => return Err(error.into()),
response = request(client, timeout) => response?,
};
anyhow::ensure!(
Some(key) == runtime.resource_cache_key(CODEX_APPS_MCP_SERVER_NAME)
&& auth == runtime.auth_cache_key_for_server(CODEX_APPS_MCP_SERVER_NAME),
"Apps connection changed during catalog discovery"
);
Ok(response)
}
pub(super) async fn initialized_apps_client(&self) -> Result<Arc<RmcpClient>> {
let mut initialized = self
.initialized_apps
.clone()
.context("Apps discovery is disabled")?;
let initialized_transport = async {
initialized
.wait_for(Option::is_some)
.await
.ok()
.and_then(|value| value.as_ref().and_then(Weak::upgrade))
};
let acquire = async {
tokio::select! {
biased;
ready = self.client.client.clone() => Ok(ready?.client),
Some(client) = initialized_transport => Ok(client),
}
};
tokio::time::timeout(self.startup_timeout, acquire)
.await
.context("timed out waiting for the initialized Apps client")?
}
}
codex-rs/codex-mcp/src/connection_manager_tests.rs
+33
View File
@@ -149,6 +149,7 @@ impl McpConnectionSet {
identity: None,
client,
startup_timeout: DEFAULT_STARTUP_TIMEOUT,
initialized_apps: None,
startup_trigger: None,
legacy_fallback: None,
_diagnostics_guard: LIVE_CONNECTIONS.track(),
@@ -622,6 +623,35 @@ pub(crate) async fn create_ready_async_managed_client(tools: Vec<ToolInfo>) -> A
}
}
#[tokio::test(start_paused = true)]
async fn stalled_apps_client_publication_expires_at_the_configured_startup_timeout() {
let mut manager = McpConnectionSet::empty(/*prefix_mcp_tool_names*/ false);
let mut client = create_ready_async_managed_client(Vec::new()).await;
client.client = futures::future::pending().boxed().shared();
client.startup_complete.store(false, Ordering::Release);
manager.insert_test_client(CODEX_APPS_MCP_SERVER_NAME, client);
let view = manager.servers.get_mut(CODEX_APPS_MCP_SERVER_NAME).unwrap();
let connection = Arc::get_mut(&mut view.connection).unwrap();
let (_publisher, initialized) = watch::channel(None);
connection.initialized_apps = Some(initialized);
connection.startup_timeout = Duration::from_secs(/*secs*/ 2);
let mut borrow = Box::pin(connection.initialized_apps_client());
assert!(futures::poll!(&mut borrow).is_pending());
tokio::time::advance(Duration::from_secs(/*secs*/ 1)).await;
assert!(futures::poll!(&mut borrow).is_pending());
tokio::time::advance(Duration::from_secs(/*secs*/ 1)).await;
let std::task::Poll::Ready(Err(error)) = futures::poll!(&mut borrow) else {
panic!(
"expected the early borrow to expire while startup and publication were still pending"
);
};
assert_eq!(
error.to_string(),
"timed out waiting for the initialized Apps client"
);
}
#[tokio::test]
async fn connection_statuses_observe_clients_without_starting_them() {
use codex_protocol::mcp::McpServerConnectionStatus as Status;
@@ -5516,6 +5546,7 @@ async fn manager_with_reusable_ready_server(
startup_timeout: config
.startup_timeout_sec
.unwrap_or(DEFAULT_STARTUP_TIMEOUT),
initialized_apps: None,
startup_trigger: None,
legacy_fallback: None,
_diagnostics_guard: LIVE_CONNECTIONS.track(),
@@ -5974,6 +6005,7 @@ async fn reconciliation_reuses_connection_without_relisting_regular_tools() -> a
startup_timeout: config
.startup_timeout_sec
.unwrap_or(DEFAULT_STARTUP_TIMEOUT),
initialized_apps: None,
startup_trigger: None,
legacy_fallback: None,
_diagnostics_guard: LIVE_CONNECTIONS.track(),
@@ -6811,6 +6843,7 @@ async fn reconciliation_replaces_closed_connections() -> anyhow::Result<()> {
startup_timeout: config
.startup_timeout_sec
.unwrap_or(DEFAULT_STARTUP_TIMEOUT),
initialized_apps: None,
startup_trigger: None,
legacy_fallback: None,
_diagnostics_guard: LIVE_CONNECTIONS.track(),
codex-rs/codex-mcp/src/mcp/mod.rs
+2
View File
@@ -183,6 +183,8 @@ pub struct McpConfig {
/// ChatGPT auth is checked separately before a materialized host-owned Apps
/// server can be used.
pub apps_enabled: bool,
/// Allow hosted Apps catalog reads after MCP initialization, before tools are ready.
pub apps_early_resource_discovery: bool,
/// Whether model-visible MCP tool namespaces should keep the legacy
/// `mcp__` prefix.
pub prefix_mcp_tool_names: bool,
codex-rs/codex-mcp/src/mcp/mod_tests.rs
+1
View File
@@ -235,6 +235,7 @@ pub(crate) fn test_mcp_config(codex_home: PathBuf) -> McpConfig {
codex_linux_sandbox_exe: None,
use_legacy_landlock: false,
apps_enabled: false,
apps_early_resource_discovery: false,
prefix_mcp_tool_names: true,
non_prefixed_mcp_tool_servers: Vec::new(),
protocol_mode: McpProtocolMode::Legacy,
codex-rs/codex-mcp/src/resource_client.rs
+18 -8
View File
@@ -304,17 +304,27 @@ impl McpResourceClient {
&self,
params: CodexAppsResourceListParams,
) -> Result<McpResourcePage> {
let catalog = matches!(params.mime_type.as_str(), "mcp/plugin" | "mcp/skill");
let params = serde_json::to_value(params)
.context("failed to serialize Codex Apps resource params")?;
let connections = self.runtime.latest_host_owned_codex_apps_connections()?;
let (managed, timeout) = connections
.client_by_name(CODEX_APPS_MCP_SERVER_NAME)
.await?;
let result = managed
.client
.send_custom_request_with_timeout("resources/list", Some(params), timeout)
.await
.context("resources/list failed for `codex_apps`")?;
let list = async move |client: Arc<RmcpClient>, timeout| {
client
.send_custom_request_with_timeout("resources/list", Some(params), timeout)
.await
.context("resources/list failed for `codex_apps`")
};
let result =
if catalog && let Some((connection, timeout)) = connections.early_apps_discovery() {
connection
.with_early_apps_client(&self.runtime, timeout, list)
.await?
} else {
let (managed, timeout) = connections
.client_by_name(CODEX_APPS_MCP_SERVER_NAME)
.await?;
list(managed.client, timeout).await?
};
let result = match result {
ServerResult::ListResourcesResult(result) => result,
ServerResult::CustomResult(result) => result
codex-rs/codex-mcp/src/rmcp_client.rs
+15 -4
View File
@@ -297,7 +297,11 @@ struct ManagedClientStartup {
}
impl ManagedClientStartup {
fn start(&self, sharing_declined: Option<CancellationToken>) -> ManagedClientFuture {
fn start(
&self,
sharing_declined: Option<CancellationToken>,
publish_apps: Option<watch::Sender<Option<std::sync::Weak<RmcpClient>>>>,
) -> ManagedClientFuture {
// A new attempt must not expose capabilities from an earlier connection.
*self
.server_capabilities
@@ -378,6 +382,7 @@ impl ManagedClientStartup {
client,
StartServerTaskParams {
sharing_declined,
publish_apps,
is_codex_apps_mcp_server,
startup_timeout: Some(startup_timeout),
tx_event,
@@ -492,6 +497,7 @@ impl AsyncManagedClient {
protocol_mode: McpProtocolMode,
catalog_item_limit: usize,
connection_pool: Option<(crate::McpConnectionPool, ConnectionKey)>,
publish_apps: Option<watch::Sender<Option<std::sync::Weak<RmcpClient>>>>,
) -> Self {
let is_codex_apps_mcp_server = server_name == CODEX_APPS_MCP_SERVER_NAME;
let reconnect_server_name = server_name.clone();
@@ -564,7 +570,7 @@ impl AsyncManagedClient {
|cancellation, sharing_declined| {
let mut shared_startup = (*startup).clone();
shared_startup.cancel_token = cancellation;
shared_startup.start(sharing_declined)
shared_startup.start(sharing_declined, /*publish_apps*/ None)
},
)
.await
@@ -586,13 +592,13 @@ impl AsyncManagedClient {
.shared();
(client, Some(connection_slot))
} else {
(startup.start(/*sharing_declined*/ None), None)
(startup.start(/*sharing_declined*/ None, publish_apps), None)
};
let startup_reconnect = is_codex_apps_mcp_server.then(|| {
let startup = Arc::clone(&startup);
Arc::new(
CodexAppsStartupReconnect::new(Arc::new(move || {
startup.start(/*sharing_declined*/ None)
startup.start(/*sharing_declined*/ None, /*publish_apps*/ None)
}))
.with_startup_status_context(
startup_submit_id,
@@ -1116,6 +1122,7 @@ async fn start_server_task(
) -> Result<ManagedClient, StartupOutcomeError> {
let StartServerTaskParams {
sharing_declined,
publish_apps,
is_codex_apps_mcp_server,
startup_timeout,
tx_event,
@@ -1238,6 +1245,9 @@ async fn start_server_task(
let fetch_ticket = codex_apps_tools_cache_context
.as_ref()
.map(|context| context.begin_fetch(ConnectorRuntimeFetchSource::Startup));
if let Some(tx) = &publish_apps {
tx.send_replace(Some(Arc::downgrade(&client)));
}
let (tools, meta) = list_tools_for_client_uncached(
&server_name,
is_codex_apps_mcp_server,
@@ -1392,6 +1402,7 @@ fn mcp_server_info_from_implementation(
}
struct StartServerTaskParams {
publish_apps: Option<watch::Sender<Option<std::sync::Weak<RmcpClient>>>>,
sharing_declined: Option<CancellationToken>,
server_capabilities: Arc<StdMutex<Option<serde_json::Value>>>,
is_codex_apps_mcp_server: bool,
codex-rs/core/config.schema.json
+6
View File
@@ -863,6 +863,9 @@
"apps": {
"type": "boolean"
},
"apps_early_resource_discovery": {
"type": "boolean"
},
"apps_mcp_path_override": {
"anyOf": [
{
@@ -7561,6 +7564,9 @@
"apps": {
"type": "boolean"
},
"apps_early_resource_discovery": {
"type": "boolean"
},
"apps_mcp_path_override": {
"anyOf": [
{
codex-rs/core/src/config/mod.rs
+3
View File
@@ -1963,6 +1963,9 @@ impl Config {
codex_linux_sandbox_exe: self.codex_linux_sandbox_exe.clone(),
use_legacy_landlock: self.features.use_legacy_landlock(),
apps_enabled: self.features.enabled(Feature::Apps),
apps_early_resource_discovery: self
.features
.enabled(Feature::AppsEarlyResourceDiscovery),
prefix_mcp_tool_names: self.prefix_mcp_tool_names(),
non_prefixed_mcp_tool_servers: if self
.features
codex-rs/ext/mcp/src/event_stream_tests.rs
+424 -15
View File
@@ -4,8 +4,12 @@
use std::collections::HashMap;
use std::convert::Infallible;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::time::Duration;
use anyhow::Context;
use anyhow::Result;
use axum::Json;
use axum::Router;
@@ -25,6 +29,7 @@ use codex_login::ExternalAuth;
use codex_login::ExternalAuthFuture;
use codex_login::ExternalAuthRefreshContext;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_mcp::CodexAppsResourceListParams;
use codex_mcp::EffectiveMcpServer;
use codex_mcp::McpEventNotification;
use codex_mcp::McpResourceClient;
@@ -38,6 +43,7 @@ use futures::StreamExt;
use pretty_assertions::assert_eq;
use serde_json::Value;
use serde_json::json;
use tokio::sync::Notify;
use tokio::sync::mpsc;
use tokio::time::timeout;
use tokio_util::sync::CancellationToken;
@@ -66,6 +72,22 @@ impl TestStream {
}
}
#[derive(Default)]
struct BlockedTools {
started: Notify,
release: Notify,
handshakes: AtomicUsize,
inventories: AtomicUsize,
inventory_returned: AtomicBool,
inventory_error: Option<&'static str>,
stalled_initialize_timeout: Option<Duration>,
modern_with_sharing: bool,
held_mime_type: Option<&'static str>,
session_id: Option<&'static str>,
resource_started: Notify,
release_resource: Notify,
}
pub(super) struct Fixture {
_home: tempfile::TempDir,
_server: tokio_util::task::AbortOnDropHandle<std::io::Result<()>>,
@@ -77,17 +99,92 @@ pub(super) struct Fixture {
impl Fixture {
pub async fn new() -> Result<Self> {
let fixture = Self::start(/*blocked_tools*/ None, /*early_resources*/ false).await?;
assert!(
fixture
.runtime
.latest_wait_for_server_ready(CODEX_APPS_MCP_SERVER_NAME, WAIT)
.await
);
Ok(fixture)
}
async fn start(
blocked_tools: Option<Arc<BlockedTools>>,
early_resources: bool,
) -> Result<Self> {
let (opened_tx, opened) = mpsc::unbounded_channel();
let modern_with_sharing = blocked_tools
.as_ref()
.is_some_and(|blocked| blocked.modern_with_sharing);
let stalled_initialize_timeout = blocked_tools
.as_ref()
.and_then(|blocked| blocked.stalled_initialize_timeout);
let router = Router::new().fallback(post(
move |headers: HeaderMap, Json(request): Json<Value>| {
let opened = opened_tx.clone();
let blocked_tools = blocked_tools.clone();
async move {
let result = match request["method"].as_str() {
Some("initialize") => json!({
"protocolVersion": "2025-06-18", "capabilities": {},
"serverInfo": {"name": "test", "version": "1"},
}),
Some("tools/list") => json!({"tools": []}),
let method = request["method"].as_str();
let session_id = blocked_tools.as_ref().and_then(|blocked| blocked.session_id);
if let Some(session_id) = session_id && method != Some("initialize") {
assert_eq!(headers["mcp-session-id"], session_id);
}
let mut result = match method {
Some("server/discover") if modern_with_sharing => {
blocked_tools.as_ref().unwrap().handshakes.fetch_add(1, Ordering::SeqCst);
json!({
"supportedVersions": ["2026-07-28"],
"capabilities": {"tools": {}, "resources": {}},
"_meta": {"io.modelcontextprotocol/serverInfo": {"name": "test", "version": "1"}},
"ttlMs": 0, "cacheScope": "private",
})
}
Some("initialize") if !modern_with_sharing => {
if let Some(blocked) = &blocked_tools {
blocked.handshakes.fetch_add(1, Ordering::SeqCst);
if stalled_initialize_timeout.is_some() {
blocked.started.notify_one();
blocked.release.notified().await;
}
}
json!({
"protocolVersion": "2025-06-18", "capabilities": {"tools": {}, "resources": {}},
"serverInfo": {"name": "test", "version": "1"},
})
}
Some("tools/list") => {
if let Some(blocked) = &blocked_tools {
blocked.inventories.fetch_add(1, Ordering::SeqCst);
blocked.started.notify_one();
blocked.release.notified().await;
blocked.inventory_returned.store(true, Ordering::SeqCst);
if let Some(message) = blocked.inventory_error {
return Json(json!({
"jsonrpc": "2.0", "id": request["id"],
"error": {"code": -32603, "message": message}
})).into_response();
}
}
json!({"tools": []})
}
Some("resources/list") => {
if let Some(blocked) = &blocked_tools {
assert_eq!(blocked.inventory_returned.load(Ordering::SeqCst), !early_resources,
"unexpected catalog request ordering relative to the initial inventory");
}
let mime_type = request["params"]["mimeType"].as_str().unwrap();
let cursor = match mime_type {
"mcp/plugin" => "plugin-page-two",
"mcp/skill" => "skill-page-two",
other => panic!("unexpected resource type: {other}"),
};
if let Some(blocked) = &blocked_tools && blocked.held_mime_type == Some(mime_type) {
blocked.resource_started.notify_one();
blocked.release_resource.notified().await;
}
json!({"resources": [], "nextCursor": cursor})
}
Some("notifications/initialized") => {
return StatusCode::ACCEPTED.into_response();
}
@@ -112,8 +209,14 @@ impl Fixture {
}
method => panic!("unexpected request: {method:?}"),
};
Json(json!({"jsonrpc": "2.0", "id": request["id"], "result": result}))
.into_response()
if modern_with_sharing {
result["resultType"] = "complete".into();
}
let response = Json(json!({"jsonrpc": "2.0", "id": request["id"], "result": result}));
if method == Some("initialize") && let Some(session_id) = session_id {
return ([("mcp-session-id", session_id)], response).into_response();
}
response.into_response()
}
},
));
@@ -131,6 +234,20 @@ impl Fixture {
let config = ConfigBuilder::default()
.codex_home(home.path().into())
.fallback_cwd(Some(home.path().into()))
.cli_overrides(vec![
(
"features.apps_early_resource_discovery".into(),
early_resources.into(),
),
(
"features.mcp_connection_sharing".into(),
modern_with_sharing.into(),
),
(
"features.codex_apps_mcp_2026_07_28".into(),
modern_with_sharing.into(),
),
])
.build()
.await?;
let plugins = plugins_manager_for_config(&config, Arc::clone(&auth));
@@ -146,6 +263,9 @@ impl Fixture {
panic!("HTTP server");
};
*bearer_token_env_var = None;
if let Some(timeout) = stalled_initialize_timeout {
server_config.startup_timeout_sec = Some(timeout);
}
let mut catalog = ResolvedMcpCatalog::builder();
catalog.register(McpServerRegistration::from_hosted_apps(
"test",
@@ -154,6 +274,7 @@ impl Fixture {
));
config.mcp_server_catalog = catalog.build();
let config = Arc::new(config);
let startup_cancellation_token = CancellationToken::new();
let runtime_input = || McpRuntimeInput {
network_policy_metadata: None,
startup_policy: McpStartupPolicy::Eager,
@@ -166,11 +287,12 @@ impl Fixture {
)]),
submit_id: "test".into(),
tx_event: None,
startup_cancellation_token: CancellationToken::new(),
startup_cancellation_token: startup_cancellation_token.clone(),
runtime_context: McpRuntimeContext::new(
Arc::new(environment_manager_without_environments()),
home.path().into(),
),
)
.with_connection_pool(Default::default()),
codex_apps_tools_cache: Default::default(),
tool_catalog_cache: Default::default(),
codex_apps_tools_cache_key: codex_mcp::codex_apps_tools_cache_key(Some(&auth_value)),
@@ -182,11 +304,6 @@ impl Fixture {
};
let runtime = Arc::new(McpRuntime::new(runtime_input()).await);
let runtime_input = runtime_input();
assert!(
runtime
.latest_wait_for_server_ready(CODEX_APPS_MCP_SERVER_NAME, WAIT)
.await
);
Ok(Self {
_home: home,
_server: tokio_util::task::AbortOnDropHandle::new(server),
@@ -198,6 +315,298 @@ impl Fixture {
}
}
#[tokio::test]
async fn apps_resource_discovery_during_initial_tools_list() -> Result<()> {
for (early_resources, modern_with_sharing) in [(false, false), (true, false), (true, true)] {
let blocked = Arc::new(BlockedTools {
modern_with_sharing,
..Default::default()
});
let fixture = Fixture::start(Some(Arc::clone(&blocked)), early_resources).await?;
timeout(WAIT, blocked.started.notified())
.await
.context("initial tools/list did not start")?;
let reader = McpResourceClient::new(Arc::clone(&fixture.runtime));
let mut page = Box::pin(
reader.list_codex_apps_resources(CodexAppsResourceListParams {
mime_type: "mcp/plugin".into(),
cursor: None,
}),
);
if early_resources {
assert_eq!(
timeout(WAIT, &mut page)
.await
.context("gated catalog read waited for the blocked inventory")??
.next_cursor
.as_deref(),
Some("plugin-page-two")
);
} else {
assert!(futures::poll!(&mut page).is_pending());
// A read already using the early client would now fail its generation check.
// A read waiting for startup can still fetch the page after inventory returns.
fixture.runtime.invalidate_resource_caches();
}
let mut ready = Box::pin(
fixture
.runtime
.latest_wait_for_server_ready(CODEX_APPS_MCP_SERVER_NAME, WAIT),
);
assert!(futures::poll!(&mut ready).is_pending());
blocked.release.notify_one();
assert!(ready.await);
if !early_resources {
assert_eq!(
timeout(WAIT, page).await??.next_cursor.as_deref(),
Some("plugin-page-two")
);
}
assert_eq!(blocked.handshakes.load(Ordering::SeqCst), 1);
assert_eq!(blocked.inventories.load(Ordering::SeqCst), 1);
}
Ok(())
}
#[tokio::test]
async fn apps_resource_discovery_times_out_during_stalled_initialization() -> Result<()> {
let startup_timeout = Duration::from_secs(/*secs*/ 1);
let blocked = Arc::new(BlockedTools {
stalled_initialize_timeout: Some(startup_timeout),
..Default::default()
});
let fixture = Fixture::start(Some(Arc::clone(&blocked)), /*early_resources*/ true).await?;
let reader = McpResourceClient::new(Arc::clone(&fixture.runtime));
let mut page = Box::pin(
reader.list_codex_apps_resources(CodexAppsResourceListParams {
mime_type: "mcp/plugin".into(),
cursor: None,
}),
);
assert!(futures::poll!(&mut page).is_pending());
timeout(WAIT, blocked.started.notified())
.await
.context("initialize did not reach the server")?;
let error = timeout(WAIT, page)
.await
.context("reader outlived its startup deadline")?
.unwrap_err();
// The handshake and early borrow each enforce the configured startup timeout.
let message = format!("{error:#}");
assert!(message.contains("timed out"), "{message}");
assert_eq!(blocked.inventories.load(Ordering::SeqCst), 0);
blocked.release.notify_one();
Ok(())
}
#[tokio::test]
async fn apps_resource_discovery_multiplexes_catalogs_while_tools_list_is_pending() -> Result<()> {
let blocked = Arc::new(BlockedTools {
held_mime_type: Some("mcp/plugin"),
session_id: Some("apps-session"),
..Default::default()
});
let fixture = Fixture::start(Some(Arc::clone(&blocked)), /*early_resources*/ true).await?;
timeout(WAIT, blocked.started.notified()).await?;
let reader = McpResourceClient::new(Arc::clone(&fixture.runtime));
let plugin_reader = reader.clone();
let plugin = tokio::spawn(async move {
plugin_reader
.list_codex_apps_resources(CodexAppsResourceListParams {
mime_type: "mcp/plugin".into(),
cursor: None,
})
.await
});
timeout(WAIT, blocked.resource_started.notified())
.await
.context("plugin request did not reach the server while tools/list was pending")?;
let skill = timeout(
WAIT,
reader.list_codex_apps_resources(CodexAppsResourceListParams {
mime_type: "mcp/skill".into(),
cursor: None,
}),
)
.await
.context("skill request queued behind the pending plugin or inventory request")??;
assert_eq!(skill.next_cursor.as_deref(), Some("skill-page-two"));
assert!(!plugin.is_finished());
blocked.release_resource.notify_one();
let plugin = timeout(WAIT, plugin).await???;
assert_eq!(plugin.next_cursor.as_deref(), Some("plugin-page-two"));
let mut ready = Box::pin(
fixture
.runtime
.latest_wait_for_server_ready(CODEX_APPS_MCP_SERVER_NAME, WAIT),
);
assert!(futures::poll!(&mut ready).is_pending());
blocked.release.notify_one();
assert!(ready.await);
assert_eq!(blocked.handshakes.load(Ordering::SeqCst), 1);
assert_eq!(blocked.inventories.load(Ordering::SeqCst), 1);
Ok(())
}
#[tokio::test]
async fn apps_resource_discovery_propagates_initial_tools_list_failure() -> Result<()> {
const INVENTORY_ERROR: &str = "initial inventory unavailable";
let blocked = Arc::new(BlockedTools {
held_mime_type: Some("mcp/plugin"),
inventory_error: Some(INVENTORY_ERROR),
..Default::default()
});
let fixture = Fixture::start(Some(Arc::clone(&blocked)), /*early_resources*/ true).await?;
timeout(WAIT, blocked.started.notified()).await?;
let reader = McpResourceClient::new(Arc::clone(&fixture.runtime));
let mut page = Box::pin(
reader.list_codex_apps_resources(CodexAppsResourceListParams {
mime_type: "mcp/plugin".into(),
cursor: None,
}),
);
tokio::select! {
result = &mut page => panic!("catalog returned before inventory failed: {result:?}"),
result = timeout(WAIT, blocked.resource_started.notified()) => result?,
}
blocked.release.notify_one();
let error = timeout(WAIT, page)
.await
.context("catalog reader kept waiting after tools/list failed")?
.unwrap_err()
.to_string();
assert!(error.starts_with("MCP startup failed:"), "{error}");
assert!(error.contains(INVENTORY_ERROR), "{error}");
assert!(
!fixture
.runtime_input
.startup_cancellation_token
.is_cancelled()
);
blocked.release_resource.notify_one();
Ok(())
}
#[tokio::test]
async fn apps_resource_discovery_observes_cancellation_after_inventory_finishes() -> Result<()> {
let blocked = Arc::new(BlockedTools {
held_mime_type: Some("mcp/plugin"),
..Default::default()
});
let fixture = Fixture::start(Some(Arc::clone(&blocked)), /*early_resources*/ true).await?;
timeout(WAIT, blocked.started.notified()).await?;
let reader = McpResourceClient::new(Arc::clone(&fixture.runtime));
let mut page = Box::pin(
reader.list_codex_apps_resources(CodexAppsResourceListParams {
mime_type: "mcp/plugin".into(),
cursor: None,
}),
);
tokio::select! {
result = &mut page => panic!("catalog returned before release: {result:?}"),
result = timeout(WAIT, blocked.resource_started.notified()) => result?,
}
blocked.release.notify_one();
assert!(
fixture
.runtime
.latest_wait_for_server_ready(CODEX_APPS_MCP_SERVER_NAME, WAIT)
.await
);
assert!(futures::poll!(&mut page).is_pending());
fixture.runtime_input.startup_cancellation_token.cancel();
let error = timeout(WAIT, page)
.await
.context("catalog ignored cancellation after startup")?
.unwrap_err();
assert!(error.to_string().to_lowercase().contains("cancel"));
blocked.release_resource.notify_one();
Ok(())
}
#[tokio::test]
async fn apps_resource_discovery_discards_invalidated_page() -> Result<()> {
for cancel_startup in [false, true] {
let blocked = Arc::new(BlockedTools {
held_mime_type: Some("mcp/skill"),
..Default::default()
});
let fixture = Fixture::start(Some(Arc::clone(&blocked)), /*early_resources*/ true).await?;
timeout(WAIT, blocked.started.notified()).await?;
let reader = McpResourceClient::new(Arc::clone(&fixture.runtime));
let page = tokio::spawn(async move {
reader
.list_codex_apps_resources(CodexAppsResourceListParams {
mime_type: "mcp/skill".into(),
cursor: None,
})
.await
});
timeout(WAIT, blocked.resource_started.notified()).await?;
let expected = if cancel_startup {
fixture.runtime.cancel_startup();
"cancel"
} else {
fixture.runtime.invalidate_resource_caches();
blocked.release_resource.notify_one();
"changed during catalog discovery"
};
let error = timeout(WAIT, page).await??.unwrap_err();
assert!(error.to_string().to_lowercase().contains(expected));
blocked.release.notify_one();
blocked.release_resource.notify_one();
}
Ok(())
}
#[tokio::test]
async fn apps_resource_discovery_discards_page_after_account_switch() -> Result<()> {
let blocked = Arc::new(BlockedTools {
held_mime_type: Some("mcp/skill"),
..Default::default()
});
let mut fixture = Fixture::start(Some(Arc::clone(&blocked)), /*early_resources*/ true).await?;
timeout(WAIT, blocked.started.notified()).await?;
let reader = McpResourceClient::new(Arc::clone(&fixture.runtime));
let mut page = Box::pin(
reader.list_codex_apps_resources(CodexAppsResourceListParams {
mime_type: "mcp/skill".into(),
cursor: None,
}),
);
tokio::select! {
result = &mut page => panic!("catalog returned before account switch: {result:?}"),
result = timeout(WAIT, blocked.resource_started.notified()) => result?,
}
let previous_auth = reader.auth_cache_key_for_server(CODEX_APPS_MCP_SERVER_NAME);
let next_auth = CodexAuth::from_external_chatgpt_tokens(
"header.e30.second",
"second-account",
/*chatgpt_plan_type*/ None,
)?;
fixture
.auth
.set_external_auth(Arc::new(StaticAuth(next_auth.clone())))
.await?;
fixture.runtime_input.codex_apps_tools_cache_key =
codex_mcp::codex_apps_tools_cache_key(Some(&next_auth));
fixture.runtime_input.auth = Some(next_auth);
timeout(WAIT, fixture.runtime.replace(fixture.runtime_input)).await?;
assert!(previous_auth != reader.auth_cache_key_for_server(CODEX_APPS_MCP_SERVER_NAME));
blocked.release_resource.notify_one();
let error = timeout(WAIT, page).await?.unwrap_err();
assert_eq!(
error.to_string(),
"Apps connection changed during catalog discovery"
);
Ok(())
}
pub(super) struct StaticAuth(pub CodexAuth);
impl ExternalAuth for StaticAuth {
codex-rs/features/src/lib.rs
+8
View File
@@ -279,6 +279,8 @@ pub enum Feature {
McpConnectionSharing,
/// Enable MCP protocol version 2026-07-28 for the host-owned Codex Apps server.
CodexAppsMcp20260728,
/// Let host-owned Apps catalog reads overlap the initial tool inventory.
AppsEarlyResourceDiscovery,
/// Let RMCP coordinate OAuth refresh through the configured credential store.
McpOAuthRefreshCoordination,
/// Enable enterprise refresh-token authorization for configured MCP resources.
@@ -1687,6 +1689,12 @@ pub const FEATURES: &[FeatureSpec] = &[
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::AppsEarlyResourceDiscovery,
key: "apps_early_resource_discovery",
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::CodexAppsMcp20260728,
key: "codex_apps_mcp_2026_07_28",