github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 08:16:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

Preserve MCP tool-list metadata in server status (#54686)

Browse Source 
## Why

MCP tool discovery discarded response-level `_meta`, leaving server-status clients unable to inspect metadata associated with the tool catalog.

## What changed

- Preserve opaque `_meta` from the first `tools/list` page through catalog refreshes and caches.
- Expose it as `toolListMeta` in `McpServerStatus`, with `null` when unavailable, and update generated schemas and bindings.
- Scope cached Apps catalog metadata to discovery settings, including client capabilities and extensions, while retaining tools from older caches without metadata.
- Restrict server capabilities in status snapshots to the selected servers.

## Testing

Add regression coverage for listing legacy cached Apps tools during failed startup, and update protocol serialization expectations for nullable `toolListMeta`.

GitOrigin-RevId: 62335e2fd8c482067a7a78a7612b58c3eaa9fb55
This commit is contained in:
victor-openai
2026-10-16 19:26:46 +00:00
committed by copyberry
parent 440312f113
commit 0e67502123
27 changed files with 428 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
Generated
+3
View File
@@ -15157,6 +15157,9 @@
}
]
},
"toolListMeta": {
"description": "Opaque `_meta` from the first page of the corresponding MCP tool catalog. Null when unavailable."
},
"tools": {
"additionalProperties": {
"$ref": "#/definitions/v2/Tool"
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json
Generated
+3
View File
@@ -11149,6 +11149,9 @@
}
]
},
"toolListMeta": {
"description": "Opaque `_meta` from the first page of the corresponding MCP tool catalog. Null when unavailable."
},
"tools": {
"additionalProperties": {
"$ref": "#/definitions/Tool"
codex-rs/app-server-protocol/schema/json/v2/ListMcpServerStatusResponse.json
Generated
+3
View File
@@ -121,6 +121,9 @@
}
]
},
"toolListMeta": {
"description": "Opaque `_meta` from the first page of the corresponding MCP tool catalog. Null when unavailable."
},
"tools": {
"additionalProperties": {
"$ref": "#/definitions/Tool"
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server-protocol/schema/typescript/v2/McpServerStatus.ts
Generated
+5 -1
View File
@@ -22,7 +22,11 @@ httpOrigin: string | null, serverInfo: McpServerInfo | null,
/**
* Capabilities advertised by the initialized MCP server; null when unavailable.
*/
serverCapabilities: JsonValue | null, tools: { [key in string]?: Tool },
serverCapabilities: JsonValue | null,
/**
* Opaque `_meta` from the first page of the corresponding MCP tool catalog. Null when unavailable.
*/
toolListMeta: JsonValue | null, tools: { [key in string]?: Tool },
/**
* Tool discovery failed and no catalog was returned.
* Null when a catalog is returned, including cached or empty catalogs.
codex-rs/app-server-protocol/src/protocol/v2/mcp.rs
+2
View File
@@ -87,6 +87,8 @@ pub struct McpServerStatus {
pub server_info: Option<McpServerInfo>,
/// Capabilities advertised by the initialized MCP server; null when unavailable.
pub server_capabilities: Option<serde_json::Value>,
/// Opaque `_meta` from the first page of the corresponding MCP tool catalog. Null when unavailable.
pub tool_list_meta: Option<serde_json::Value>,
pub tools: std::collections::HashMap<String, McpTool>,
/// Tool discovery failed and no catalog was returned.
/// Null when a catalog is returned, including cached or empty catalogs.
codex-rs/app-server-protocol/src/protocol/v2/tests.rs
+5
View File
@@ -2671,6 +2671,7 @@ fn mcp_server_elicitation_response_serializes_nullable_content() {
fn mcp_server_status_serializes_absent_server_info_as_null() {
let response = ListMcpServerStatusResponse {
data: vec![McpServerStatus {
tool_list_meta: None,
server_capabilities: None,
tools_error: None,
name: "not-ready".to_string(),
@@ -2696,6 +2697,7 @@ fn mcp_server_status_serializes_absent_server_info_as_null() {
"httpOrigin": null,
"serverInfo": null,
"serverCapabilities": null,
"toolListMeta": null,
"tools": {},
"toolsError": null,
"resources": [],
@@ -2722,6 +2724,7 @@ fn mcp_server_status_accepts_older_inventory_without_runtime_status() {
assert_eq!(
status,
McpServerStatus {
tool_list_meta: None,
server_capabilities: None,
tools_error: None,
name: "older-server".to_string(),
@@ -2796,6 +2799,7 @@ fn mcp_server_status_updated_serializes_failure_reason() {
fn mcp_server_status_serializes_absent_server_info_metadata_as_null() {
let response = ListMcpServerStatusResponse {
data: vec![McpServerStatus {
tool_list_meta: None,
server_capabilities: None,
tools_error: None,
name: "initialized".to_string(),
@@ -2827,6 +2831,7 @@ fn mcp_server_status_serializes_absent_server_info_metadata_as_null() {
"pluginId": "lookup@test",
"httpOrigin": null,
"serverCapabilities": null,
"toolListMeta": null,
"serverInfo": {
"name": "lookup-server",
"title": null,
codex-rs/app-server/src/request_processors/mcp_processor.rs
+2
View File
@@ -452,6 +452,7 @@ impl McpRequestProcessor {
let McpServerStatusSnapshot {
mut server_infos,
mut server_capabilities,
mut tool_list_meta,
mut tools_by_server,
mut tools_errors,
mut resources,
@@ -525,6 +526,7 @@ impl McpRequestProcessor {
}),
server_info: server_infos.remove(name),
server_capabilities: server_capabilities.remove(name),
tool_list_meta: tool_list_meta.remove(name).map(Arc::unwrap_or_clone),
tools: tools_by_server.remove(name).unwrap_or_default(),
tools_error: tools_errors.remove(name),
resources: resources.remove(name).unwrap_or_default(),
codex-rs/codex-mcp/src/binding_tests.rs
+7 -6
View File
@@ -80,6 +80,7 @@ async fn test_step(
);
let tool_catalog = Arc::new(ClientToolCatalog::new(
vec![tool.clone()],
/*meta*/ None,
/*updates*/ None,
));
let managed_client = Arc::new(ManagedClient {
@@ -313,8 +314,8 @@ async fn prepared_call_is_rejected_after_catalog_refresh() {
step.tool_catalog
.refresh(
|| async { Ok((step.step.tools().to_vec(), ())) },
|_, ()| {},
|| async { Ok((step.step.tools().to_vec(), None, ())) },
|_, _, ()| {},
)
.await
.expect("refresh tool catalog");
@@ -347,8 +348,8 @@ async fn stale_prepared_call_does_not_run_preparation() {
.expect("step should prepare the advertised tool");
step.tool_catalog
.refresh(
|| async { Ok((step.step.tools().to_vec(), ())) },
|_, ()| {},
|| async { Ok((step.step.tools().to_vec(), None, ())) },
|_, _, ()| {},
)
.await
.expect("refresh tool catalog");
@@ -394,8 +395,8 @@ async fn preparation_holds_catalog_authority_until_it_finishes() {
preparation_started.notified().await;
let refresh = step.tool_catalog.refresh(
|| async { Ok((step.step.tools().to_vec(), ())) },
|_, ()| {},
|| async { Ok((step.step.tools().to_vec(), None, ())) },
|_, _, ()| {},
);
tokio::pin!(refresh);
assert!(
codex-rs/codex-mcp/src/client_tool_catalog.rs
+36 -15
View File
@@ -59,7 +59,10 @@ struct CatalogState {
/// Ordinary MCP clients own their tools; Apps clients observe the shared provider.
enum CatalogSource {
Local(Arc<[ToolInfo]>),
Local {
tools: Arc<[ToolInfo]>,
meta: Option<Arc<serde_json::Value>>,
},
Live {
updates: ToolCatalogUpdates,
tools_version: u64,
@@ -69,12 +72,14 @@ enum CatalogSource {
pub(crate) struct ToolCatalogSnapshot {
pub(crate) revision: u64,
pub(crate) tools: Arc<[ToolInfo]>,
pub(crate) meta: Option<Arc<serde_json::Value>>,
}
impl ClientToolCatalog {
/// Live clients read the provider; callers publish startup tools before subscribing.
pub(crate) fn new(
tools: impl Into<Arc<[ToolInfo]>>,
meta: Option<Arc<serde_json::Value>>,
updates: Option<ToolCatalogUpdates>,
) -> Self {
let source = match updates {
@@ -88,7 +93,10 @@ impl ClientToolCatalog {
tools_version,
}
}
None => CatalogSource::Local(tools.into()),
None => CatalogSource::Local {
tools: tools.into(),
meta,
},
};
Self {
current: RwLock::new(CatalogState {
@@ -112,8 +120,8 @@ impl ClientToolCatalog {
loop {
{
let current = self.current.read().await;
let tools = match &current.source {
CatalogSource::Local(tools) => Some(Arc::clone(tools)),
let catalog = match &current.source {
CatalogSource::Local { tools, meta } => Some((Arc::clone(tools), meta.clone())),
CatalogSource::Live { updates, .. } => {
// Hold the watch borrow while checking its version so the tools and
// revision come from the same publication.
@@ -121,15 +129,18 @@ impl ClientToolCatalog {
(!updates.has_changed().unwrap_or(false)).then(|| {
snapshot
.as_ref()
.map(|snapshot| snapshot.shared_tools())
.map(|snapshot| {
(snapshot.shared_tools(), snapshot.shared_tool_list_meta())
})
.unwrap_or_default()
})
}
};
if let Some(tools) = tools {
if let Some((tools, meta)) = catalog {
let snapshot = ToolCatalogSnapshot {
revision: current.revision,
tools,
meta,
};
return (current, snapshot);
}
@@ -148,7 +159,7 @@ impl ClientToolCatalog {
*tools_version = version;
changed
}
CatalogSource::Local(_) => false,
CatalogSource::Local { .. } => false,
};
if changed {
current.revision += 1;
@@ -175,7 +186,7 @@ impl ClientToolCatalog {
) -> Result<()>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<Vec<ToolInfo>>>,
Fut: Future<Output = Result<(Vec<ToolInfo>, Option<Arc<serde_json::Value>>)>>,
{
if self.notifications_current(notifications) {
return Ok(());
@@ -199,7 +210,11 @@ impl ClientToolCatalog {
if notifications.current() != revision {
continue;
}
current.source = CatalogSource::Local(tools?.into());
let (tools, meta) = tools?;
current.source = CatalogSource::Local {
tools: tools.into(),
meta,
};
current.revision += 1;
self.listed_notification_revision
.store(revision, Ordering::Release);
@@ -225,16 +240,22 @@ impl ClientToolCatalog {
pub(crate) async fn refresh<C, R, F, Fut, P>(&self, fetch: F, publish: P) -> Result<R>
where
F: FnOnce() -> Fut,
Fut: Future<Output = Result<(Vec<ToolInfo>, C)>>,
P: FnOnce(&[ToolInfo], C) -> R,
Fut: Future<Output = Result<(Vec<ToolInfo>, Option<Arc<serde_json::Value>>, C)>>,
P: FnOnce(&[ToolInfo], Option<Arc<serde_json::Value>>, C) -> R,
{
let _refresh = self.refresh_lock.lock().await;
let (tools, context) = fetch().await?;
let (tools, meta, context) = fetch().await?;
let _calls = self.calls.write().await;
let mut current = self.current.write().await;
let result = publish(&tools, context);
let result = publish(&tools, meta.clone(), context);
match &mut current.source {
CatalogSource::Local(current_tools) => *current_tools = tools.into(),
CatalogSource::Local {
tools: current_tools,
meta: current_meta,
} => {
*current_tools = tools.into();
*current_meta = meta;
}
CatalogSource::Live {
updates,
tools_version,
@@ -274,7 +295,7 @@ impl ClientToolCatalog {
return None;
}
let current = match current.source {
CatalogSource::Local(_) => {
CatalogSource::Local { .. } => {
drop(current);
None
}
codex-rs/codex-mcp/src/connection_manager.rs
+17 -8
View File
@@ -239,23 +239,24 @@ impl McpServerView {
}
}
async fn listed_tools(
async fn listed_catalog(
&self,
server_name: &str,
tool_plugin_context: &ToolPluginContext,
) -> Result<Vec<ToolInfo>, StartupOutcomeError> {
let tools = if let Some(client) = self.connection.client.ready_client() {
) -> Result<(Vec<ToolInfo>, Option<Arc<serde_json::Value>>), StartupOutcomeError> {
let (tools, meta) = if let Some(client) = self.connection.client.ready_client() {
self.refresh_notified_catalog(server_name, &client).await;
client.listed_tools().await
client.listed_catalog().await
} else {
self.connection.client.listed_tools().await?
self.connection.client.listed_catalog().await?
};
let tools = filter_tools(tools, &self.tool_filter);
Ok(if self.connection.client.is_codex_apps_mcp_server {
let tools = if self.connection.client.is_codex_apps_mcp_server {
prepare_codex_apps_tools_for_model(tools, tool_plugin_context)
} else {
prepare_regular_mcp_tools_for_model(tools, tool_plugin_context)
})
};
Ok((tools, meta))
}
}
@@ -448,6 +449,10 @@ impl McpConnectionSet {
&configured_config.auth,
protocol_mode.preferred_protocol_version().as_str(),
catalog_item_limit,
&client_elicitation_capability,
client_mcp_extensions
.iter()
.collect::<std::collections::BTreeMap<_, _>>(),
]);
scope.sort_all_objects();
codex_apps_tools_cache
@@ -1097,9 +1102,13 @@ impl McpConnectionSet {
}
/// Capabilities belong to the initialized connection, never a shared tool cache.
pub(crate) fn list_available_server_capabilities(&self) -> HashMap<String, serde_json::Value> {
pub(crate) fn list_available_server_capabilities(
&self,
include_server: impl Fn(&str) -> bool,
) -> HashMap<String, serde_json::Value> {
self.servers
.iter()
.filter(|(name, _)| include_server(name))
.filter_map(|(name, view)| {
view.connection
.client
codex-rs/codex-mcp/src/connection_manager/tool_catalog.rs
+34 -12
View File
@@ -202,6 +202,19 @@ impl McpConnectionSet {
&self,
include_server: impl Fn(&str) -> bool,
) -> (Vec<ToolInfo>, HashMap<String, String>) {
let (tools, errors, _) = self.list_catalogs_with_errors(include_server).await;
(tools, errors)
}
pub(crate) async fn list_catalogs_with_errors(
&self,
include_server: impl Fn(&str) -> bool,
) -> (
Vec<ToolInfo>,
HashMap<String, String>,
HashMap<String, Arc<serde_json::Value>>,
) {
let mut metadata = HashMap::new();
let mut tools = Vec::new();
let mut errors = HashMap::new();
let mut available_server_count = 0;
@@ -219,7 +232,7 @@ impl McpConnectionSet {
.startup_complete
.load(Ordering::Acquire);
let server_tools = view
.listed_tools(server_name, &self.tool_plugin_context)
.listed_catalog(server_name, &self.tool_plugin_context)
.instrument(trace_span!(
"list_tools_for_server",
server_name = %server_name,
@@ -228,10 +241,13 @@ impl McpConnectionSet {
))
.await;
let result = match server_tools {
Ok(server_tools) => Ok(server_tools
.into_iter()
.map(|tool| Self::with_server_metadata(tool, &view.metadata))
.collect::<Vec<_>>()),
Ok((server_tools, meta)) => Ok((
server_tools
.into_iter()
.map(|tool| Self::with_server_metadata(tool, &view.metadata))
.collect::<Vec<_>>(),
meta,
)),
Err(error) => {
trace!(
server_name = %server_name,
@@ -248,7 +264,10 @@ impl McpConnectionSet {
.await;
for (server_name, server_tools) in server_results {
match server_tools {
Ok(server_tools) => {
Ok((server_tools, meta)) => {
if let Some(meta) = meta {
metadata.insert(server_name.clone(), meta);
}
available_server_count += 1;
tools.extend(server_tools);
}
@@ -269,7 +288,7 @@ impl McpConnectionSet {
tool_count = tools.len(),
"built MCP tool list"
);
(tools, errors)
(tools, errors, metadata)
}
#[instrument(level = "trace", skip_all)]
@@ -706,7 +725,7 @@ impl McpConnectionSet {
cache_context.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh)
},
);
let client_tools = list_tools_for_client_uncached(
let (client_tools, meta) = list_tools_for_client_uncached(
CODEX_APPS_MCP_SERVER_NAME,
/*is_codex_apps_mcp_server*/ true,
/*codex_apps_refresh_trigger*/ "explicit",
@@ -721,9 +740,9 @@ impl McpConnectionSet {
"failed to refresh tools for MCP server '{CODEX_APPS_MCP_SERVER_NAME}'"
)
})?;
Ok((client_tools, (fetch_ticket, list_start)))
Ok((client_tools, meta, (fetch_ticket, list_start)))
},
|client_tools, (fetch_ticket, list_start)| {
|client_tools, meta, (fetch_ticket, list_start)| {
// Discovery can accept another scope's winner; executable catalogs
// receive only the latest successful fetch from their own scope.
let tools = match (
@@ -731,11 +750,14 @@ impl McpConnectionSet {
fetch_ticket,
) {
(Some(cache_context), Some(fetch_ticket)) => cache_context
.publish_if_newest_accepted(
.publish_runtime_if_newest_accepted(
fetch_ticket,
&managed_client.server_info,
client_tools.to_vec(),
),
meta,
)
.tools()
.to_vec(),
(None, None) => client_tools.to_vec(),
_ => unreachable!("Codex Apps fetch ticket requires cache context"),
};
codex-rs/codex-mcp/src/connection_manager_tests.rs
+96 -27
View File
@@ -217,10 +217,11 @@ fn create_codex_apps_tools_cache_context(
}
fn store_current_tools(cache_context: &ConnectorRuntimeContext<ToolInfo>, tools: Vec<ToolInfo>) {
let _ = cache_context.publish_if_newest_accepted(
let _ = cache_context.publish_runtime_if_newest_accepted(
cache_context.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools,
/*tool_list_meta*/ None,
);
}
@@ -486,7 +487,9 @@ async fn create_test_managed_client(tools: Vec<ToolInfo>) -> ManagedClient {
.expect("create in-process RMCP client"),
),
server_info: create_test_server_info("Ready"),
tool_catalog: Arc::new(ClientToolCatalog::new(tools, /*updates*/ None)),
tool_catalog: Arc::new(ClientToolCatalog::new(
tools, /*meta*/ None, /*updates*/ None,
)),
tool_timeout: None,
server_instructions: None,
server_supports_sandbox_state_meta_capability: false,
@@ -794,7 +797,11 @@ pub(crate) async fn create_test_manager_with_ready_apps_client(
_auth_change_notifications: None,
client,
server_info: create_test_server_info("Codex Apps"),
tool_catalog: Arc::new(ClientToolCatalog::new(vec![tool], /*updates*/ None)),
tool_catalog: Arc::new(ClientToolCatalog::new(
vec![tool],
/*meta*/ None,
/*updates*/ None,
)),
tool_timeout: Some(Duration::from_secs(5)),
server_instructions: None,
server_supports_sandbox_state_meta_capability: false,
@@ -2339,7 +2346,8 @@ async fn capture_binding_preserves_ready_client_catalog_and_shares_prepared_meta
codex_home.path().to_path_buf(),
Some("account-one"),
Some("user-one"),
);
)
.with_live_scope("apps".to_string());
store_current_tools(
&cache_context,
vec![create_test_tool(
@@ -2639,12 +2647,12 @@ async fn tool_catalog_cache_sanitizes_tools_and_tracks_environment_generation()
let first_environment = resolve_environment();
let first_environment_weak = Arc::downgrade(&first_environment);
let first_context = cache_context(&first_environment);
first_context.publish_if_newest(first_context.begin_fetch(), &[]);
first_context.publish_if_newest(first_context.begin_fetch(), &[], /*meta*/ None);
assert!(!first_context.has_tools());
let mut tool = create_test_tool("docs", "search");
tool.tool.annotations = Some(rmcp::model::ToolAnnotations::new().read_only(true));
first_context.publish_if_newest(first_context.begin_fetch(), &[tool]);
first_context.publish_if_newest(first_context.begin_fetch(), &[tool], /*meta*/ None);
assert_eq!(
first_context.current_tools().expect("cached tools")[0]
.tool
@@ -2670,8 +2678,16 @@ async fn tool_catalog_cache_sanitizes_tools_and_tracks_environment_generation()
let older = first_context.begin_fetch();
let newer = first_context.begin_fetch();
first_context.publish_if_newest(newer, &[create_test_tool("docs", "new")]);
first_context.publish_if_newest(older, &[create_test_tool("docs", "old")]);
first_context.publish_if_newest(
newer,
&[create_test_tool("docs", "new")],
/*meta*/ None,
);
first_context.publish_if_newest(
older,
&[create_test_tool("docs", "old")],
/*meta*/ None,
);
assert_eq!(
first_context.current_tools().expect("cached tools")[0].callable_name,
"new"
@@ -3299,6 +3315,7 @@ async fn capture_binding_shares_optional_startup_grace_across_connection_sets()
cache_context.publish_if_newest(
cache_context.begin_fetch(),
&[create_test_tool("pending", "cached_tool")],
/*meta*/ None,
);
let deadline_after_publication = tokio::time::Instant::now() + Duration::from_secs(1);
assert_eq!(
@@ -3384,6 +3401,7 @@ async fn capture_binding_uses_cache_published_during_optional_startup() {
cache_context.publish_if_newest(
cache_context.begin_fetch(),
&[create_test_tool("pending", "peer_tool")],
/*meta*/ None,
);
tokio::time::advance(Duration::from_secs(/*secs*/ 2)).await;
@@ -3433,6 +3451,7 @@ async fn capture_binding_retains_cached_tools_that_expire_while_waiting_for_anot
cache_context.publish_if_newest(
cache_context.begin_fetch(),
&[create_test_tool("server_a", "cached_tool")],
/*meta*/ None,
);
tokio::time::advance(Duration::from_secs(30 * 60 - 1)).await;
@@ -3498,6 +3517,7 @@ async fn capture_binding_omits_cache_disabled_while_waiting_for_another_server()
cache_context.publish_if_newest(
cache_context.begin_fetch(),
&[create_test_tool("cached", "cached_tool")],
/*meta*/ None,
);
let (mut cached, _started, _release) = create_gated_async_managed_client(
create_test_managed_client(vec![create_test_tool("cached", "live_tool")]).await,
@@ -3837,6 +3857,47 @@ async fn list_all_tools_resolves_server_catalogs_concurrently() {
);
}
#[tokio::test]
async fn catalog_listing_preserves_legacy_apps_tools_during_failed_startup() {
for startup_complete in [false, true] {
let mut manager = create_test_manager_with_failed_apps_startup(
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "drive_search")],
Arc::new(|| futures::future::pending().boxed().shared()),
);
let connection = Arc::get_mut(
&mut manager
.servers
.get_mut(CODEX_APPS_MCP_SERVER_NAME)
.expect("Apps server")
.connection,
)
.expect("unshared connection");
let context = connection
.client
.codex_apps_tools_cache_context
.take()
.expect("Apps cache")
.with_live_scope("endpoint".to_string());
assert!(context.current_catalog_snapshot().is_none());
connection.client.codex_apps_tools_cache_context = Some(context);
connection.client.startup_reconnect = None;
connection
.client
.startup_complete
.store(startup_complete, Ordering::Release);
let (tools, errors, metadata) = manager.list_catalogs_with_errors(|_| true).await;
assert_eq!(
(model_tool_names(&tools), errors, metadata),
(
HashSet::from([ToolName::namespaced("mcp__codex_apps", "drive_search")]),
HashMap::new(),
HashMap::new(),
),
);
}
}
#[tokio::test]
async fn list_all_tools_blocks_while_client_is_pending_without_cached_tools() {
let pending_client = futures::future::pending::<Result<ManagedClient, StartupOutcomeError>>()
@@ -5385,7 +5446,7 @@ async fn apps_catalog_broadcast_preserves_running_calls_and_rejects_stale_calls(
let original = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "original")];
let updated = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "updated")];
store_current_tools(&context, original.clone());
let catalog = ClientToolCatalog::new(original.clone(), context.subscribe());
let catalog = ClientToolCatalog::new(original.clone(), /*meta*/ None, context.subscribe());
let snapshot = catalog.read(Arc::new).await;
store_current_tools(&context, original.clone());
@@ -5417,7 +5478,7 @@ async fn apps_catalog_broadcast_preserves_running_calls_and_rejects_stale_calls(
);
// A client whose startup finishes late adopts the already-published result.
let late = ClientToolCatalog::new(original, context.subscribe());
let late = ClientToolCatalog::new(original, /*meta*/ None, context.subscribe());
assert_eq!(late.read(|catalog| catalog.tools.to_vec()).await, updated);
Ok(())
}
@@ -5436,7 +5497,7 @@ async fn apps_catalog_broadcast_restores_equivalent_calls() -> anyhow::Result<()
create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "second"),
];
store_current_tools(&context, original.clone());
let catalog = ClientToolCatalog::new(original.clone(), context.subscribe());
let catalog = ClientToolCatalog::new(original.clone(), /*meta*/ None, context.subscribe());
let snapshot = catalog.read(Arc::new).await;
let mut changed = original.clone();
@@ -5460,8 +5521,8 @@ async fn apps_catalog_broadcast_restores_equivalent_calls() -> anyhow::Result<()
);
catalog
.refresh(
|| async { Ok((snapshot.tools.to_vec(), ())) },
|tools, ()| store_current_tools(&context, tools.to_vec()),
|| async { Ok((snapshot.tools.to_vec(), None, ())) },
|tools, _, ()| store_current_tools(&context, tools.to_vec()),
)
.await?;
assert_eq!(
@@ -5487,8 +5548,8 @@ fn idle_apps_clients_do_not_retain_replaced_tools() {
let schema = Arc::downgrade(&tool.tool.input_schema);
store_current_tools(&context, vec![tool]);
let clients = [
ClientToolCatalog::new(Vec::new(), context.subscribe()),
ClientToolCatalog::new(Vec::new(), context.subscribe()),
ClientToolCatalog::new(Vec::new(), /*meta*/ None, context.subscribe()),
ClientToolCatalog::new(Vec::new(), /*meta*/ None, context.subscribe()),
];
store_current_tools(
&context,
@@ -5511,20 +5572,24 @@ async fn apps_catalog_broadcast_survives_an_older_local_refresh() -> anyhow::Res
)
.with_live_scope("apps".to_string());
store_current_tools(&context, Vec::new());
let catalog = ClientToolCatalog::new(Vec::new(), context.subscribe());
let catalog = ClientToolCatalog::new(Vec::new(), /*meta*/ None, context.subscribe());
let older_ticket = context.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh);
let newer = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "newer")];
store_current_tools(&context, newer.clone());
assert_eq!(catalog.read(|catalog| catalog.tools.to_vec()).await, newer);
catalog
.refresh(
|| async { Ok((Vec::new(), older_ticket)) },
|tools, ticket| {
context.publish_if_newest_accepted(
ticket,
&create_test_server_info("Apps"),
tools.to_vec(),
)
|| async { Ok((Vec::new(), None, older_ticket)) },
|tools, _, ticket| {
context
.publish_runtime_if_newest_accepted(
ticket,
&create_test_server_info("Apps"),
tools.to_vec(),
/*tool_list_meta*/ None,
)
.tools()
.to_vec()
},
)
.await?;
@@ -5573,7 +5638,7 @@ async fn refreshed_catalog_follows_reused_client_without_mutating_old_bindings()
let startup_tools = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "startup")];
client
.tool_catalog
.refresh(|| async { Ok((startup_tools, ())) }, |_, ()| {})
.refresh(|| async { Ok((startup_tools, None, ())) }, |_, _, ()| {})
.await?;
let old_binding = capture_binding(&previous).await;
let old_tools = serde_json::to_value(old_binding.tools())?;
@@ -5660,7 +5725,7 @@ async fn reconciliation_reuses_connection_without_relisting_regular_tools() -> a
}),
)
.await?;
let initial_tools = list_tools_for_client_uncached(
let (initial_tools, initial_meta) = list_tools_for_client_uncached(
"docs",
/*is_codex_apps_mcp_server*/ false,
/*codex_apps_refresh_trigger*/ "test",
@@ -5674,7 +5739,11 @@ async fn reconciliation_reuses_connection_without_relisting_regular_tools() -> a
_auth_change_notifications: None,
client: Arc::clone(&client),
server_info: create_test_server_info("Mutable tools"),
tool_catalog: Arc::new(ClientToolCatalog::new(initial_tools, /*updates*/ None)),
tool_catalog: Arc::new(ClientToolCatalog::new(
initial_tools,
initial_meta,
/*updates*/ None,
)),
tool_timeout: None,
server_instructions: initialize.instructions,
server_supports_sandbox_state_meta_capability: false,
@@ -5749,7 +5818,7 @@ async fn reconciliation_reuses_connection_without_relisting_regular_tools() -> a
cache.track_tool_list_changes = true;
let mut ticket = cache.begin_fetch();
ticket.tool_list_revision = Some(client.tool_list_revision());
cache.publish_if_newest(ticket, old_step.tools());
cache.publish_if_newest(ticket, old_step.tools(), /*meta*/ None);
assert!(cache.has_tools());
// Reused connections must use the new view's timeout, not their startup timeout.
config.startup_timeout_sec = Some(Duration::from_secs(1));
codex-rs/codex-mcp/src/mcp/mod.rs
+8 -3
View File
@@ -495,6 +495,7 @@ pub async fn read_mcp_resource(
pub struct McpServerStatusSnapshot {
pub server_infos: HashMap<String, McpServerInfo>,
pub server_capabilities: HashMap<String, serde_json::Value>,
pub tool_list_meta: HashMap<String, Arc<serde_json::Value>>,
pub tools_by_server: HashMap<String, HashMap<String, Tool>>,
pub tools_errors: HashMap<String, String>,
pub resources: HashMap<String, Vec<Resource>>,
@@ -518,6 +519,7 @@ pub async fn collect_mcp_server_status_snapshot_with_detail(
return McpServerStatusSnapshot {
server_infos: HashMap::new(),
server_capabilities: HashMap::new(),
tool_list_meta: HashMap::new(),
tools_by_server: HashMap::new(),
tools_errors: HashMap::new(),
resources: HashMap::new(),
@@ -806,13 +808,13 @@ pub(crate) async fn collect_mcp_server_status_snapshot_from_manager(
) -> McpServerStatusSnapshot {
let selected_servers: HashSet<&str> = server_names.iter().map(String::as_str).collect();
let include_server = |name: &str| selected_servers.contains(name);
let ((server_infos, (tools, tools_errors)), resources, resource_templates) = tokio::join!(
let ((server_infos, (tools, tools_errors, tool_list_meta)), resources, resource_templates) = tokio::join!(
async {
let server_infos = mcp_connection_manager
.list_available_server_infos(include_server)
.await;
let tools = mcp_connection_manager
.list_tools_with_errors(include_server)
.list_catalogs_with_errors(include_server)
.await;
(server_infos, tools)
},
@@ -836,6 +838,8 @@ pub(crate) async fn collect_mcp_server_status_snapshot_from_manager(
},
);
let server_capabilities =
mcp_connection_manager.list_available_server_capabilities(include_server);
let mut tools_by_server = HashMap::<String, HashMap<String, Tool>>::new();
for tool_info in tools {
let raw_tool_name = tool_info.tool.name.to_string();
@@ -859,7 +863,8 @@ pub(crate) async fn collect_mcp_server_status_snapshot_from_manager(
}
McpServerStatusSnapshot {
server_capabilities: mcp_connection_manager.list_available_server_capabilities(),
server_capabilities,
tool_list_meta,
server_infos,
tools_by_server,
tools_errors,
codex-rs/codex-mcp/src/rmcp_client.rs
+52 -20
View File
@@ -128,7 +128,7 @@ pub(crate) struct ManagedClient {
}
impl ManagedClient {
pub(crate) async fn listed_tools(&self) -> Vec<ToolInfo> {
pub(crate) async fn listed_catalog(&self) -> (Vec<ToolInfo>, Option<Arc<serde_json::Value>>) {
let total_start = Instant::now();
self.tool_catalog
.read(|catalog| {
@@ -137,17 +137,17 @@ impl ManagedClient {
if catalog.revision == 0
&& let Some(cache_context) = &self.codex_apps_tools_cache_context
{
let tools = cache_context.current_tools();
let tools = cache_context.current_catalog_snapshot();
emit_duration(
MCP_TOOLS_LIST_DURATION_METRIC,
total_start.elapsed(),
&[("cache", if tools.is_some() { "hit" } else { "miss" })],
);
if let Some(tools) = tools {
return tools;
return (tools.tools().to_vec(), tools.shared_tool_list_meta());
}
}
catalog.tools.to_vec()
(catalog.tools.to_vec(), catalog.meta)
})
.await
}
@@ -598,10 +598,6 @@ impl AsyncManagedClient {
.is_some_and(McpToolCatalogCacheContext::has_tools)
}
pub(crate) fn cached_tools(&self) -> Option<Vec<ToolInfo>> {
self.cached_tools_or(/*fallback*/ None)
}
pub(crate) fn cached_tools_or(&self, fallback: Option<Vec<ToolInfo>>) -> Option<Vec<ToolInfo>> {
self.codex_apps_tools_cache_context
.as_ref()
@@ -613,16 +609,34 @@ impl AsyncManagedClient {
})
}
pub(crate) async fn listed_tools(&self) -> Result<Vec<ToolInfo>, StartupOutcomeError> {
// Plugin provenance is resolved per-session rather than stored in shared cache payloads.
pub(crate) fn cached_catalog(&self) -> Option<(Vec<ToolInfo>, Option<Arc<serde_json::Value>>)> {
self.codex_apps_tools_cache_context
.as_ref()
.and_then(|context| {
context
.current_catalog_snapshot()
.map(|snapshot| (snapshot.tools().to_vec(), snapshot.shared_tool_list_meta()))
// Older caches still supply tools without discovery-scoped metadata.
.or_else(|| context.current_tools().map(|tools| (tools, None)))
})
.or_else(|| {
self.tool_catalog_cache_context
.as_ref()
.and_then(|cache| cache.current_catalog_or(/*fallback*/ None))
})
}
pub(crate) async fn listed_catalog(
&self,
) -> Result<(Vec<ToolInfo>, Option<Arc<serde_json::Value>>), StartupOutcomeError> {
if !self.startup_complete.load(Ordering::Acquire)
&& let Some(startup_tools) = self.cached_tools()
&& let Some(catalog) = self.cached_catalog()
{
Ok(startup_tools)
Ok(catalog)
} else {
match self.client().await {
Ok(client) => Ok(client.listed_tools().await),
Err(error) if self.is_codex_apps_mcp_server => self.cached_tools().ok_or(error),
Ok(client) => Ok(client.listed_catalog().await),
Err(error) if self.is_codex_apps_mcp_server => self.cached_catalog().ok_or(error),
Err(error) => Err(error),
}
}
@@ -673,14 +687,23 @@ pub(crate) async fn list_tools_for_client_uncached(
timeout: Option<Duration>,
catalog_item_limit: usize,
server_instructions: Option<&str>,
) -> Result<Vec<ToolInfo>> {
) -> Result<(Vec<ToolInfo>, Option<Arc<serde_json::Value>>)> {
let meta = StdMutex::new(None);
let fetch_start = Instant::now();
let tools = collect_paginated_with_limit("tools/list", timeout, catalog_item_limit, |params| {
let client = Arc::clone(client);
let meta = &meta;
let first_page = params.as_ref().is_none_or(|params| params.cursor.is_none());
async move {
let response = client
.list_tools_with_connector_ids(params, timeout)
.await?;
if first_page {
*meta
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner) =
response.meta.map(Arc::new);
}
Ok((response.tools, response.next_cursor))
}
})
@@ -708,7 +731,10 @@ pub(crate) async fn list_tools_for_client_uncached(
&[],
);
}
Ok(tools)
let meta = meta
.into_inner()
.unwrap_or_else(std::sync::PoisonError::into_inner);
Ok((tools, meta))
}
/// Filters disabled connectors, presents declared Codex Apps file parameters to the model as
@@ -1023,7 +1049,7 @@ async fn start_server_task(
let fetch_ticket = codex_apps_tools_cache_context
.as_ref()
.map(|context| context.begin_fetch(ConnectorRuntimeFetchSource::Startup));
let tools = list_tools_for_client_uncached(
let (tools, meta) = list_tools_for_client_uncached(
&server_name,
is_codex_apps_mcp_server,
/*codex_apps_refresh_trigger*/ "initial",
@@ -1037,11 +1063,16 @@ async fn start_server_task(
let client_tools: Arc<[ToolInfo]> =
match (codex_apps_tools_cache_context.as_ref(), fetch_ticket) {
(Some(context), Some(ticket)) if server_disables_tool_catalog_cache => {
context.publish_runtime_if_newest_accepted(ticket, &server_info, tools.clone());
context.publish_runtime_if_newest_accepted(
ticket,
&server_info,
tools.clone(),
meta.clone(),
);
tools.into()
}
(Some(context), Some(ticket)) => context
.publish_runtime_if_newest_accepted(ticket, &server_info, tools)
.publish_runtime_if_newest_accepted(ticket, &server_info, tools, meta.clone())
.shared_tools(),
(None, None) => tools.into(),
_ => unreachable!("Codex Apps fetch ticket requires cache context"),
@@ -1055,7 +1086,7 @@ async fn start_server_task(
// Discovery can be session-specific, so do not cache later live catalogs.
fetch_ticket.tool_list_revision = Some(client.tool_list_revision());
}
cache_context.publish_if_newest(fetch_ticket, &client_tools);
cache_context.publish_if_newest(fetch_ticket, &client_tools, meta.clone());
}
if is_codex_apps_mcp_server || tool_catalog_cache_context.is_some() {
emit_duration(
@@ -1070,6 +1101,7 @@ async fn start_server_task(
server_info,
tool_catalog: Arc::new(ClientToolCatalog::new(
client_tools,
meta,
codex_apps_tools_cache_context
.as_ref()
.and_then(ConnectorRuntimeContext::subscribe),
codex-rs/codex-mcp/src/server_tests.rs
+1 -1
View File
@@ -171,7 +171,7 @@ fn tool_catalog_cache_does_not_reuse_catalogs_across_credential_policies() {
.expect("explicit authorization permits catalog caching")
};
let host_context = context(&host_identity);
host_context.publish_if_newest(host_context.begin_fetch(), &[]);
host_context.publish_if_newest(host_context.begin_fetch(), &[], /*meta*/ None);
assert_eq!(context(&host_identity).current_tools(), Some(Vec::new()));
assert_eq!(context(&executor_identity).current_tools(), None);
codex-rs/codex-mcp/src/tool_catalog_cache.rs
+17 -2
View File
@@ -70,6 +70,7 @@ struct OptionalStartupDeadline {
struct ToolCatalogSnapshot {
tools: Vec<ToolInfo>,
meta: Option<Arc<serde_json::Value>>,
published_at: Instant,
tool_list_revision: Option<ToolListRevision>,
}
@@ -192,6 +193,14 @@ impl McpToolCatalogCacheContext {
&self,
fallback: Option<Vec<ToolInfo>>,
) -> Option<Vec<ToolInfo>> {
self.current_catalog_or(fallback.map(|tools| (tools, None)))
.map(|(tools, _)| tools)
}
pub(crate) fn current_catalog_or(
&self,
fallback: Option<(Vec<ToolInfo>, Option<Arc<serde_json::Value>>)>,
) -> Option<(Vec<ToolInfo>, Option<Arc<serde_json::Value>>)> {
let state = lock_unpoisoned(&self.entry.state);
if state.disabled_by_server
|| state
@@ -205,7 +214,7 @@ impl McpToolCatalogCacheContext {
.snapshot
.as_ref()
.filter(|snapshot| snapshot.published_at.elapsed() <= TOOL_CATALOG_CACHE_TTL)
.map(|snapshot| snapshot.tools.clone())
.map(|snapshot| (snapshot.tools.clone(), snapshot.meta.clone()))
.or(fallback)
}
@@ -226,7 +235,12 @@ impl McpToolCatalogCacheContext {
state.snapshot = None;
}
pub(crate) fn publish_if_newest(&self, ticket: McpToolCatalogFetchTicket, tools: &[ToolInfo]) {
pub(crate) fn publish_if_newest(
&self,
ticket: McpToolCatalogFetchTicket,
tools: &[ToolInfo],
meta: Option<Arc<serde_json::Value>>,
) {
let mut state = lock_unpoisoned(&self.entry.state);
if state.disabled_by_server || ticket.generation <= state.last_accepted_generation {
return;
@@ -242,6 +256,7 @@ impl McpToolCatalogCacheContext {
state.optional_startup_deadline = None;
state.snapshot = Some(ToolCatalogSnapshot {
tools,
meta,
published_at: Instant::now(),
tool_list_revision: ticket.tool_list_revision,
});
codex-rs/connectors/src/connector_runtime/mod.rs
+35 -11
View File
@@ -107,12 +107,18 @@ pub fn connector_runtime_cache_path(codex_home: &Path, auth: Option<&CodexAuth>)
#[derive(Debug, Clone)]
pub struct ConnectorRuntimeSnapshot<T> {
tools: Arc<[T]>,
tool_list_meta: Option<Arc<serde_json::Value>>,
discovery_scope: Vec<String>,
refreshed_at: SystemTime,
generation: u64,
tools_version: u64,
}
impl<T> ConnectorRuntimeSnapshot<T> {
pub fn shared_tool_list_meta(&self) -> Option<Arc<serde_json::Value>> {
self.tool_list_meta.as_ref().map(Arc::clone)
}
pub fn tools(&self) -> &[T] {
&self.tools
}
@@ -258,6 +264,19 @@ impl<T: ConnectorRuntimePayload> ConnectorRuntimeContext<T> {
self.entry.current_snapshot.load_full()
}
pub fn current_catalog_snapshot(&self) -> Option<Arc<ConnectorRuntimeSnapshot<T>>> {
let catalog = self.live_catalog.as_ref()?;
catalog.updates.borrow().clone().or_else(|| {
let scope: Vec<_> = catalog
.scope
.iter()
.map(|part| persistence::sha1_hex(part))
.collect();
self.current_snapshot()
.filter(|snapshot| snapshot.discovery_scope.starts_with(&scope))
})
}
pub fn has_current_tools(&self) -> bool {
self.current_snapshot().is_some()
}
@@ -298,18 +317,21 @@ impl<T: ConnectorRuntimePayload> ConnectorRuntimeContext<T> {
ticket: ConnectorRuntimeFetchTicket,
server_info: &McpServerInfo,
tools: Vec<T>,
tool_list_meta: Option<Arc<serde_json::Value>>,
) -> Arc<ConnectorRuntimeSnapshot<T>> {
match self.entry.disk_cache {
ConnectorRuntimeDiskCache::Enabled => self.publish_runtime_if_newest_accepted_with(
ticket,
server_info,
tools,
tool_list_meta,
persist_codex_apps_cache,
),
ConnectorRuntimeDiskCache::Disabled => self.publish_runtime_if_newest_accepted_with(
ticket,
server_info,
tools,
tool_list_meta,
|_, _, _| {},
),
}
@@ -320,6 +342,7 @@ impl<T: ConnectorRuntimePayload> ConnectorRuntimeContext<T> {
ticket: ConnectorRuntimeFetchTicket,
server_info: &McpServerInfo,
tools: Vec<T>,
tool_list_meta: Option<Arc<serde_json::Value>>,
persist: impl FnOnce(&ConnectorRuntimeContext<T>, &McpServerInfo, &ConnectorRuntimeSnapshot<T>),
) -> Arc<ConnectorRuntimeSnapshot<T>> {
let publish_start = Instant::now();
@@ -337,6 +360,18 @@ impl<T: ConnectorRuntimePayload> ConnectorRuntimeContext<T> {
_ => (tools.into(), ticket.generation),
};
let snapshot = Arc::new(ConnectorRuntimeSnapshot {
tool_list_meta,
discovery_scope: self
.live_catalog
.as_ref()
.map(|catalog| {
catalog
.scope
.iter()
.map(|part| persistence::sha1_hex(part))
.collect()
})
.unwrap_or_default(),
tools,
tools_version,
refreshed_at: SystemTime::now(),
@@ -381,17 +416,6 @@ impl<T: ConnectorRuntimePayload> ConnectorRuntimeContext<T> {
);
snapshot
}
pub fn publish_if_newest_accepted(
&self,
ticket: ConnectorRuntimeFetchTicket,
server_info: &McpServerInfo,
tools: Vec<T>,
) -> Vec<T> {
self.publish_runtime_if_newest_accepted(ticket, server_info, tools)
.tools
.to_vec()
}
}
#[derive(Debug, Clone, Copy)]
codex-rs/connectors/src/connector_runtime/persistence.rs
+13 -1
View File
@@ -62,6 +62,8 @@ pub(crate) fn load_cached_connector_runtime_for_identity<T: ConnectorRuntimePayl
(cache.schema_version == CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION).then_some(
ConnectorRuntimeSnapshot {
tools: cache.tools,
tool_list_meta: cache.tool_list_meta.map(Arc::new),
discovery_scope: cache.discovery_scope,
refreshed_at: modified_at,
generation: 0,
tools_version: 0,
@@ -80,6 +82,8 @@ where
let bytes = serde_json::to_vec_pretty(&CodexAppsToolsDiskCache {
schema_version: CODEX_APPS_TOOLS_CACHE_SCHEMA_VERSION,
tools: snapshot.tools.clone(),
tool_list_meta: snapshot.tool_list_meta.as_deref().cloned(),
discovery_scope: snapshot.discovery_scope.clone(),
})
.context("failed to serialize connector runtime cache")?;
write_codex_apps_cache_file(&cache_path, "runtime", bytes)
@@ -206,6 +210,10 @@ fn write_codex_apps_cache_file(
#[derive(Debug, Clone, Serialize, Deserialize)]
struct CodexAppsToolsDiskCache<T> {
#[serde(default, skip_serializing_if = "Option::is_none")]
tool_list_meta: Option<serde_json::Value>,
#[serde(default)]
discovery_scope: Vec<String>,
schema_version: u8,
tools: Arc<[T]>,
}
@@ -216,7 +224,7 @@ struct CodexAppsServerInfoDiskCache {
server_info: McpServerInfo,
}
fn sha1_hex(s: &str) -> String {
pub(super) fn sha1_hex(s: &str) -> String {
let mut hasher = Sha1::new();
hasher.update(s.as_bytes());
let sha1 = hasher.finalize();
@@ -233,6 +241,8 @@ pub(crate) fn write_cached_codex_apps_tools_for_test<T>(
{
let snapshot = ConnectorRuntimeSnapshot {
tools: tools.into(),
tool_list_meta: None,
discovery_scope: Vec::new(),
refreshed_at: SystemTime::now(),
generation: 0,
tools_version: 0,
@@ -265,6 +275,8 @@ where
{
let snapshot = ConnectorRuntimeSnapshot {
tools: tools.into(),
tool_list_meta: None,
discovery_scope: Vec::new(),
refreshed_at: SystemTime::now(),
generation: 0,
tools_version: 0,
codex-rs/connectors/src/connector_runtime/tests.rs
+74 -21
View File
@@ -371,11 +371,25 @@ fn codex_apps_tools_cache_publishes_newest_shared_snapshot() {
let newer_tools = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "newer")];
let older_tools = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "older")];
let published_tools =
cache_context_2.publish_if_newest_accepted(newer_ticket, &server_info, newer_tools);
let published_tools = cache_context_2
.publish_runtime_if_newest_accepted(
newer_ticket,
&server_info,
newer_tools,
/*tool_list_meta*/ None,
)
.tools()
.to_vec();
assert_eq!(cache_context_1.current_tools(), Some(published_tools));
let current_tools =
cache_context_1.publish_if_newest_accepted(older_ticket, &server_info, older_tools);
let current_tools = cache_context_1
.publish_runtime_if_newest_accepted(
older_ticket,
&server_info,
older_tools,
/*tool_list_meta*/ None,
)
.tools()
.to_vec();
assert_eq!(current_tools[0].callable_name, "newer");
assert_eq!(
@@ -403,11 +417,15 @@ fn codex_apps_tools_cache_keeps_live_publish_when_disk_persistence_fails() {
},
);
let tools = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "live")];
let published_tools = cache_context.publish_if_newest_accepted(
cache_context.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools.clone(),
);
let published_tools = cache_context
.publish_runtime_if_newest_accepted(
cache_context.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools.clone(),
/*tool_list_meta*/ None,
)
.tools()
.to_vec();
assert_eq!(published_tools, tools);
assert_eq!(cache_context.current_tools(), Some(tools));
@@ -450,11 +468,15 @@ fn connector_runtime_without_cache_publishes_without_writing() {
},
);
let tools = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "live")];
let published_tools = context.publish_if_newest_accepted(
context.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools.clone(),
);
let published_tools = context
.publish_runtime_if_newest_accepted(
context.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools.clone(),
/*tool_list_meta*/ None,
)
.tools()
.to_vec();
assert_eq!(published_tools, tools);
assert_eq!(context.current_tools(), Some(tools));
@@ -514,15 +536,35 @@ fn live_catalogs_isolate_scopes_and_reject_older_refreshes() {
let tools_a = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "newer-a")];
let tools_b = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "newest-b")];
scope_b.publish_if_newest_accepted(newest_b, &server_info, tools_b.clone());
scope_b
.publish_runtime_if_newest_accepted(
newest_b,
&server_info,
tools_b.clone(),
/*tool_list_meta*/ None,
)
.tools()
.to_vec();
assert!(updates_a.borrow().is_none());
let discovery = scope_a.publish_if_newest_accepted(newer_a, &server_info, tools_a.clone());
let discovery = scope_a
.publish_runtime_if_newest_accepted(
newer_a,
&server_info,
tools_a.clone(),
/*tool_list_meta*/ None,
)
.tools()
.to_vec();
assert_eq!(discovery, tools_b);
scope_a.publish_if_newest_accepted(
older_a,
&server_info,
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "stale-a")],
);
scope_a
.publish_runtime_if_newest_accepted(
older_a,
&server_info,
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "stale-a")],
/*tool_list_meta*/ None,
)
.tools()
.to_vec();
assert_eq!(
updates_a.borrow().as_ref().expect("scope A tools").tools(),
&tools_a
@@ -552,6 +594,7 @@ fn contexts_for_different_identities_keep_isolated_snapshots() {
context_a.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools_a.clone(),
/*tool_list_meta*/ None,
);
let older_ticket_a = context_a.begin_fetch(ConnectorRuntimeFetchSource::Startup);
let context_b = manager.context(
@@ -587,17 +630,20 @@ fn contexts_for_different_identities_keep_isolated_snapshots() {
context_b.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
tools_b.clone(),
/*tool_list_meta*/ None,
);
let newer_tools_a = vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "newer-a")];
let newer_snapshot_a = same_context_a.publish_runtime_if_newest_accepted(
same_context_a.begin_fetch(ConnectorRuntimeFetchSource::HardRefresh),
&create_test_server_info("Codex Apps"),
newer_tools_a.clone(),
/*tool_list_meta*/ None,
);
let stale_snapshot_a = context_a.publish_runtime_if_newest_accepted(
older_ticket_a,
&create_test_server_info("Codex Apps"),
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "stale-a")],
/*tool_list_meta*/ None,
);
assert_eq!(snapshot_a.tools(), &tools_a);
@@ -657,6 +703,7 @@ fn live_provider_reuses_equal_tools_only_within_its_scope() {
context.begin_fetch(ConnectorRuntimeFetchSource::Startup),
&create_test_server_info("Apps"),
tools.clone(),
/*tool_list_meta*/ None,
)
};
let original = publish(&first);
@@ -746,6 +793,7 @@ fn accepted_generations_finish_persistence_in_order() {
older_ticket,
&create_test_server_info("Codex Apps"),
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "older")],
/*tool_list_meta*/ None,
move |_, _, _| {
older_persisting_tx
.send(())
@@ -765,6 +813,7 @@ fn accepted_generations_finish_persistence_in_order() {
newer_ticket,
&create_test_server_info("Codex Apps"),
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "newer")],
/*tool_list_meta*/ None,
move |_, _, _| {
newer_persisting_tx
.send(())
@@ -807,6 +856,7 @@ fn personal_and_workspace_contexts_are_distinct_even_with_matching_ids() {
personal_context.begin_fetch(ConnectorRuntimeFetchSource::Startup),
&create_test_server_info("Codex Apps"),
personal_tools.clone(),
/*tool_list_meta*/ None,
);
let workspace_context = manager.context(
@@ -823,6 +873,7 @@ fn personal_and_workspace_contexts_are_distinct_even_with_matching_ids() {
workspace_context.begin_fetch(ConnectorRuntimeFetchSource::Startup),
&create_test_server_info("Codex Apps"),
workspace_tools.clone(),
/*tool_list_meta*/ None,
);
assert_eq!(personal_context.current_tools(), Some(personal_tools));
@@ -848,6 +899,7 @@ fn live_publish_sets_timestamp_and_stale_publish_preserves_it() {
current_ticket,
&create_test_server_info("Codex Apps"),
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "current")],
/*tool_list_meta*/ None,
);
let after = SystemTime::now();
@@ -858,6 +910,7 @@ fn live_publish_sets_timestamp_and_stale_publish_preserves_it() {
stale_ticket,
&create_test_server_info("Codex Apps"),
vec![create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "stale")],
/*tool_list_meta*/ None,
);
assert!(Arc::ptr_eq(&current, &stale));
assert_eq!(stale.refreshed_at(), current.refreshed_at());
codex-rs/rmcp-client/src/rmcp_client.rs
+2
View File
@@ -389,6 +389,7 @@ pub struct ToolWithConnectorId {
}
pub struct ListToolsWithConnectorIdResult {
pub meta: Option<serde_json::Value>,
pub next_cursor: Option<String>,
pub tools: Vec<ToolWithConnectorId>,
}
@@ -750,6 +751,7 @@ impl RmcpClient {
})
.collect::<Result<Vec<_>>>()?;
Ok(ListToolsWithConnectorIdResult {
meta: result.meta.map(|meta| serde_json::json!(meta)),
next_cursor: result.next_cursor,
tools,
})
codex-rs/tui/src/app/background_requests.rs
+2
View File
@@ -1599,6 +1599,7 @@ mod tests {
let statuses = vec![
McpServerStatus {
server_capabilities: None,
tool_list_meta: None,
tools_error: None,
name: "docs".to_string(),
runtime_status: None,
@@ -1624,6 +1625,7 @@ mod tests {
},
McpServerStatus {
server_capabilities: None,
tool_list_meta: None,
tools_error: None,
name: "disabled".to_string(),
runtime_status: None,
codex-rs/tui/src/app/tests.rs
+1
View File
@@ -504,6 +504,7 @@ async fn handle_mcp_inventory_result_respects_origin_thread() {
app.handle_mcp_inventory_result(
Ok(vec![McpServerStatus {
server_capabilities: None,
tool_list_meta: None,
tools_error: None,
name: "docs".to_string(),
runtime_status: None,
codex-rs/tui/src/history_cell/mcp_tests.rs
+1
View File
@@ -25,6 +25,7 @@ fn mcp_inventory_connection_states() {
.into_iter()
.map(|(name, runtime_status)| McpServerStatus {
server_capabilities: None,
tool_list_meta: None,
name: name.to_string(),
runtime_status,
plugin_id: None,
codex-rs/tui/src/history_cell/tests.rs
+2
View File
@@ -1134,6 +1134,7 @@ async fn mcp_tools_output_lists_tools_for_hyphenated_server_names() {
fn mcp_tools_output_from_statuses_renders_status_only_servers() {
let statuses = vec![McpServerStatus {
server_capabilities: None,
tool_list_meta: None,
tools_error: None,
name: "plugin_docs".to_string(),
runtime_status: None,
@@ -1169,6 +1170,7 @@ fn mcp_tools_output_from_statuses_renders_status_only_servers() {
fn mcp_tools_output_from_statuses_renders_verbose_inventory() {
let statuses = vec![McpServerStatus {
server_capabilities: None,
tool_list_meta: None,
tools_error: None,
name: "plugin_docs".to_string(),
runtime_status: None,
sdk/python/src/openai_codex/generated/v2_all.py
+7
View File
@@ -8839,6 +8839,13 @@ class McpServerStatus(BaseModel):
),
] = None
server_info: Annotated[McpServerInfo | None, Field(alias="serverInfo")] = None
tool_list_meta: Annotated[
Any | None,
Field(
alias="toolListMeta",
description="Opaque `_meta` from the first page of the corresponding MCP tool catalog. Null when unavailable.",
),
] = None
tools: dict[str, Tool]
tools_error: Annotated[
str | None,