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

core: capture request-stable connector runtime tools

Browse Source
This commit is contained in:
Matthew Zeng
2026-07-08 15:11:17 -07:00
parent adefe7420c
commit b2fb980f67
16 changed files with 691 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/codex-mcp/src/connection_manager.rs
+47
View File
@@ -122,6 +122,12 @@ pub struct McpConnectionManager {
startup_cancellation_token: CancellationToken,
}
#[derive(Clone, Copy)]
enum CodexAppsToolSource<'a> {
Live,
Captured(Option<&'a ConnectorRuntimeSnapshot>),
}
impl McpConnectionManager {
#[allow(clippy::too_many_arguments)]
pub async fn new(
@@ -493,8 +499,49 @@ impl McpConnectionManager {
/// Returns all tools with model-visible names normalized.
#[instrument(level = "trace", skip_all, fields(mcp_server_count = self.clients.len()))]
pub async fn list_all_tools(&self) -> Vec<ToolInfo> {
self.list_all_tools_from_source(CodexAppsToolSource::Live)
.await
}
/// Returns model-visible tools using a request-captured connector runtime.
///
/// The captured snapshot replaces only the host-owned `codex_apps` tool
/// source. `None` represents a captured empty connector runtime, while all
/// other MCP servers continue to read their live clients.
#[instrument(level = "trace", skip_all, fields(mcp_server_count = self.clients.len()))]
pub async fn list_all_tools_with_connector_runtime_snapshot(
&self,
snapshot: Option<&ConnectorRuntimeSnapshot>,
) -> Vec<ToolInfo> {
self.list_all_tools_from_source(CodexAppsToolSource::Captured(snapshot))
.await
}
/// Returns the snapshot owned by this connection manager's `codex_apps`
/// client context.
pub fn connector_runtime_snapshot(&self) -> Option<Arc<ConnectorRuntimeSnapshot>> {
self.clients
.get(CODEX_APPS_MCP_SERVER_NAME)
.and_then(AsyncManagedClient::connector_runtime_snapshot)
}
async fn list_all_tools_from_source(
&self,
codex_apps_source: CodexAppsToolSource<'_>,
) -> Vec<ToolInfo> {
let mut tools = Vec::new();
for (server_name, managed_client) in &self.clients {
if managed_client.is_codex_apps_mcp_server
&& let CodexAppsToolSource::Captured(snapshot) = codex_apps_source
{
tools.extend(
managed_client
.listed_tools_from_connector_runtime_snapshot(snapshot)
.into_iter()
.map(|tool| self.with_server_metadata(tool)),
);
continue;
}
managed_client.reconnect_failed_startup().await;
let has_cached_tools = managed_client.has_cached_tools();
let startup_complete = managed_client
codex-rs/codex-mcp/src/connection_manager_tests.rs
+239
View File
@@ -28,6 +28,8 @@ use codex_config::McpServerToolConfig;
use codex_config::types::AuthKeyringBackendKind;
use codex_config::types::OAuthCredentialsStoreMode;
use codex_exec_server::EnvironmentManager;
use codex_plugin::AppConnectorId;
use codex_plugin::PluginCapabilitySummary;
use codex_protocol::ToolName;
use codex_protocol::mcp::McpServerInfo;
use codex_protocol::models::PermissionProfile;
@@ -48,6 +50,7 @@ use std::collections::HashSet;
use std::io;
use std::sync::Arc;
use std::sync::atomic::AtomicUsize;
use std::time::SystemTime;
use tempfile::tempdir;
use tokio::io::DuplexStream;
@@ -96,6 +99,13 @@ fn create_test_server_info(title: &str) -> McpServerInfo {
}
}
fn create_test_connector_runtime_snapshot(tools: Vec<ToolInfo>) -> ConnectorRuntimeSnapshot {
ConnectorRuntimeSnapshot {
tools,
refreshed_at: SystemTime::now(),
}
}
struct TestInProcessTransportFactory;
impl InProcessTransportFactory for TestInProcessTransportFactory {
@@ -850,6 +860,235 @@ async fn list_all_tools_uses_shared_codex_apps_cache_while_client_is_pending() {
assert_eq!(tool.callable_name, "calendar_create_event");
}
#[tokio::test]
async fn captured_connector_runtime_substitutes_apps_and_keeps_custom_mcp_live() {
let mut apps_client = create_ready_async_managed_client(vec![create_test_tool(
CODEX_APPS_MCP_SERVER_NAME,
"live_app_tool",
)])
.await;
apps_client.is_codex_apps_mcp_server = true;
let custom_client =
create_ready_async_managed_client(vec![create_test_tool("custom", "live_custom_tool")])
.await;
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
let permission_profile = Constrained::allow_any(PermissionProfile::default());
let mut manager = McpConnectionManager::new_uninitialized(
&approval_policy,
&permission_profile,
/*prefix_mcp_tool_names*/ true,
);
manager
.clients
.insert(CODEX_APPS_MCP_SERVER_NAME.to_string(), apps_client);
manager.clients.insert("custom".to_string(), custom_client);
let snapshot = create_test_connector_runtime_snapshot(vec![create_test_tool(
CODEX_APPS_MCP_SERVER_NAME,
"captured_app_tool",
)]);
let tools = manager
.list_all_tools_with_connector_runtime_snapshot(Some(&snapshot))
.await;
assert_eq!(
model_tool_names(&tools),
HashSet::from([
ToolName::namespaced("mcp__codex_apps", "captured_app_tool"),
ToolName::namespaced("mcp__custom", "live_custom_tool"),
])
);
}
#[tokio::test]
async fn captured_empty_connector_runtime_skips_pending_apps_but_keeps_custom_mcp_live() {
let pending_client = futures::future::pending::<Result<ManagedClient, StartupOutcomeError>>()
.boxed()
.shared();
let apps_client = AsyncManagedClient {
client: pending_client,
is_codex_apps_mcp_server: true,
cached_server_info: None,
codex_apps_tools_cache_context: None,
tool_filter: ToolFilter::default(),
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
startup_reconnect: None,
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
cancel_token: CancellationToken::new(),
};
let custom_client =
create_ready_async_managed_client(vec![create_test_tool("custom", "live_custom_tool")])
.await;
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
let permission_profile = Constrained::allow_any(PermissionProfile::default());
let mut manager = McpConnectionManager::new_uninitialized(
&approval_policy,
&permission_profile,
/*prefix_mcp_tool_names*/ true,
);
manager
.clients
.insert(CODEX_APPS_MCP_SERVER_NAME.to_string(), apps_client);
manager.clients.insert("custom".to_string(), custom_client);
let tools = tokio::time::timeout(
Duration::from_millis(10),
manager.list_all_tools_with_connector_runtime_snapshot(/*snapshot*/ None),
)
.await
.expect("captured empty connector runtime should not wait for apps startup");
assert_eq!(
model_tool_names(&tools),
HashSet::from([ToolName::namespaced("mcp__custom", "live_custom_tool")])
);
}
#[tokio::test]
async fn captured_connector_runtime_applies_filter_provenance_and_name_normalization() {
let config = crate::mcp::McpConfig {
chatgpt_base_url: "https://chatgpt.com".to_string(),
apps_mcp_product_sku: None,
codex_home: PathBuf::new(),
mcp_oauth_credentials_store_mode: OAuthCredentialsStoreMode::default(),
auth_keyring_backend_kind: AuthKeyringBackendKind::default(),
mcp_oauth_callback_port: None,
mcp_oauth_callback_url: None,
skill_mcp_dependency_install_enabled: true,
approval_policy: Constrained::allow_any(AskForApproval::OnRequest),
codex_linux_sandbox_exe: None,
use_legacy_landlock: false,
apps_enabled: true,
prefix_mcp_tool_names: true,
client_elicitation_capability: ElicitationCapability::default(),
mcp_server_catalog: crate::ResolvedMcpCatalog::default(),
connector_snapshot: codex_connectors::ConnectorSnapshot::from_plugin_capability_summaries(
&[PluginCapabilitySummary {
config_name: "test-plugin-id".to_string(),
display_name: "test-plugin".to_string(),
app_connector_ids: vec![AppConnectorId("connector_test".to_string())],
..PluginCapabilitySummary::default()
}],
),
};
let provenance = Arc::new(crate::mcp::tool_plugin_provenance(&config));
let mut apps_client = create_ready_async_managed_client(Vec::new()).await;
apps_client.is_codex_apps_mcp_server = true;
apps_client.tool_filter = ToolFilter {
enabled: Some(HashSet::from(["allowed-tool".to_string()])),
disabled: HashSet::new(),
};
apps_client.tool_plugin_provenance = Arc::clone(&provenance);
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
let permission_profile = Constrained::allow_any(PermissionProfile::default());
let mut manager = McpConnectionManager::new_uninitialized(
&approval_policy,
&permission_profile,
/*prefix_mcp_tool_names*/ true,
);
manager.tool_plugin_provenance = provenance;
manager
.clients
.insert(CODEX_APPS_MCP_SERVER_NAME.to_string(), apps_client);
let mut allowed = create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "allowed-tool");
allowed.connector_id = Some("connector_test".to_string());
let snapshot = create_test_connector_runtime_snapshot(vec![
allowed,
create_test_tool(CODEX_APPS_MCP_SERVER_NAME, "filtered-tool"),
]);
let tools = manager
.list_all_tools_with_connector_runtime_snapshot(Some(&snapshot))
.await;
assert_eq!(tools.len(), 1);
let tool = &tools[0];
assert_eq!(
tool.canonical_tool_name(),
ToolName::namespaced("mcp__codex_apps", "allowed_tool")
);
assert_eq!(tool.callable_name, "allowed_tool");
assert_eq!(tool.tool.name.as_ref(), "allowed-tool");
assert_eq!(tool.plugin_display_names, vec!["test-plugin"]);
assert_eq!(
tool.tool.description.as_deref(),
Some("Test tool: allowed-tool. This tool is part of plugin `test-plugin`.")
);
}
#[tokio::test]
async fn captured_connector_runtime_is_stable_until_its_context_is_discarded() {
let codex_home = tempdir().expect("tempdir");
let cache = CodexAppsToolsCache::default();
let context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-one".to_string()),
chatgpt_user_id: Some("user-one".to_string()),
is_workspace_account: false,
},
);
context.store_current_tools_for_test(vec![create_test_tool(
CODEX_APPS_MCP_SERVER_NAME,
"captured_tool",
)]);
let approval_policy = Constrained::allow_any(AskForApproval::OnRequest);
let permission_profile = Constrained::allow_any(PermissionProfile::default());
let mut manager = McpConnectionManager::new_uninitialized(
&approval_policy,
&permission_profile,
/*prefix_mcp_tool_names*/ true,
);
manager.clients.insert(
CODEX_APPS_MCP_SERVER_NAME.to_string(),
AsyncManagedClient {
client: futures::future::pending().boxed().shared(),
is_codex_apps_mcp_server: true,
cached_server_info: None,
codex_apps_tools_cache_context: Some(context.clone()),
tool_filter: ToolFilter::default(),
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
startup_reconnect: None,
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
cancel_token: CancellationToken::new(),
},
);
let snapshot = manager
.connector_runtime_snapshot()
.expect("manager-owned snapshot");
context.store_current_tools_for_test(vec![create_test_tool(
CODEX_APPS_MCP_SERVER_NAME,
"newer_tool",
)]);
let tools = manager
.list_all_tools_with_connector_runtime_snapshot(Some(&snapshot))
.await;
assert_eq!(
tools
.iter()
.map(|tool| tool.callable_name.as_str())
.collect::<Vec<_>>(),
vec!["captured_tool"]
);
let _replacement_context = cache.context(
codex_home.path().to_path_buf(),
CodexAppsToolsCacheKey {
account_id: Some("account-two".to_string()),
chatgpt_user_id: Some("user-two".to_string()),
is_workspace_account: false,
},
);
assert!(manager.connector_runtime_snapshot().is_none());
let tools = manager
.list_all_tools_with_connector_runtime_snapshot(Some(&snapshot))
.await;
assert!(tools.is_empty());
}
#[tokio::test]
async fn list_available_server_infos_uses_cache_while_client_is_pending() {
let pending_client = futures::future::pending::<Result<ManagedClient, StartupOutcomeError>>()
codex-rs/codex-mcp/src/rmcp_client.rs
+25
View File
@@ -23,6 +23,7 @@ use crate::codex_apps::normalize_codex_apps_callable_namespace;
use crate::codex_apps::normalize_codex_apps_tool_title;
use crate::connector_runtime::CodexAppsToolsCacheContext;
use crate::connector_runtime::CodexAppsToolsFetchSource;
use crate::connector_runtime::ConnectorRuntimeSnapshot;
use crate::connector_runtime::load_startup_cached_codex_apps_server_info;
use crate::elicitation::ElicitationRequestManager;
use crate::mcp::CODEX_APPS_MCP_SERVER_NAME;
@@ -546,6 +547,30 @@ impl AsyncManagedClient {
.map(|tools| filter_tools(tools, &self.tool_filter))
}
pub(crate) fn connector_runtime_snapshot(&self) -> Option<Arc<ConnectorRuntimeSnapshot>> {
self.codex_apps_tools_cache_context
.as_ref()
.and_then(CodexAppsToolsCacheContext::current_snapshot)
}
pub(crate) fn listed_tools_from_connector_runtime_snapshot(
&self,
snapshot: Option<&ConnectorRuntimeSnapshot>,
) -> Vec<ToolInfo> {
debug_assert!(self.is_codex_apps_mcp_server);
if self
.codex_apps_tools_cache_context
.as_ref()
.is_some_and(|context| !context.is_active())
{
return Vec::new();
}
let tools = snapshot
.map(|snapshot| filter_tools(snapshot.tools().to_vec(), &self.tool_filter))
.unwrap_or_default();
prepare_codex_apps_tools_for_model(tools, &self.tool_plugin_provenance)
}
pub(crate) async fn listed_tools(&self) -> Option<Vec<ToolInfo>> {
if !self.connector_runtime_context_is_active() {
return None;
codex-rs/core/config.schema.json
+6
View File
@@ -431,6 +431,9 @@
}
]
},
"apps_runtime_state_refactor": {
"type": "boolean"
},
"auth_elicitation": {
"type": "boolean"
},
@@ -4803,6 +4806,9 @@
}
]
},
"apps_runtime_state_refactor": {
"type": "boolean"
},
"auth_elicitation": {
"type": "boolean"
},
codex-rs/core/src/codex_delegate.rs
+44 -6
View File
@@ -6,11 +6,14 @@ use async_channel::Sender;
use codex_analytics::GuardianApprovalRequestSource;
use codex_async_utils::OrCancelExt;
use codex_extension_api::LoadedUserInstructions;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_mcp::McpConnectionManager;
use codex_protocol::protocol::ApplyPatchApprovalRequestEvent;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ExecApprovalRequestEvent;
use codex_protocol::protocol::McpInvocation;
use codex_protocol::protocol::McpToolCallBeginEvent;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RequestUserInputEvent;
use codex_protocol::protocol::ReviewDecision;
@@ -45,6 +48,7 @@ use crate::mcp_tool_call::build_guardian_mcp_tool_review_request;
use crate::mcp_tool_call::is_mcp_tool_approval_question_id;
use crate::mcp_tool_call::lookup_mcp_tool_metadata;
use crate::mcp_tool_call::mcp_approvals_reviewer;
use crate::mcp_tool_call::mcp_tool_approval_metadata_from_begin_event;
use crate::session::Codex;
use crate::session::CodexSpawnArgs;
use crate::session::CodexSpawnOk;
@@ -52,6 +56,7 @@ use crate::session::SUBMISSION_CHANNEL_CAPACITY;
use crate::session::emit_subagent_session_started;
use crate::session::session::Session;
use crate::session::turn_context::TurnContext;
use codex_features::Feature;
use codex_login::AuthManager;
use codex_models_manager::manager::SharedModelsManager;
use codex_protocol::error::CodexErr;
@@ -366,19 +371,18 @@ async fn forward_events(
id,
msg: EventMsg::McpToolCallBegin(event),
} => {
// Runtime refreshes are published before a request step is captured, so
// the child runtime at call begin is the one executing this invocation.
// Cache its metadata now; the later approval event has only a call ID.
// Cache metadata now; the later approval event has only a call ID. Under
// the runtime refactor, consume the child's internal request-stable
// metadata handoff rather than re-listing a newer connector snapshot.
let metadata = if let Some(turn_context) =
codex.session.turn_context_for_sub_id(&id).await
{
let mcp = codex.session.services.latest_mcp_runtime();
lookup_mcp_tool_metadata(
delegated_mcp_tool_metadata_for_begin(
codex.session.as_ref(),
turn_context.as_ref(),
mcp.manager(),
&event.invocation.server,
&event.invocation.tool,
&event,
)
.await
} else {
@@ -412,6 +416,10 @@ async fn forward_events(
id,
msg: EventMsg::McpToolCallEnd(event),
} => {
let _ = codex
.session
.take_delegated_mcp_tool_metadata(&event.call_id)
.await;
pending_mcp_invocations.lock().await.remove(&event.call_id);
if !forward_event_or_shutdown(
&codex,
@@ -439,6 +447,36 @@ async fn forward_events(
}
}
pub(crate) async fn delegated_mcp_tool_metadata_for_begin(
session: &Session,
turn_context: &TurnContext,
manager: &McpConnectionManager,
event: &McpToolCallBeginEvent,
) -> Option<McpToolApprovalMetadata> {
if turn_context
.config
.features
.enabled(Feature::AppsRuntimeStateRefactor)
&& event.invocation.server == CODEX_APPS_MCP_SERVER_NAME
{
return session
.take_delegated_mcp_tool_metadata(&event.call_id)
.await
.or_else(|| mcp_tool_approval_metadata_from_begin_event(event));
}
let mcp_tools = manager.list_all_tools().await;
lookup_mcp_tool_metadata(
session,
turn_context,
manager,
&mcp_tools,
&event.invocation.server,
&event.invocation.tool,
)
.await
}
/// Ask the delegate to stop and drain its events so background sends do not hit a closed channel.
async fn shutdown_delegate(codex: &Codex) {
let _ = codex.submit(Op::Interrupt).await;
codex-rs/core/src/mcp_tool_call.rs
+111 -58
View File
@@ -39,6 +39,7 @@ use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY;
use codex_mcp::McpConnectionManager;
use codex_mcp::McpPermissionPromptAutoApproveContext;
use codex_mcp::SandboxState;
use codex_mcp::ToolInfo;
use codex_mcp::auth_elicitation_completed_result;
use codex_mcp::build_auth_elicitation_plan;
use codex_mcp::declared_openai_file_input_param_names;
@@ -66,7 +67,9 @@ use codex_protocol::mcp_approval_meta::TOOL_TITLE_KEY as MCP_TOOL_APPROVAL_TOOL_
use codex_protocol::openai_models::InputModality;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::McpInvocation;
use codex_protocol::protocol::McpToolCallBeginEvent;
use codex_protocol::protocol::ReviewDecision;
use codex_protocol::protocol::SessionSource;
use codex_protocol::request_user_input::RequestUserInputAnswer;
use codex_protocol::request_user_input::RequestUserInputArgs;
use codex_protocol::request_user_input::RequestUserInputQuestion;
@@ -145,10 +148,12 @@ pub(crate) async fn handle_mcp_tool_call(
arguments: arguments_value.clone(),
};
let mcp_tools = step_context.mcp_tools().await;
let metadata = lookup_mcp_tool_metadata(
sess.as_ref(),
turn_context.as_ref(),
manager,
&mcp_tools,
&server,
&tool_name,
)
@@ -224,6 +229,17 @@ pub(crate) async fn handle_mcp_tool_call(
.unwrap_or_else(|| JsonValue::Object(serde_json::Map::new())),
};
}
if turn_context
.config
.features
.enabled(Feature::AppsRuntimeStateRefactor)
&& server == CODEX_APPS_MCP_SERVER_NAME
&& matches!(&turn_context.session_source, SessionSource::SubAgent(_))
&& let Some(metadata) = metadata.clone()
{
sess.store_delegated_mcp_tool_metadata(call_id.clone(), metadata)
.await;
}
notify_mcp_tool_call_started(
sess.as_ref(),
turn_context.as_ref(),
@@ -438,7 +454,7 @@ async fn handle_approved_mcp_tool_call(
truncate_mcp_tool_result_for_event(&result),
)
.await;
maybe_track_codex_app_used(sess, turn_context, manager, &server, &tool_name).await;
maybe_track_codex_app_used(sess, turn_context, &server, metadata).await;
let outcome = mcp_call_metric_outcome(&result);
emit_mcp_call_metrics(
@@ -950,24 +966,22 @@ async fn notify_mcp_tool_call_completed(
sess.emit_turn_item_completed(turn_context, item).await;
}
struct McpAppUsageMetadata {
connector_id: Option<String>,
app_name: Option<String>,
}
async fn maybe_track_codex_app_used(
sess: &Session,
turn_context: &TurnContext,
manager: &McpConnectionManager,
server: &str,
tool_name: &str,
metadata: Option<&McpToolApprovalMetadata>,
) {
if server != CODEX_APPS_MCP_SERVER_NAME {
return;
}
let metadata = lookup_mcp_app_usage_metadata(manager, server, tool_name).await;
let (connector_id, app_name) = metadata
.map(|metadata| (metadata.connector_id, metadata.app_name))
.map(|metadata| {
(
metadata.connector_id.clone(),
metadata.connector_name.clone(),
)
})
.unwrap_or((None, None));
let invocation_type = if let Some(connector_id) = connector_id.as_deref() {
let mentioned_connector_ids = sess.get_connector_selection().await;
@@ -1005,7 +1019,7 @@ enum McpToolApprovalDecision {
Cancel,
}
#[derive(Clone)]
#[derive(Clone, Debug, PartialEq)]
pub(crate) struct McpToolApprovalMetadata {
annotations: Option<ToolAnnotations>,
connector_id: Option<String>,
@@ -1022,6 +1036,53 @@ pub(crate) struct McpToolApprovalMetadata {
openai_file_input_params: Option<Vec<String>>,
}
impl Session {
pub(crate) async fn store_delegated_mcp_tool_metadata(
&self,
call_id: String,
metadata: McpToolApprovalMetadata,
) {
self.pending_delegated_mcp_tool_metadata
.lock()
.await
.insert(call_id, metadata);
}
pub(crate) async fn take_delegated_mcp_tool_metadata(
&self,
call_id: &str,
) -> Option<McpToolApprovalMetadata> {
self.pending_delegated_mcp_tool_metadata
.lock()
.await
.remove(call_id)
}
}
pub(crate) fn mcp_tool_approval_metadata_from_begin_event(
event: &McpToolCallBeginEvent,
) -> Option<McpToolApprovalMetadata> {
let metadata = McpToolApprovalMetadata {
annotations: None,
connector_id: event.connector_id.clone(),
link_id: event.link_id.clone(),
connector_name: event.app_name.clone(),
connector_description: None,
connected_account_email: None,
plugin_id: event.plugin_id.clone(),
tool_title: None,
tool_description: None,
mcp_app_resource_uri: event.mcp_app_resource_uri.clone(),
template_id: event.template_id.clone(),
codex_apps_meta: None,
openai_file_input_params: None,
};
(metadata.connector_id.is_some()
|| metadata.connector_name.is_some()
|| metadata.plugin_id.is_some())
.then_some(metadata)
}
const MCP_TOOL_OPENAI_OUTPUT_TEMPLATE_META_KEY: &str = "openai/outputTemplate";
const MCP_TOOL_UI_RESOURCE_URI_META_KEY: &str = "ui/resourceUri";
const MCP_TOOL_LINK_ID_META_KEY: &str = "link_id";
@@ -1475,42 +1536,53 @@ pub(crate) async fn lookup_mcp_tool_metadata(
sess: &Session,
turn_context: &TurnContext,
manager: &McpConnectionManager,
mcp_tools: &[ToolInfo],
server: &str,
tool_name: &str,
) -> Option<McpToolApprovalMetadata> {
let plugin_id = manager
.plugin_id_for_mcp_server_name(server)
.map(str::to_string);
let tools = manager.list_all_tools().await;
let tool_info = tools
.into_iter()
.find(|tool_info| tool_info.server_name == server && tool_info.tool.name == tool_name)?;
let tool_info = mcp_tools
.iter()
.find(|tool_info| tool_info.server_name == server && tool_info.tool.name == tool_name)?
.clone();
let connector_description = if server == CODEX_APPS_MCP_SERVER_NAME {
let connectors = match connectors::list_cached_accessible_connectors_from_mcp_tools(
turn_context.config.as_ref(),
)
.await
if turn_context
.config
.features
.enabled(Feature::AppsRuntimeStateRefactor)
{
Some(connectors) => Some(connectors),
None => {
connectors::list_accessible_connectors_from_mcp_tools_with_mcp_manager(
turn_context.config.as_ref(),
/*force_refetch*/ false,
sess.services.turn_environments.environment_manager(),
Arc::clone(&sess.services.mcp_manager),
)
.await
.ok()
.map(|status| status.connectors)
}
};
connectors.and_then(|connectors| {
let connector_id = tool_info.connector_id.as_deref()?;
connectors
.into_iter()
.find(|connector| connector.id == connector_id)
.and_then(|connector| connector.description)
})
// Gateway tool metadata is a request-stable runtime fallback, not canonical app
// metadata. Canonical descriptions remain owned by app/read.
tool_info.namespace_description.clone()
} else {
let connectors = match connectors::list_cached_accessible_connectors_from_mcp_tools(
turn_context.config.as_ref(),
)
.await
{
Some(connectors) => Some(connectors),
None => {
connectors::list_accessible_connectors_from_mcp_tools_with_mcp_manager(
turn_context.config.as_ref(),
/*force_refetch*/ false,
sess.services.turn_environments.environment_manager(),
Arc::clone(&sess.services.mcp_manager),
)
.await
.ok()
.map(|status| status.connectors)
}
};
connectors.and_then(|connectors| {
let connector_id = tool_info.connector_id.as_deref()?;
connectors
.into_iter()
.find(|connector| connector.id == connector_id)
.and_then(|connector| connector.description)
})
}
} else {
None
};
@@ -1594,25 +1666,6 @@ fn get_mcp_app_resource_uri(
})
}
async fn lookup_mcp_app_usage_metadata(
manager: &McpConnectionManager,
server: &str,
tool_name: &str,
) -> Option<McpAppUsageMetadata> {
let tools = manager.list_all_tools().await;
tools.into_iter().find_map(|tool_info| {
if tool_info.server_name == server && tool_info.tool.name == tool_name {
Some(McpAppUsageMetadata {
connector_id: tool_info.connector_id,
app_name: tool_info.connector_name,
})
} else {
None
}
})
}
fn build_mcp_tool_approval_question(
question_id: String,
server: &str,
codex-rs/core/src/mcp_tool_call_tests.rs
+109
View File
@@ -93,6 +93,115 @@ fn approval_metadata(
}
}
#[tokio::test]
async fn delegated_mcp_metadata_handoff_preserves_full_snapshot_metadata() {
let (session, mut turn_context) = make_session_and_context().await;
Arc::make_mut(&mut turn_context.config)
.features
.enable(Feature::AppsRuntimeStateRefactor)
.expect("enable runtime-state refactor");
let metadata = McpToolApprovalMetadata {
annotations: Some(annotations(Some(false), Some(true), Some(true))),
connector_id: Some("connector_docs".to_string()),
link_id: Some("link-docs".to_string()),
connector_name: Some("Docs".to_string()),
connector_description: Some("Search workspace documents".to_string()),
connected_account_email: Some("user@example.com".to_string()),
plugin_id: Some("docs-plugin".to_string()),
tool_title: Some("Search docs".to_string()),
tool_description: Some("Find matching documents".to_string()),
mcp_app_resource_uri: Some("ui://docs/search".to_string()),
template_id: Some("search-template".to_string()),
codex_apps_meta: Some(serde_json::Map::from_iter([(
"request_type".to_string(),
serde_json::Value::String("search".to_string()),
)])),
openai_file_input_params: Some(vec!["attachment".to_string()]),
};
session
.store_delegated_mcp_tool_metadata("call-1".to_string(), metadata.clone())
.await;
let event = McpToolCallBeginEvent {
call_id: "call-1".to_string(),
invocation: McpInvocation {
server: CODEX_APPS_MCP_SERVER_NAME.to_string(),
tool: "search".to_string(),
arguments: Some(serde_json::json!({"query": "roadmap"})),
},
connector_id: metadata.connector_id.clone(),
mcp_app_resource_uri: metadata.mcp_app_resource_uri.clone(),
link_id: metadata.link_id.clone(),
app_name: metadata.connector_name.clone(),
template_id: metadata.template_id.clone(),
action_name: None,
plugin_id: metadata.plugin_id.clone(),
};
let mcp = session.services.latest_mcp_runtime();
let resolved = crate::codex_delegate::delegated_mcp_tool_metadata_for_begin(
&session,
&turn_context,
mcp.manager(),
&event,
)
.await;
assert_eq!(resolved, Some(metadata.clone()));
assert_eq!(
build_guardian_mcp_tool_review_request("call-1", &event.invocation, resolved.as_ref(),),
GuardianApprovalRequest::McpToolCall {
id: "call-1".to_string(),
server: CODEX_APPS_MCP_SERVER_NAME.to_string(),
tool_name: "search".to_string(),
arguments: Some(serde_json::json!({"query": "roadmap"})),
connector_id: Some("connector_docs".to_string()),
connector_name: Some("Docs".to_string()),
connector_description: Some("Search workspace documents".to_string()),
connected_account_email: Some("user@example.com".to_string()),
tool_title: Some("Search docs".to_string()),
tool_description: Some("Find matching documents".to_string()),
annotations: Some(GuardianMcpAnnotations {
destructive_hint: Some(true),
open_world_hint: Some(true),
read_only_hint: Some(false),
}),
}
);
session
.store_delegated_mcp_tool_metadata("call-2".to_string(), metadata.clone())
.await;
let custom_event = McpToolCallBeginEvent {
call_id: "call-2".to_string(),
invocation: McpInvocation {
server: "custom".to_string(),
tool: "search".to_string(),
arguments: None,
},
connector_id: None,
mcp_app_resource_uri: None,
link_id: None,
app_name: None,
template_id: None,
action_name: None,
plugin_id: None,
};
assert_eq!(
crate::codex_delegate::delegated_mcp_tool_metadata_for_begin(
&session,
&turn_context,
mcp.manager(),
&custom_event,
)
.await,
None
);
assert_eq!(
session.take_delegated_mcp_tool_metadata("call-2").await,
Some(metadata)
);
}
fn mcp_turn_metadata_context(turn_context: &TurnContext) -> McpTurnMetadataContext<'_> {
McpTurnMetadataContext {
model: turn_context.model_info.slug.as_str(),
codex-rs/core/src/session/mod.rs
+16
View File
@@ -39,6 +39,7 @@ use crate::exec_policy::ExecPolicyManager;
use crate::image_preparation::prepare_response_items;
use crate::parse_turn_item;
use crate::realtime_conversation::RealtimeConversationManager;
use crate::session::step_context::ConnectorRuntimeCapture;
use crate::session::step_context::StepContext;
use crate::session::turn_context::TurnEnvironment;
use crate::session_prefix::format_inter_agent_completion_message;
@@ -2890,11 +2891,26 @@ impl Session {
&selected_capability_roots,
)
.await;
let connector_runtime = if turn_context
.config
.features
.enabled(Feature::AppsRuntimeStateRefactor)
{
let snapshot = if turn_context.apps_enabled() {
mcp.manager().connector_runtime_snapshot()
} else {
None
};
ConnectorRuntimeCapture::Snapshot(snapshot)
} else {
ConnectorRuntimeCapture::LegacyLive
};
Arc::new(StepContext::new(
turn_context,
environments,
selected_capability_roots,
mcp,
connector_runtime,
loaded_agents_md,
))
}
codex-rs/core/src/session/session.rs
+7
View File
@@ -43,6 +43,12 @@ pub(crate) struct Session {
pub(crate) active_turn: Mutex<Option<ActiveTurn>>,
pub(crate) input_queue: InputQueue,
pub(crate) guardian_review_session: GuardianReviewSessionManager,
/// Full request-stable metadata for delegated MCP calls awaiting parent event forwarding.
///
/// This stays internal rather than widening the public tool-call-begin event with approval
/// details or connected-account data.
pub(crate) pending_delegated_mcp_tool_metadata:
Mutex<HashMap<String, crate::mcp_tool_call::McpToolApprovalMetadata>>,
pub(crate) services: SessionServices,
pub(super) next_internal_sub_id: AtomicU64,
}
@@ -1143,6 +1149,7 @@ impl Session {
active_turn: Mutex::new(None),
input_queue: InputQueue::new(),
guardian_review_session: GuardianReviewSessionManager::default(),
pending_delegated_mcp_tool_metadata: Mutex::new(HashMap::new()),
services,
next_internal_sub_id: AtomicU64::new(0),
});
codex-rs/core/src/session/step_context.rs
+31
View File
@@ -5,6 +5,16 @@ use crate::environment_selection::TurnEnvironmentSnapshot;
use crate::session::McpRuntimeSnapshot;
use crate::session::turn_context::TurnContext;
use codex_exec_server::ResolvedSelectedCapabilityRoot;
use codex_mcp::ConnectorRuntimeSnapshot;
use codex_mcp::ToolInfo;
#[derive(Debug)]
pub(crate) enum ConnectorRuntimeCapture {
/// Preserve the legacy behavior where every read observes the live Codex Apps client/cache.
LegacyLive,
/// Use exactly this committed state. `None` means no Codex Apps tools were committed at capture.
Snapshot(Option<Arc<ConnectorRuntimeSnapshot>>),
}
/// Request-scoped state that may change between model sampling requests.
#[derive(Debug)]
@@ -15,6 +25,11 @@ pub(crate) struct StepContext {
pub(crate) selected_capability_roots: Vec<ResolvedSelectedCapabilityRoot>,
/// The exact MCP config and manager used to advertise and execute tools for this step.
pub(crate) mcp: Arc<McpRuntimeSnapshot>,
/// The committed hosted-connector state captured for this step.
///
/// The snapshot variant is used only by the runtime-state refactor. Custom MCP servers remain
/// live in both modes.
pub(crate) connector_runtime: ConnectorRuntimeCapture,
/// The canonical AGENTS.md value observed with this environment snapshot.
pub(crate) loaded_agents_md: Option<Arc<LoadedAgentsMd>>,
}
@@ -25,6 +40,7 @@ impl StepContext {
environments: TurnEnvironmentSnapshot,
selected_capability_roots: Vec<ResolvedSelectedCapabilityRoot>,
mcp: Arc<McpRuntimeSnapshot>,
connector_runtime: ConnectorRuntimeCapture,
loaded_agents_md: Option<Arc<LoadedAgentsMd>>,
) -> Self {
Self {
@@ -32,7 +48,22 @@ impl StepContext {
environments,
selected_capability_roots,
mcp,
connector_runtime,
loaded_agents_md,
}
}
/// Returns tools from the exact hosted-connector snapshot captured for this step while leaving
/// unrelated MCP servers on their normal live path.
pub(crate) async fn mcp_tools(&self) -> Vec<ToolInfo> {
match &self.connector_runtime {
ConnectorRuntimeCapture::LegacyLive => self.mcp.manager().list_all_tools().await,
ConnectorRuntimeCapture::Snapshot(snapshot) => {
self.mcp
.manager()
.list_all_tools_with_connector_runtime_snapshot(snapshot.as_deref())
.await
}
}
}
}
codex-rs/core/src/session/tests.rs
+3
View File
@@ -197,6 +197,7 @@ impl StepContext {
environments,
Vec::new(),
crate::session::McpRuntimeSnapshot::new_uninitialized_for_test(&turn.config),
ConnectorRuntimeCapture::LegacyLive,
/*loaded_agents_md*/ None,
))
}
@@ -5529,6 +5530,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
active_turn: Mutex::new(None),
input_queue: super::input_queue::InputQueue::new(),
guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(),
pending_delegated_mcp_tool_metadata: Mutex::new(HashMap::new()),
services,
next_internal_sub_id: AtomicU64::new(0),
};
@@ -7660,6 +7662,7 @@ where
active_turn: Mutex::new(None),
input_queue: super::input_queue::InputQueue::new(),
guardian_review_session: crate::guardian::GuardianReviewSessionManager::default(),
pending_delegated_mcp_tool_metadata: Mutex::new(HashMap::new()),
services,
next_internal_sub_id: AtomicU64::new(0),
});
codex-rs/core/src/session/turn.rs
+7 -9
View File
@@ -261,6 +261,10 @@ pub(crate) async fn run_turn(
.config
.features
.enabled(Feature::DeferredExecutor)
|| turn_context
.config
.features
.enabled(Feature::AppsRuntimeStateRefactor)
{
world_state = sess
.record_step_world_state_if_changed(&world_state, step_context.as_ref())
@@ -549,13 +553,7 @@ async fn build_skills_and_plugins(
// Plugin mentions need raw MCP/app inventory even when app tools
// are normally hidden so we can describe the plugin's currently
// usable capabilities for this turn.
match step_context
.mcp
.manager_arc()
.list_all_tools()
.or_cancel(cancellation_token)
.await
{
match step_context.mcp_tools().or_cancel(cancellation_token).await {
Ok(mcp_tools) => mcp_tools,
Err(_) if turn_context.apps_enabled() => return None,
Err(_) => Vec::new(),
@@ -1182,8 +1180,8 @@ pub(crate) async fn built_tools(
let turn_context = step_context.turn.as_ref();
let mcp_connection_manager = step_context.mcp.manager();
let has_mcp_servers = mcp_connection_manager.has_servers();
let all_mcp_tools = mcp_connection_manager
.list_all_tools()
let all_mcp_tools = step_context
.mcp_tools()
.or_cancel(cancellation_token)
.await?;
let loaded_plugins = sess
codex-rs/core/src/session/world_state.rs
+1 -1
View File
@@ -41,7 +41,7 @@ impl Session {
}
let apps_available =
if turn_context.config.include_apps_instructions && turn_context.apps_enabled() {
let tools = step_context.mcp.manager().list_all_tools().await;
let tools = step_context.mcp_tools().await;
connectors::with_app_enabled_state(
connectors::accessible_connectors_from_mcp_tools(&tools),
&turn_context.config,
codex-rs/core/src/tools/spec_plan_tests.rs
+2
View File
@@ -32,6 +32,7 @@ use pretty_assertions::assert_eq;
use serde_json::json;
use crate::config::CurrentTimeReminderConfig;
use crate::session::step_context::ConnectorRuntimeCapture;
use crate::session::step_context::StepContext;
use crate::session::tests::make_session_and_context;
use crate::session::turn_context::TurnContext;
@@ -689,6 +690,7 @@ async fn environment_tools_follow_the_step_context() {
environments,
Vec::new(),
crate::session::McpRuntimeSnapshot::new_uninitialized_for_test(&turn.config),
ConnectorRuntimeCapture::LegacyLive,
/*loaded_agents_md*/ None,
));
codex-rs/core/tests/suite/mcp_tool_exposure.rs
+35 -2
View File
@@ -98,6 +98,15 @@ async fn code_mode_only_exposes_direct_model_only_mcp_namespaces() -> Result<()>
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn apps_guidance_appears_after_background_recovery_within_a_turn() -> Result<()> {
assert_apps_recover_between_sampling_steps(Feature::DeferredExecutor).await
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn connector_runtime_snapshot_updates_between_sampling_steps() -> Result<()> {
assert_apps_recover_between_sampling_steps(Feature::AppsRuntimeStateRefactor).await
}
async fn assert_apps_recover_between_sampling_steps(step_refresh_feature: Feature) -> Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
@@ -141,11 +150,17 @@ async fn apps_guidance_appears_after_background_recovery_within_a_turn() -> Resu
)
.await;
let mut builder =
apps_enabled_builder(apps_server.chatgpt_base_url.clone()).with_config(|config| {
apps_enabled_builder(apps_server.chatgpt_base_url.clone()).with_config(move |config| {
config
.features
.enable(Feature::DeferredExecutor)
.enable(step_refresh_feature)
.expect("test config should allow feature update");
config
.features
.enable(Feature::CodeModeOnly)
.expect("test config should allow feature update");
config.code_mode.direct_only_tool_namespaces =
vec![SEARCH_CALENDAR_NAMESPACE.to_string()];
config
.features
.enable(Feature::DefaultModeRequestUserInput)
@@ -183,6 +198,15 @@ async fn apps_guidance_appears_after_background_recovery_within_a_turn() -> Resu
.count(),
0
);
assert!(
namespace_child_tool(
&initial_request.body_json(),
SEARCH_CALENDAR_NAMESPACE,
SEARCH_CALENDAR_CREATE_TOOL,
)
.is_none(),
"Calendar should not be exposed before Apps recovers"
);
tokio::time::timeout(Duration::from_secs(3), async {
loop {
@@ -222,6 +246,15 @@ async fn apps_guidance_appears_after_background_recovery_within_a_turn() -> Resu
.count(),
1
);
assert!(
namespace_child_tool(
&requests[1].body_json(),
SEARCH_CALENDAR_NAMESPACE,
SEARCH_CALENDAR_CREATE_TOOL,
)
.is_some(),
"Calendar should be exposed after Apps recovers"
);
Ok(())
}
codex-rs/features/src/lib.rs
+8
View File
@@ -154,6 +154,8 @@ pub enum Feature {
SpawnCsv,
/// Enable apps.
Apps,
/// Use the connector-owned runtime snapshot for installed app state.
AppsRuntimeStateRefactor,
/// Enable MCP apps.
EnableMcpApps,
/// Removed compatibility flag for the legacy Apps MCP path override.
@@ -1060,6 +1062,12 @@ pub const FEATURES: &[FeatureSpec] = &[
stage: Stage::Stable,
default_enabled: true,
},
FeatureSpec {
id: Feature::AppsRuntimeStateRefactor,
key: "apps_runtime_state_refactor",
stage: Stage::UnderDevelopment,
default_enabled: false,
},
FeatureSpec {
id: Feature::EnableMcpApps,
key: "enable_mcp_apps",