mirror of
https://github.com/openai/codex.git
synced 2026-10-20 08:19:50 +00:00
Use thread connections for enterprise MCP status listings (#55394)
## Why Enterprise-managed MCP authentication requires the thread's admitted connection. Bulk `mcpServerStatus/list` requests need to use that connection to report the enterprise server's catalog consistently with single-server requests. ## What changed - Read enterprise server metadata from the thread runtime when listing all MCP servers, preserving ordinary server discovery. - Check the expected server registration before discovery so edited or disabled registrations cannot reuse an old connection's catalog. ## Testing Add coverage for bulk and single-server status consistency, catalog metadata, and registration changes. Extend enterprise authentication lifecycle tests to cover bearer renewal, grant replacement, and logout during a pending exchange. Add a scenario asserting that enterprise sign-in failure preserves ordinary MCP tool schemas, calls, and model history. GitOrigin-RevId: 3bf3bd672511d0c753cd95ec777892f72bd6eb89
This commit is contained in:
@@ -401,7 +401,7 @@ impl McpRequestProcessor {
|
||||
(thread.as_ref(), params.server_name.as_deref())
|
||||
{
|
||||
thread
|
||||
.mcp_server_status_snapshot(server, detail)
|
||||
.mcp_server_status_snapshot(server, detail, /*expected_registration*/ None)
|
||||
.await
|
||||
.map_err(|error| {
|
||||
internal_error(format!("failed to read MCP server status: {error:#}"))
|
||||
@@ -432,7 +432,7 @@ impl McpRequestProcessor {
|
||||
),
|
||||
),
|
||||
};
|
||||
let snapshot = collect_mcp_server_status_snapshot_with_detail(
|
||||
let mut snapshot = collect_mcp_server_status_snapshot_with_detail(
|
||||
&mcp_config,
|
||||
auth.as_ref(),
|
||||
runtime_context,
|
||||
@@ -442,6 +442,54 @@ impl McpRequestProcessor {
|
||||
params.server_name.as_deref(),
|
||||
)
|
||||
.await;
|
||||
if mcp_config.xaa_enabled
|
||||
&& let Some(thread) = thread.as_ref()
|
||||
{
|
||||
// EMA can only use its admitted thread connection. Keep ordinary
|
||||
// status discovery unchanged and replace the threadless EMA results.
|
||||
for (name, server) in mcp_config.mcp_server_catalog.configured_servers() {
|
||||
if !matches!(server.auth, McpServerAuth::EmaAuth) {
|
||||
continue;
|
||||
}
|
||||
let (selected_config, mut selected) = thread
|
||||
.mcp_server_status_snapshot(
|
||||
&name,
|
||||
detail,
|
||||
mcp_config.mcp_server_catalog.server(&name),
|
||||
)
|
||||
.await
|
||||
.map_err(|error| {
|
||||
internal_error(format!("failed to read MCP server status: {error:#}"))
|
||||
})?;
|
||||
if selected_config.mcp_server_catalog.server(&name)
|
||||
!= mcp_config.mcp_server_catalog.server(&name)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
snapshot.server_infos.remove(&name);
|
||||
snapshot.server_infos.extend(selected.server_infos);
|
||||
snapshot.server_capabilities.remove(&name);
|
||||
if let Some(capabilities) = selected.server_capabilities.remove(&name) {
|
||||
snapshot
|
||||
.server_capabilities
|
||||
.insert(name.clone(), capabilities);
|
||||
}
|
||||
snapshot.tools_by_server.remove(&name);
|
||||
snapshot.tools_by_server.extend(selected.tools_by_server);
|
||||
snapshot.tool_list_meta.remove(&name);
|
||||
snapshot.tool_list_meta.extend(selected.tool_list_meta);
|
||||
snapshot.tools_errors.remove(&name);
|
||||
snapshot.tools_errors.extend(selected.tools_errors);
|
||||
snapshot.resources.remove(&name);
|
||||
snapshot.resources.extend(selected.resources);
|
||||
snapshot.resource_templates.remove(&name);
|
||||
snapshot
|
||||
.resource_templates
|
||||
.extend(selected.resource_templates);
|
||||
snapshot.auth_statuses.remove(&name);
|
||||
snapshot.auth_statuses.extend(selected.auth_statuses);
|
||||
}
|
||||
}
|
||||
(Arc::new(mcp_config), snapshot)
|
||||
};
|
||||
|
||||
|
||||
@@ -15,6 +15,7 @@ use codex_app_server::in_process::InProcessClientHandle;
|
||||
use codex_app_server::in_process::InProcessServerEvent;
|
||||
use codex_app_server::in_process::InProcessStartArgs;
|
||||
use codex_app_server_protocol::InitializeParams;
|
||||
use codex_app_server_protocol::McpServerStartupState;
|
||||
use codex_arg0::Arg0DispatchPaths;
|
||||
use codex_config::CloudConfigBundleLoader;
|
||||
use codex_config::LoaderOverrides;
|
||||
@@ -205,8 +206,120 @@ async fn enterprise_login_attempt_lifecycle() -> Result<()> {
|
||||
|
||||
keyring::set_default_credential_builder(Box::<TestKeyring>::default());
|
||||
let server = MockServer::builder().start().await;
|
||||
let idp_server = MockServer::start().await;
|
||||
let origin = server.uri();
|
||||
let issuer = format!("{origin}/idp");
|
||||
let issuer = format!("{}/idp", idp_server.uri());
|
||||
let resource = format!("{origin}/api/codex/ps/mcp");
|
||||
core_test_support::apps_test_server::AppsTestServer::mount(&server).await?;
|
||||
let catalog_meta = json!({"enterpriseCatalog": "calendar"});
|
||||
let response_meta = catalog_meta.clone();
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/api/codex/ps/mcp"))
|
||||
.and(wiremock::matchers::header_regex(
|
||||
"authorization",
|
||||
"^Bearer enterprise-resource-",
|
||||
))
|
||||
.and(body_partial_json(json!({"method": "tools/list"})))
|
||||
.respond_with(move |request: &wiremock::Request| {
|
||||
let body: Value = serde_json::from_slice(&request.body).unwrap();
|
||||
ResponseTemplate::new(200).set_body_json(json!({
|
||||
"jsonrpc": "2.0", "id": body["id"],
|
||||
"result": {
|
||||
"tools": [{"name": "calendar_list_events", "inputSchema": {"type": "object"}}],
|
||||
"_meta": response_meta,
|
||||
},
|
||||
}))
|
||||
})
|
||||
.with_priority(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
Mock::given(method("GET"))
|
||||
.and(path(
|
||||
"/.well-known/oauth-protected-resource/api/codex/ps/mcp",
|
||||
))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"resource": resource, "authorization_servers": [format!("{origin}/as")],
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/.well-known/oauth-authorization-server/as"))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
|
||||
"issuer": format!("{origin}/as"), "token_endpoint": format!("{origin}/as/token"),
|
||||
"grant_types_supported": ["urn:ietf:params:oauth:grant-type:jwt-bearer"],
|
||||
"token_endpoint_auth_methods_supported": ["none"],
|
||||
})))
|
||||
.mount(&server)
|
||||
.await;
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)?
|
||||
.as_secs();
|
||||
let jag = format!(
|
||||
"{}.{}.c2lnbmF0dXJl",
|
||||
URL_SAFE_NO_PAD.encode(r#"{"alg":"ES256","typ":"oauth-id-jag+jwt"}"#),
|
||||
URL_SAFE_NO_PAD.encode(
|
||||
json!({
|
||||
"iss": issuer, "aud": format!("{origin}/as"), "sub": "enterprise-user",
|
||||
"client_id": "mcp-client", "jti": "enterprise-grant", "iat": now,
|
||||
"exp": now + 3600, "resource": resource, "scope": "files.read",
|
||||
})
|
||||
.to_string()
|
||||
),
|
||||
);
|
||||
let expected_jag = jag.clone();
|
||||
let resource_exchanges = Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
||||
let exchanges = Arc::clone(&resource_exchanges);
|
||||
Mock::given(method("POST")).and(path("/as/token"))
|
||||
.respond_with(move |request: &wiremock::Request| {
|
||||
let fields = url::form_urlencoded::parse(&request.body).into_owned().collect::<HashMap<_, _>>();
|
||||
assert_eq!(fields.get("assertion"), Some(&expected_jag));
|
||||
ResponseTemplate::new(200).set_body_json(json!({
|
||||
"access_token": format!("enterprise-resource-{}", exchanges.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1),
|
||||
"token_type": "Bearer", "expires_in": 3600,
|
||||
}))
|
||||
}).mount(&server).await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/api/codex/ps/mcp"))
|
||||
.and(header("authorization", "Bearer enterprise-resource-1"))
|
||||
.and(body_partial_json(json!({"method": "tools/call"})))
|
||||
.respond_with(ResponseTemplate::new(401))
|
||||
.with_priority(1)
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
let token_resource = resource.clone();
|
||||
let pending_exchange = Arc::new(Mutex::new(
|
||||
None::<(
|
||||
tokio::sync::oneshot::Sender<()>,
|
||||
std::sync::mpsc::Receiver<()>,
|
||||
)>,
|
||||
));
|
||||
let exchange_gate = Arc::clone(&pending_exchange);
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/idp/token"))
|
||||
.and(wiremock::matchers::body_string_contains(
|
||||
"grant-type%3Atoken-exchange",
|
||||
))
|
||||
.respond_with(move |request: &wiremock::Request| {
|
||||
let fields = url::form_urlencoded::parse(&request.body)
|
||||
.into_owned()
|
||||
.collect::<HashMap<_, _>>();
|
||||
assert_eq!(
|
||||
fields.get("subject_token").map(String::as_str),
|
||||
Some("enterprise-refresh")
|
||||
);
|
||||
if let Some((arrived, release)) = exchange_gate.lock().unwrap().take() {
|
||||
let _ = arrived.send(());
|
||||
let _ = release.recv();
|
||||
}
|
||||
ResponseTemplate::new(200).set_body_json(json!({
|
||||
"access_token": jag, "issued_token_type": "urn:ietf:params:oauth:token-type:id-jag",
|
||||
"token_type": "N_A", "resource": token_resource, "scope": "files.read",
|
||||
}))
|
||||
})
|
||||
.with_priority(1)
|
||||
.mount(&idp_server)
|
||||
.await;
|
||||
Mock::given(method("GET"))
|
||||
.and(path("/backend-api/wham/config/bundle"))
|
||||
.respond_with(
|
||||
@@ -223,7 +336,7 @@ async fn enterprise_login_attempt_lifecycle() -> Result<()> {
|
||||
"token_endpoint": format!("{issuer}/token"),
|
||||
"token_endpoint_auth_methods_supported": ["none"],
|
||||
})))
|
||||
.mount(&server)
|
||||
.mount(&idp_server)
|
||||
.await;
|
||||
let assertion = format!(
|
||||
"{}.{}.signature",
|
||||
@@ -239,7 +352,7 @@ async fn enterprise_login_attempt_lifecycle() -> Result<()> {
|
||||
"access_token":"enterprise-access", "refresh_token":"enterprise-refresh",
|
||||
"id_token":assertion, "token_type":"Bearer",
|
||||
})))
|
||||
.mount(&server)
|
||||
.mount(&idp_server)
|
||||
.await;
|
||||
|
||||
let home = TempDir::new()?;
|
||||
@@ -262,11 +375,15 @@ issuer = "{issuer}"
|
||||
client_id = "idp-client"
|
||||
|
||||
[mcp_servers.enterprise]
|
||||
url = "{origin}/mcp"
|
||||
url = "{resource}"
|
||||
auth = "ema_auth"
|
||||
scopes = ["files.read"]
|
||||
oauth = {{ client_id = "mcp-client", authorization_server_issuer = "{origin}/as" }}
|
||||
|
||||
[mcp_servers.ordinary]
|
||||
url = "{resource}"
|
||||
http_headers = {{ Authorization = "Bearer ordinary-access" }}
|
||||
|
||||
[mcp_servers.disabled-enterprise]
|
||||
url = "{origin}/mcp"
|
||||
enabled = false
|
||||
@@ -455,6 +572,171 @@ oauth = {{ client_id = "mcp-client", authorization_server_issuer = "{origin}/as"
|
||||
);
|
||||
}
|
||||
|
||||
// A fresh thread uses the saved enterprise grant and renews it for MCP requests.
|
||||
let connected = enterprise_rpc(&client, "thread/start", json!({})).await?;
|
||||
let connected_id = connected["thread"]["id"]
|
||||
.as_str()
|
||||
.context("connected thread ID")?;
|
||||
let call = |server: &str| {
|
||||
json!({
|
||||
"threadId": connected_id, "server": server,
|
||||
"tool": "calendar_list_events", "arguments": {"query": "hello"},
|
||||
})
|
||||
};
|
||||
// A rejected operation is not replayed. The next explicit call renews its bearer.
|
||||
assert_eq!(
|
||||
enterprise_rpc(&client, "mcpServer/tool/call", call("enterprise")).await?["isError"],
|
||||
true
|
||||
);
|
||||
let enterprise_result =
|
||||
enterprise_rpc(&client, "mcpServer/tool/call", call("enterprise")).await?;
|
||||
let ordinary_result = enterprise_rpc(&client, "mcpServer/tool/call", call("ordinary")).await?;
|
||||
assert_eq!(enterprise_result, ordinary_result);
|
||||
let bulk_status = enterprise_rpc(
|
||||
&client,
|
||||
"mcpServerStatus/list",
|
||||
json!({"threadId": connected_id}),
|
||||
)
|
||||
.await?;
|
||||
let statuses = bulk_status["data"].as_array().context("server statuses")?;
|
||||
for name in ["enterprise", "ordinary"] {
|
||||
let selected_status = enterprise_rpc(
|
||||
&client,
|
||||
"mcpServerStatus/list",
|
||||
json!({"threadId": connected_id, "serverName": name}),
|
||||
)
|
||||
.await?;
|
||||
let status = statuses
|
||||
.iter()
|
||||
.find(|status| status["name"] == name)
|
||||
.context("connected server status")?;
|
||||
assert_eq!(status, &selected_status["data"][0]);
|
||||
assert!(!status["tools"]["calendar_list_events"].is_null());
|
||||
if name == "enterprise" {
|
||||
assert_eq!(status["toolListMeta"], catalog_meta);
|
||||
}
|
||||
}
|
||||
// Bulk inventory must not attach the old connection's catalog to an edited registration.
|
||||
let inventory_request_count = |requests: &[wiremock::Request]| {
|
||||
requests
|
||||
.iter()
|
||||
.filter(|request| {
|
||||
request.headers.get("authorization").is_some_and(|value| {
|
||||
value.as_bytes().starts_with(b"Bearer enterprise-resource-")
|
||||
}) && serde_json::from_slice::<Value>(&request.body).is_ok_and(|body| {
|
||||
matches!(
|
||||
body["method"].as_str(),
|
||||
Some("resources/list" | "resources/templates/list")
|
||||
)
|
||||
})
|
||||
})
|
||||
.count()
|
||||
};
|
||||
let inventory_requests_before =
|
||||
inventory_request_count(&server.received_requests().await.context("requests")?);
|
||||
assert!(inventory_requests_before > 0);
|
||||
let config_path = home.path().join("config.toml");
|
||||
let saved_config = std::fs::read_to_string(&config_path)?;
|
||||
let mut edited_config: toml::Value = toml::from_str(&saved_config)?;
|
||||
for (url, enabled) in [
|
||||
("https://replacement.example/mcp", true),
|
||||
(resource.as_str(), false),
|
||||
] {
|
||||
let enterprise = edited_config["mcp_servers"]["enterprise"]
|
||||
.as_table_mut()
|
||||
.context("enterprise config")?;
|
||||
enterprise.insert("url".into(), toml::Value::String(url.into()));
|
||||
enterprise.insert("enabled".into(), toml::Value::Boolean(enabled));
|
||||
std::fs::write(&config_path, toml::to_string(&edited_config)?)?;
|
||||
let changed = enterprise_rpc(
|
||||
&client,
|
||||
"mcpServerStatus/list",
|
||||
json!({"threadId": connected_id}),
|
||||
)
|
||||
.await?;
|
||||
assert_eq!(
|
||||
inventory_request_count(&server.received_requests().await.context("requests")?),
|
||||
inventory_requests_before,
|
||||
"bulk status must not query the previous enterprise registration",
|
||||
);
|
||||
let changed = changed["data"].as_array().context("changed statuses")?;
|
||||
let enterprise = changed
|
||||
.iter()
|
||||
.find(|status| status["name"] == "enterprise")
|
||||
.context("changed enterprise status")?;
|
||||
assert_eq!(
|
||||
enterprise["httpOrigin"],
|
||||
url::Url::parse(url)?.origin().ascii_serialization()
|
||||
);
|
||||
assert_eq!(enterprise["tools"], json!({}));
|
||||
assert_eq!(enterprise["resources"], json!([]));
|
||||
assert_eq!(enterprise["resourceTemplates"], json!([]));
|
||||
for field in ["serverInfo", "serverCapabilities", "toolListMeta"] {
|
||||
assert!(enterprise[field].is_null(), "stale {field}: {enterprise}");
|
||||
}
|
||||
assert_eq!(
|
||||
changed.iter().find(|status| status["name"] == "ordinary"),
|
||||
statuses.iter().find(|status| status["name"] == "ordinary"),
|
||||
);
|
||||
}
|
||||
std::fs::write(config_path, saved_config)?;
|
||||
let authorization = server
|
||||
.received_requests()
|
||||
.await
|
||||
.context("requests")?
|
||||
.into_iter()
|
||||
.filter(|request| {
|
||||
serde_json::from_slice::<Value>(&request.body)
|
||||
.ok()
|
||||
.is_some_and(|body| body["method"] == "tools/call")
|
||||
})
|
||||
.map(|request| {
|
||||
request.headers["authorization"]
|
||||
.to_str()
|
||||
.unwrap()
|
||||
.to_owned()
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(
|
||||
authorization,
|
||||
[
|
||||
"Bearer enterprise-resource-1",
|
||||
"Bearer enterprise-resource-2",
|
||||
"Bearer ordinary-access"
|
||||
]
|
||||
);
|
||||
assert_eq!(
|
||||
resource_exchanges.load(std::sync::atomic::Ordering::SeqCst),
|
||||
2
|
||||
);
|
||||
|
||||
// Replacing the enterprise grant fences old threads, including cached bearers.
|
||||
let relogin = enterprise_rpc(
|
||||
&client,
|
||||
"mcpServer/oauth/login",
|
||||
json!({"name":"enterprise","threadId":connected_id}),
|
||||
)
|
||||
.await?;
|
||||
browser
|
||||
.get(enterprise_callback(&relogin, &issuer, callback_port)?)
|
||||
.header(http::header::CONNECTION, "close")
|
||||
.send()
|
||||
.await?;
|
||||
assert_eq!(
|
||||
enterprise_completion(&mut client).await?,
|
||||
json!({
|
||||
"name":"enterprise","threadId":connected_id,"loginId":relogin["loginId"],"success":true,
|
||||
})
|
||||
);
|
||||
assert_eq!(
|
||||
enterprise_rpc(&client, "mcpServer/tool/call", call("enterprise")).await?["isError"],
|
||||
true
|
||||
);
|
||||
assert_eq!(
|
||||
enterprise_rpc(&client, "mcpServer/tool/call", call("ordinary")).await?,
|
||||
ordinary_result
|
||||
);
|
||||
|
||||
let stalled = enterprise_rpc(
|
||||
&client,
|
||||
"mcpServer/oauth/login",
|
||||
@@ -469,6 +751,11 @@ oauth = {{ client_id = "mcp-client", authorization_server_issuer = "{origin}/as"
|
||||
// It cannot finish draining the declared body while this socket stays open.
|
||||
let mut response = [0; 256];
|
||||
assert!(timeout(DEFAULT_READ_TIMEOUT, unfinished.read(&mut response)).await?? > 0);
|
||||
let (exchange_arrived, exchange_wait) = tokio::sync::oneshot::channel();
|
||||
let (release_exchange, exchange_release) = std::sync::mpsc::channel();
|
||||
*pending_exchange.lock().unwrap() = Some((exchange_arrived, exchange_release));
|
||||
let pending = enterprise_rpc(&client, "thread/start", json!({})).await?;
|
||||
timeout(DEFAULT_READ_TIMEOUT, exchange_wait).await??;
|
||||
let logout = timeout(
|
||||
std::time::Duration::from_secs(15),
|
||||
enterprise_rpc(&client, "account/logout", json!({})),
|
||||
@@ -476,8 +763,60 @@ oauth = {{ client_id = "mcp-client", authorization_server_issuer = "{origin}/as"
|
||||
.await;
|
||||
// Release the worker even when the regression fails, so child shutdown cannot hang.
|
||||
drop(unfinished);
|
||||
let _ = release_exchange.send(());
|
||||
logout.context("an unfinished callback must not prevent primary logout")??;
|
||||
// Observe the original startup's completion before a tool call can refresh it.
|
||||
let mut startup_finished = false;
|
||||
let mut login_completion = None;
|
||||
timeout(DEFAULT_READ_TIMEOUT, async {
|
||||
while !startup_finished || login_completion.is_none() {
|
||||
if let InProcessServerEvent::ServerNotification(notification) =
|
||||
client.next_event().await.context("logout notification")?
|
||||
{
|
||||
match *notification {
|
||||
ServerNotification::McpServerStatusUpdated(notification)
|
||||
if notification.thread_id.as_deref()
|
||||
== pending["thread"]["id"].as_str()
|
||||
&& notification.name == "enterprise"
|
||||
&& notification.status != McpServerStartupState::Starting =>
|
||||
{
|
||||
startup_finished = true;
|
||||
}
|
||||
ServerNotification::McpServerOauthLoginCompleted(notification) => {
|
||||
login_completion = Some(serde_json::to_value(notification)?);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
anyhow::Ok(())
|
||||
})
|
||||
.await??;
|
||||
assert_eq!(
|
||||
login_completion.context("enterprise login completion")?,
|
||||
json!({"name":"enterprise","threadId":thread_id,"loginId":stalled["loginId"],"success":false,"error":"Enterprise sign-in failed."})
|
||||
);
|
||||
assert_eq!(
|
||||
resource_exchanges.load(std::sync::atomic::Ordering::SeqCst),
|
||||
2
|
||||
);
|
||||
assert!(
|
||||
enterprise_rpc(
|
||||
&client,
|
||||
"mcpServer/tool/call",
|
||||
json!({
|
||||
"threadId": pending["thread"]["id"], "server": "enterprise",
|
||||
"tool": "calendar_list_events", "arguments": {},
|
||||
})
|
||||
)
|
||||
.await
|
||||
.is_err()
|
||||
);
|
||||
assert!(!home.path().join("auth.json").exists());
|
||||
assert_eq!(
|
||||
enterprise_rpc(&client, "mcpServer/tool/call", call("ordinary")).await?,
|
||||
ordinary_result
|
||||
);
|
||||
assert!(
|
||||
stored_oauth_credentials(
|
||||
&idp.credential_name(
|
||||
@@ -490,10 +829,6 @@ oauth = {{ client_id = "mcp-client", authorization_server_issuer = "{origin}/as"
|
||||
)?
|
||||
.is_none()
|
||||
);
|
||||
assert_eq!(
|
||||
enterprise_completion(&mut client).await?,
|
||||
json!({"name":"enterprise","threadId":thread_id,"loginId":stalled["loginId"],"success":false,"error":"Enterprise sign-in failed."})
|
||||
);
|
||||
client.shutdown().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -7,6 +7,7 @@ use super::McpRuntimeContext;
|
||||
use crate::McpConfig;
|
||||
use crate::McpServerStatusSnapshot;
|
||||
use crate::McpSnapshotDetail;
|
||||
use crate::ResolvedMcpServer;
|
||||
use crate::mcp::collect_mcp_server_status_snapshot_from_manager;
|
||||
use crate::mcp::compute_auth_statuses;
|
||||
use crate::mcp::effective_mcp_servers;
|
||||
@@ -14,10 +15,12 @@ use crate::mcp::effective_mcp_servers;
|
||||
impl McpRuntime {
|
||||
/// Reuses the selected connection and catalog without starting unrelated servers.
|
||||
/// Configuration and metadata belong to the same captured runtime publication.
|
||||
/// An expected registration excludes mismatched connections before discovery.
|
||||
pub async fn server_status_snapshot(
|
||||
&self,
|
||||
server: &str,
|
||||
detail: McpSnapshotDetail,
|
||||
expected_registration: Option<&ResolvedMcpServer>,
|
||||
runtime_context: &McpRuntimeContext,
|
||||
) -> anyhow::Result<(Arc<McpConfig>, McpServerStatusSnapshot)> {
|
||||
let current = self.current.load_full();
|
||||
@@ -32,7 +35,11 @@ impl McpRuntime {
|
||||
.ok_or_else(|| anyhow::anyhow!("MCP runtime is not configured"))?,
|
||||
);
|
||||
let mut servers = effective_mcp_servers(&config, current.auth.as_ref());
|
||||
servers.retain(|name, _| name == server);
|
||||
servers.retain(|name, _| {
|
||||
name == server
|
||||
&& expected_registration
|
||||
.is_none_or(|expected| config.mcp_server_catalog.server(name) == Some(expected))
|
||||
});
|
||||
let auth_statuses = compute_auth_statuses(
|
||||
servers.iter(),
|
||||
config.mcp_oauth_credentials_store_mode,
|
||||
|
||||
@@ -1152,6 +1152,7 @@ impl CodexThread {
|
||||
&self,
|
||||
server: &str,
|
||||
detail: codex_mcp::McpSnapshotDetail,
|
||||
expected_registration: Option<&codex_mcp::ResolvedMcpServer>,
|
||||
) -> anyhow::Result<(
|
||||
Arc<codex_mcp::McpConfig>,
|
||||
codex_mcp::McpServerStatusSnapshot,
|
||||
@@ -1161,7 +1162,7 @@ impl CodexThread {
|
||||
self.session
|
||||
.services
|
||||
.mcp_runtime
|
||||
.server_status_snapshot(server, detail, &runtime_context)
|
||||
.server_status_snapshot(server, detail, expected_registration, &runtime_context)
|
||||
.await
|
||||
}
|
||||
|
||||
|
||||
@@ -146,6 +146,9 @@ mod indirect_namespace_prefixes;
|
||||
#[path = "scenarios_mcp_resource_messages.rs"]
|
||||
mod mcp_resource_messages;
|
||||
|
||||
#[path = "scenarios_mcp_ema.rs"]
|
||||
mod mcp_ema;
|
||||
|
||||
#[path = "scenarios_guardian_agent_messages_tests.rs"]
|
||||
mod guardian_agent_messages;
|
||||
|
||||
@@ -294,7 +297,8 @@ fn configure_scenario_catalog(config: &mut Config) {
|
||||
stack.requirements().clone(),
|
||||
stack.requirements_toml().clone(),
|
||||
)
|
||||
.expect("fixture config layers");
|
||||
.expect("fixture config layers")
|
||||
.with_cloud_config_binding(stack.cloud_config_binding().cloned());
|
||||
config.model_catalog = Some(bundled_models_response().expect("bundled model catalog"));
|
||||
config.cloud_skill_enabled = false;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
//! Failed enterprise authentication leaves ordinary MCP calls and model history unchanged.
|
||||
|
||||
use anyhow::Result;
|
||||
use codex_config::test_support::CloudConfigBundleFixture;
|
||||
use codex_login::CodexAuth;
|
||||
use codex_protocol::openai_models::ToolMode;
|
||||
use codex_protocol::protocol::EventMsg;
|
||||
use codex_protocol::protocol::McpStartupStatus;
|
||||
use core_test_support::apps_test_server::AppsTestServer;
|
||||
use core_test_support::context_snapshot;
|
||||
use core_test_support::context_snapshot::ContextSnapshotOptions;
|
||||
use core_test_support::responses;
|
||||
use core_test_support::skip_if_no_network;
|
||||
use core_test_support::test_codex::test_codex;
|
||||
use core_test_support::wait_for_event_match;
|
||||
use pretty_assertions::assert_eq;
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
async fn enterprise_failure_preserves_ordinary_mcp_history() -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
let mut histories = Vec::new();
|
||||
for enabled in [false, true] {
|
||||
let server = responses::start_mock_server().await;
|
||||
let mcp = responses::start_mock_server().await;
|
||||
AppsTestServer::mount(&mcp).await?;
|
||||
let origin = mcp.uri();
|
||||
let policy = format!(
|
||||
r#"
|
||||
[features]
|
||||
use_xaa = {enabled}
|
||||
apps = false
|
||||
secret_auth_storage = false
|
||||
[mcp_enterprise_managed_auth.idp]
|
||||
issuer = "{origin}/idp"
|
||||
client_id = "enterprise-client"
|
||||
[mcp_servers.enterprise]
|
||||
url = "{origin}/enterprise"
|
||||
auth = "ema_auth"
|
||||
[mcp_servers.ordinary]
|
||||
url = "{origin}/api/codex/ps/mcp"
|
||||
http_headers = {{ Authorization = "Bearer ordinary-access" }}
|
||||
"#
|
||||
);
|
||||
let test = test_codex()
|
||||
.with_auth(CodexAuth::from_api_key("test-api-key"))
|
||||
.with_cloud_config_bundle(CloudConfigBundleFixture::loader_with_enterprise_config(
|
||||
policy,
|
||||
))
|
||||
.with_config(super::configure_scenario_catalog)
|
||||
.with_model_info_override("gpt-6-astra", |model| {
|
||||
model.tool_mode = Some(ToolMode::Direct);
|
||||
model.use_responses_lite = false;
|
||||
model.supports_search_tool = false;
|
||||
})
|
||||
.build_with_auto_env(&server)
|
||||
.await?;
|
||||
if enabled {
|
||||
let error = wait_for_event_match(&test.codex, |event| match event {
|
||||
EventMsg::McpStartupUpdate(update) if update.server == "enterprise" => {
|
||||
match &update.status {
|
||||
McpStartupStatus::Failed { error, .. } => Some(error.clone()),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
insta::assert_snapshot!("enterprise_sign_in_required", error);
|
||||
}
|
||||
let startup = wait_for_event_match(&test.codex, |event| match event {
|
||||
EventMsg::McpStartupComplete(summary) => Some(summary.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert_eq!(startup.ready, ["ordinary"]);
|
||||
assert_eq!(
|
||||
startup
|
||||
.failed
|
||||
.into_iter()
|
||||
.map(|failure| failure.server)
|
||||
.collect::<Vec<_>>(),
|
||||
if enabled {
|
||||
vec!["enterprise"]
|
||||
} else {
|
||||
Vec::new()
|
||||
}
|
||||
);
|
||||
let mock = responses::mount_sse_sequence(
|
||||
&server,
|
||||
vec![
|
||||
responses::sse(vec![
|
||||
responses::ev_function_call_with_namespace(
|
||||
"calendar-call",
|
||||
"mcp__ordinary",
|
||||
"calendar_list_events",
|
||||
"{}",
|
||||
),
|
||||
responses::ev_completed("call-response"),
|
||||
]),
|
||||
responses::sse(vec![
|
||||
responses::ev_assistant_message("answer", "Here are your calendar events."),
|
||||
responses::ev_completed("answer-response"),
|
||||
]),
|
||||
],
|
||||
)
|
||||
.await;
|
||||
test.submit_text_turn("List my calendar events.").await?;
|
||||
let requests = mock.requests();
|
||||
assert_eq!(requests.len(), 2);
|
||||
histories.push(context_snapshot::format_request_history_snapshot(
|
||||
"Ordinary MCP keeps the same tool schemas, calls, and history when enterprise authentication fails.",
|
||||
&requests,
|
||||
&ContextSnapshotOptions::default().rewrite_known_segments().include_request_settings(),
|
||||
));
|
||||
assert!(
|
||||
mcp.received_requests()
|
||||
.await
|
||||
.expect("requests")
|
||||
.iter()
|
||||
.all(|request| {
|
||||
!request.url.path().contains("enterprise")
|
||||
&& !request.url.path().contains("idp")
|
||||
})
|
||||
);
|
||||
test.codex.shutdown_and_wait().await?;
|
||||
}
|
||||
assert_eq!(histories[0], histories[1]);
|
||||
insta::assert_snapshot!(
|
||||
"enterprise_failure_preserves_ordinary_mcp_history",
|
||||
histories[1]
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
+62
@@ -0,0 +1,62 @@
|
||||
---
|
||||
source: core/tests/suite/scenarios_mcp_ema.rs
|
||||
assertion_line: 125
|
||||
expression: "histories[1]"
|
||||
---
|
||||
Scenario: Ordinary MCP keeps the same tool schemas, calls, and history when enterprise authentication fails.
|
||||
|
||||
## Window 1
|
||||
Settings:
|
||||
include: ["reasoning.encrypted_content"]
|
||||
model: "gpt-6-astra"
|
||||
parallel_tool_calls: true
|
||||
prompt_cache_key: "<PROMPT_CACHE_KEY 1>"
|
||||
reasoning: {"effort":"low"}
|
||||
store: false
|
||||
stream: true
|
||||
text: {"verbosity":"low"}
|
||||
tool_choice: "auto"
|
||||
tools (13; hash=75ef17d8adc1b3db):
|
||||
- function/exec_command: Runs a command in a PTY, returning output or a session ID for ongoing interaction.; args=[cmd, justification, login, max_output_tokens, prefix_rule, sandbox_permissions, shell, tty, workdir, yield_time_ms]; hash=c04108e51a8548c5
|
||||
- function/write_stdin: Writes characters to an existing unified exec session and returns recent output.; args=[chars, max_output_tokens, session_id, yield_time_ms]; hash=7fb62d8a3e2df4e2
|
||||
- function/list_mcp_resources: Lists resources provided by MCP servers. Resources allow servers to share data that provides cont...; args=[cursor, server]; hash=6792edaff13561bd
|
||||
- function/list_mcp_resource_templates: Lists resource templates provided by MCP servers. Parameterized resource templates allow servers ...; args=[cursor, server]; hash=60f17c90ef7a4ebb
|
||||
- function/read_mcp_resource: Read a specific resource from an MCP server given the server name and resource URI.; args=[server, uri]; hash=24246896b39008ae
|
||||
- function/request_user_input: Request user input for one to three short questions and wait for the response. This tool is only ...; args=[questions]; hash=64a729b25c424fc6
|
||||
- function/request_user_input_async: Ask the user one or more questions during ongoing work. Use this tool only to request missing inf...; args=[questions]; hash=d6d74f33db330162
|
||||
- namespace/clock: Tools for reading and waiting on time.; hash=a80435c7a3a3943f
|
||||
- function/curr_time
|
||||
- function/sleep
|
||||
- custom/apply_patch: The `apply_patch` tool can be used to edit files. This is a FREEFORM tool, so do not wrap the pat...; hash=dea9f12b0111a4ba
|
||||
- function/view_image: View a local image file from the filesystem when visual inspection is needed. Use this for images...; args=[detail, path]; hash=3b766d5bdec5d50e
|
||||
- namespace/collaboration: Tools for spawning and managing sub-agents.; hash=2c8d6295978dbe63
|
||||
- function/followup_task
|
||||
- function/interrupt_agent
|
||||
- function/list_agents
|
||||
- function/send_message
|
||||
- function/spawn_agent
|
||||
- function/wait_agent
|
||||
- namespace/mcp__ordinary: Tools in the mcp__ordinary namespace.; hash=101b30f5cdc3cae0
|
||||
- function/calendar_create_event
|
||||
- function/calendar_extract_text
|
||||
- function/calendar_list_events
|
||||
- web_search; hash=f103e327c6870a46
|
||||
-- request 1 (turn) --
|
||||
00:message/developer:
|
||||
<MODEL_INSTRUCTIONS>
|
||||
01:message/developer[2]:
|
||||
[01] <PERMISSIONS_INSTRUCTIONS>
|
||||
[02] <COLLABORATION_MODE>
|
||||
02:message/developer:
|
||||
<MULTI_AGENT_ROLE>
|
||||
03:message/developer:
|
||||
<MULTI_AGENT_MODE>
|
||||
04:message/user:
|
||||
<ENVIRONMENT_CONTEXT>
|
||||
05:message/user:
|
||||
List my calendar events.
|
||||
-- request 2 (turn) --
|
||||
06:function_call/mcp__ordinary.calendar_list_events:{}
|
||||
07:function_call_output:Wall time: <DURATION> seconds
|
||||
Output:
|
||||
{"_codex_apps":null}
|
||||
+6
@@ -0,0 +1,6 @@
|
||||
---
|
||||
source: core/tests/suite/scenarios_mcp_ema.rs
|
||||
assertion_line: 65
|
||||
expression: error
|
||||
---
|
||||
The enterprise MCP server is not logged in. Sign in to your enterprise provider in Codex, then start a new thread.
|
||||
Reference in New Issue
Block a user