Reuse stable MCP bindings across sampling steps (#37101)

## What changed

- Cache an `McpBinding` on each published MCP runtime once server startup has
  settled and the tool catalog revision is stable.
- Reuse the cached binding while the catalog revision remains unchanged, and
  rebuild it after a runtime publish or catalog change.
- Treat terminal failures from optional servers as stable, while continuing to
  wait for required servers and reconnecting Codex Apps servers.

## Testing

- Cover binding reuse within one published runtime and invalidation after a
  republish.
- Cover stability checks for pending, failed optional, required, and
  reconnected servers.

GitOrigin-RevId: 783a6b04897f6c484a6d56fd9cee542814041759
This commit is contained in:
jif authored and copyberry committed 2026-08-05 14:04:39 +00:00
1 parent 5c44f11064
commit 952e87d3f2
6 files changed
+174 -14

No files matched your search

+3 -3
View File
@@ -1,4 +1,4 @@
//! Immutable MCP state bound to one model sampling request. //! Immutable MCP catalog and execution handles.
use std::collections::HashMap; use std::collections::HashMap;
use std::fmt; use std::fmt;
@@ -26,7 +26,7 @@ use crate::rmcp_client::ManagedClient;
use crate::server::McpServerMetadata; use crate::server::McpServerMetadata;
use crate::tools::ToolInfo; use crate::tools::ToolInfo;
/// The exact tool catalog and execution handles for one model sampling request. /// The exact tool catalog and execution handles shared by compatible sampling steps.
pub struct McpBinding { pub struct McpBinding {
connections: Arc<McpConnectionSet>, connections: Arc<McpConnectionSet>,
clients: Arc<McpBindingClients>, clients: Arc<McpBindingClients>,
@@ -75,7 +75,7 @@ impl McpBinding {
self.plugins_available self.plugins_available
} }
/// Returns the frozen catalog advertised for this sampling request. /// Returns the frozen catalog captured for this binding.
pub fn tools(&self) -> &[ToolInfo] { pub fn tools(&self) -> &[ToolInfo] {
&self.tools &self.tools
} }
+1 -1
View File
@@ -598,7 +598,7 @@ impl McpConnectionSet {
let Some(view) = self.servers.get(server_name) else { let Some(view) = self.servers.get(server_name) else {
return false; return false;
}; };
view.connection.client().await.is_ok() view.connection.client.ready_transport().is_some() || view.connection.client().await.is_ok()
} }
/// Stop all MCP clients owned by this manager and terminate stdio server processes. /// Stop all MCP clients owned by this manager and terminate stdio server processes.
@@ -58,6 +58,32 @@ pub fn tool_is_model_visible(tool: &ToolInfo) -> bool {
} }
impl McpConnectionSet { impl McpConnectionSet {
pub(crate) async fn stable_catalog_revision(&self) -> Option<u64> {
for (server_name, view) in &self.servers {
if !view
.connection
.client
.startup_complete
.load(Ordering::Acquire)
{
return None;
}
let Some(client) = view.connection.client.ready_transport() else {
if !view.connection.client.is_codex_apps_mcp_server
&& self.required_servers.binary_search(server_name).is_err()
&& matches!(view.connection.client.client.peek(), Some(Err(_)))
{
continue;
}
return None;
};
if client.is_closed().await {
return None;
}
}
Some(*self.tool_catalog_revision.read().await)
}
/// Returns all tools with model-visible names normalized. /// Returns all tools with model-visible names normalized.
#[instrument(level = "trace", skip_all, fields(mcp_server_count = self.servers.len()))] #[instrument(level = "trace", skip_all, fields(mcp_server_count = self.servers.len()))]
pub async fn list_all_tools(&self) -> Vec<ToolInfo> { pub async fn list_all_tools(&self) -> Vec<ToolInfo> {
@@ -1911,6 +1911,7 @@ async fn capture_binding_skips_pending_optional_servers_after_one_shared_startup
} }
let manager = Arc::new(manager); let manager = Arc::new(manager);
assert_eq!(manager.stable_catalog_revision().await, None);
let binding = tokio::time::timeout(Duration::from_millis(1500), capture_binding(&manager)) let binding = tokio::time::timeout(Duration::from_millis(1500), capture_binding(&manager))
.await .await
.expect("all optional servers should share a single startup grace"); .expect("all optional servers should share a single startup grace");
@@ -1953,6 +1954,39 @@ async fn capture_binding_skips_pending_optional_servers_after_one_shared_startup
assert!(binding.is_err(), "explicitly requested servers must wait"); assert!(binding.is_err(), "explicitly requested servers must wait");
} }
#[tokio::test]
async fn stable_catalog_revision_ignores_terminal_optional_server_failures() {
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
let permission_profile = Constrained::allow_any(PermissionProfile::default());
let mut manager = McpConnectionSet::new_uninitialized(
&approval_policy,
&permission_profile,
/*prefix_mcp_tool_names*/ true,
);
let ready = create_ready_async_managed_client(vec![create_test_tool("ready", "echo")]).await;
assert!(ready.client().await.is_ok());
let mut failed = ready.clone();
manager.insert_test_client("ready", ready);
failed.client = futures::future::ready::<Result<ManagedClient, StartupOutcomeError>>(Err(
StartupOutcomeError::Failed {
error: "optional startup failed".to_string(),
is_authentication_required: false,
},
))
.boxed()
.shared();
assert!(failed.client().await.is_err());
manager.insert_test_client("failed", failed);
assert_eq!(manager.stable_catalog_revision().await, Some(0));
manager.required_servers.push("failed".to_string());
assert_eq!(manager.stable_catalog_revision().await, None);
manager.required_servers.clear();
let binding = capture_binding(&Arc::new(manager)).await;
assert!(binding.prepare_call("ready", "echo").is_some());
}
#[tokio::test(start_paused = true)] #[tokio::test(start_paused = true)]
async fn capture_binding_shares_optional_startup_grace_across_connection_sets() { async fn capture_binding_shares_optional_startup_grace_across_connection_sets() {
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest); let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
@@ -2519,6 +2553,7 @@ async fn list_all_tools_reconnects_failed_codex_apps_startup_and_reuses_client()
}; };
let manager = Arc::new(manager); let manager = Arc::new(manager);
assert_eq!(manager.stable_catalog_revision().await, None);
let reconnect_finished_wait = reconnect_finished.notified(); let reconnect_finished_wait = reconnect_finished.notified();
let tools = manager.list_all_tools().await; let tools = manager.list_all_tools().await;
assert!(tools.is_empty()); assert!(tools.is_empty());
@@ -2533,6 +2568,7 @@ async fn list_all_tools_reconnects_failed_codex_apps_startup_and_reuses_client()
vec!["drive_search"] vec!["drive_search"]
); );
assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1); assert_eq!(attempts.load(std::sync::atomic::Ordering::SeqCst), 1);
assert_eq!(manager.stable_catalog_revision().await, Some(0));
let step = capture_binding(&manager).await; let step = capture_binding(&manager).await;
let prepared = step let prepared = step
+18
View File
@@ -492,6 +492,24 @@ impl AsyncManagedClient {
self.client.clone().await self.client.clone().await
} }
pub(crate) fn ready_transport(&self) -> Option<Arc<RmcpClient>> {
let recovered = self.startup_reconnect.as_ref().and_then(|reconnect| {
reconnect
.state
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.current_client
.as_ref()
.map(|client| Arc::clone(&client.client))
});
recovered.or_else(|| {
self.client
.peek()
.and_then(|result| result.as_ref().ok())
.map(|client| Arc::clone(&client.client))
})
}
pub(crate) async fn reconnect_failed_startup(&self) { pub(crate) async fn reconnect_failed_startup(&self) {
let Some(startup_reconnect) = self.startup_reconnect.as_ref() else { let Some(startup_reconnect) = self.startup_reconnect.as_ref() else {
return; return;
+90 -10
View File
@@ -8,6 +8,7 @@
use std::collections::HashMap; use std::collections::HashMap;
use std::path::PathBuf; use std::path::PathBuf;
use std::sync::Arc; use std::sync::Arc;
use std::sync::Mutex;
use std::sync::atomic::AtomicBool; use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering; use std::sync::atomic::Ordering;
use std::time::Duration; use std::time::Duration;
@@ -84,6 +85,12 @@ struct PublishedMcpRuntime {
auth_token: Option<String>, auth_token: Option<String>,
plugins_available: bool, plugins_available: bool,
ready_selected_capability_roots: Vec<SelectedCapabilityRoot>, ready_selected_capability_roots: Vec<SelectedCapabilityRoot>,
cached_binding: Mutex<Option<CachedMcpBinding>>,
}
struct CachedMcpBinding {
catalog_revision: u64,
binding: Arc<McpBinding>,
} }
struct McpReconnectGuard<'a> { struct McpReconnectGuard<'a> {
@@ -148,6 +155,7 @@ impl McpRuntime {
auth_token: None, auth_token: None,
plugins_available: false, plugins_available: false,
ready_selected_capability_roots: Vec::new(), ready_selected_capability_roots: Vec::new(),
cached_binding: Mutex::new(None),
}), }),
reconnect_pending: AtomicBool::new(false), reconnect_pending: AtomicBool::new(false),
elicitation_router: ElicitationRequestRouter::default(), elicitation_router: ElicitationRequestRouter::default(),
@@ -204,6 +212,7 @@ impl McpRuntime {
auth_token, auth_token,
plugins_available, plugins_available,
ready_selected_capability_roots, ready_selected_capability_roots,
cached_binding: Mutex::new(None),
})); }));
let _ = publish.send(true); let _ = publish.send(true);
} }
@@ -223,14 +232,51 @@ impl McpRuntime {
&self, &self,
required_servers: &[String], required_servers: &[String],
) -> Option<Arc<McpBinding>> { ) -> Option<Arc<McpBinding>> {
let current = self.current.load_full(); Self::binding_from_published_runtime(self.current.load_full(), required_servers).await
}
async fn binding_from_published_runtime(
current: Arc<PublishedMcpRuntime>,
required_servers: &[String],
) -> Option<Arc<McpBinding>> {
let config = Arc::clone(current.config.as_ref()?); let config = Arc::clone(current.config.as_ref()?);
Some(Arc::new( let stable_catalog_revision = current.connections.stable_catalog_revision().await;
if let Some(catalog_revision) = stable_catalog_revision {
let cached = current
.cached_binding
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(cached) = cached.as_ref()
&& cached.catalog_revision == catalog_revision
{
return Some(Arc::clone(&cached.binding));
}
}
let binding = Arc::new(
current current
.connections .connections
.capture_binding_with_metadata(config, current.plugins_available, required_servers) .capture_binding_with_metadata(config, current.plugins_available, required_servers)
.await, .await,
)) );
if let Some(catalog_revision) = stable_catalog_revision
&& current.connections.stable_catalog_revision().await == Some(catalog_revision)
{
let mut cached = current
.cached_binding
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(cached) = cached.as_ref()
&& cached.catalog_revision == catalog_revision
{
return Some(Arc::clone(&cached.binding));
}
*cached = Some(CachedMcpBinding {
catalog_revision,
binding: Arc::clone(&binding),
});
}
Some(binding)
} }
/// Returns whether the published snapshot still belongs to the current credentials. /// Returns whether the published snapshot still belongs to the current credentials.
@@ -261,16 +307,11 @@ impl McpRuntime {
/// Captures the current runtime after its selected server has finished startup. /// Captures the current runtime after its selected server has finished startup.
pub async fn current_binding_for_call(&self, server: &str) -> Option<Arc<McpBinding>> { pub async fn current_binding_for_call(&self, server: &str) -> Option<Arc<McpBinding>> {
let current = self.current.load_full(); let current = self.current.load_full();
let config = Arc::clone(current.config.as_ref()?); current.config.as_ref()?;
if !current.connections.wait_for_server_startup(server).await { if !current.connections.wait_for_server_startup(server).await {
return None; return None;
} }
Some(Arc::new( Self::binding_from_published_runtime(current, /*required_servers*/ &[]).await
current
.connections
.capture_binding_with_metadata(config, current.plugins_available, &[])
.await,
))
} }
/// Returns the latest published configuration without waiting for clients. /// Returns the latest published configuration without waiting for clients.
@@ -513,6 +554,45 @@ mod tests {
assert!(!gate.wait().await); assert!(!gate.wait().await);
} }
#[tokio::test]
async fn cached_bindings_are_scoped_to_the_published_runtime() {
let published = Arc::new(PublishedMcpRuntime {
connections: Arc::new(McpConnectionSet::empty(/*prefix_mcp_tool_names*/ true)),
config: Some(Arc::new(crate::mcp::tests::test_mcp_config(
std::env::temp_dir(),
))),
auth: None,
auth_token: None,
plugins_available: false,
ready_selected_capability_roots: Vec::new(),
cached_binding: Mutex::new(None),
});
let first = McpRuntime::binding_from_published_runtime(
Arc::clone(&published),
/*required_servers*/ &[],
)
.await
.expect("first binding");
let repeated = McpRuntime::binding_from_published_runtime(
Arc::clone(&published),
/*required_servers*/ &[],
)
.await
.expect("repeated binding");
assert!(Arc::ptr_eq(&first, &repeated));
let previous = Arc::into_inner(published).expect("published runtime has no other owners");
let republished = Arc::new(PublishedMcpRuntime {
cached_binding: Mutex::new(None),
..previous
});
let refreshed =
McpRuntime::binding_from_published_runtime(republished, /*required_servers*/ &[])
.await
.expect("republished binding");
assert!(!Arc::ptr_eq(&first, &refreshed));
}
fn http_server(environment_id: &str) -> McpServerConfig { fn http_server(environment_id: &str) -> McpServerConfig {
McpServerConfig { McpServerConfig {
auth: Default::default(), auth: Default::default(),