Start dormant MCP servers for hook calls (#55263)

## Why

MCP tool hooks need to invoke model-hidden tools even when a subagent's cached server has not started. Their timeout must also cover server startup.

## What changed

- Let cached servers with allowed model-hidden tools remain dormant until called.
- Wait for the selected server when executing an MCP hook, leaving unrelated cached servers dormant, and apply the hook timeout to the entire operation.
- Capture tool namespace attribution after startup and retain it only while catalog revisions remain unchanged.
- Validate the server, environment, and tool filter before starting a server or capturing attribution.

## Testing

Extend integration coverage for cached subagent hooks under both startup readiness modes, startup timeouts without context injection, and namespace attribution for hook and token-budget context while keeping tools hidden from the model. Verify that a direct call starts a cached server whose tools are hidden from model exposure.

GitOrigin-RevId: 7a54b69aa2515722aeb925e03e1667bf63aa67ca
This commit is contained in:
Charlie Marsh
2026-10-19 11:40:15 +00:00
committed by copyberry
parent 71ab2ea77b
commit cc44ffb3e2
6 changed files with 413 additions and 52 deletions
+25 -13
View File
@@ -207,7 +207,8 @@ impl McpServerView {
|| match self.startup_readiness {
McpStartupReadiness::Connection => !tools.is_empty(),
McpStartupReadiness::Catalog => tools.iter().any(|tool| {
self.tool_filter.allows(&tool.tool.name) && tool_is_model_visible(tool)
self.tool_filter.allows(&tool.tool.name)
&& (self.connection.startup_is_dormant() || tool_is_model_visible(tool))
}),
}
}
@@ -700,10 +701,9 @@ impl McpConnectionSet {
.as_ref()
.and_then(McpToolCatalogCacheContext::current_tools)
.is_some_and(|tools| {
tools.into_iter().any(|tool| {
configured_tool_filter.allows(&tool.tool.name)
&& tool_is_model_visible(&tool)
})
tools
.into_iter()
.any(|tool| configured_tool_filter.allows(&tool.tool.name))
});
let (startup_trigger, startup_receiver) = if defer_startup {
let (trigger, receiver) = watch::channel(false);
@@ -1036,18 +1036,13 @@ impl McpConnectionSet {
}
}
/// Invoke the tool indicated by the (server, tool) pair.
#[allow(clippy::too_many_arguments)]
pub async fn call_tool(
/// Rejects invalid host calls before they can start a server or capture attribution.
pub(crate) fn validate_tool_call(
&self,
server: &str,
tool: &str,
environment_id: Option<&str>,
arguments: Option<serde_json::Value>,
mut meta: Option<serde_json::Value>,
requested_timeout: Option<Duration>,
wait_for_server: bool,
) -> Result<CallToolResult> {
) -> Result<()> {
let view = self
.servers
.get(server)
@@ -1065,6 +1060,23 @@ impl McpConnectionSet {
"tool '{tool}' is disabled for MCP server '{server}'"
));
}
Ok(())
}
/// Invoke the tool indicated by the (server, tool) pair.
#[allow(clippy::too_many_arguments)]
pub async fn call_tool(
&self,
server: &str,
tool: &str,
environment_id: Option<&str>,
arguments: Option<serde_json::Value>,
mut meta: Option<serde_json::Value>,
requested_timeout: Option<Duration>,
wait_for_server: bool,
) -> Result<CallToolResult> {
self.validate_tool_call(server, tool, environment_id)?;
let view = &self.servers[server];
let client = if wait_for_server {
view.connection
.client()
+34 -10
View File
@@ -657,9 +657,9 @@ impl McpRuntime {
.await
}
/// Calls a tool for the host, using a cached namespace only if its catalogs stay unchanged.
/// Missing or changed catalogs leave the namespace unknown. Attribution never starts
/// servers or refreshes their catalogs.
/// Calls a tool for the host, retaining attribution while its catalogs stay unchanged.
/// Waiting calls capture a binding after the selected server starts; non-waiting calls
/// use the cached binding. Missing or changed catalogs leave the namespace unknown.
#[allow(clippy::too_many_arguments)]
pub async fn latest_call_tool(
&self,
@@ -672,12 +672,36 @@ impl McpRuntime {
wait_for_server: bool,
) -> anyhow::Result<McpToolCallResult> {
let current = self.current.load_full();
let cached = current
.cached_binding
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.as_ref()
.and_then(|cached| Some((cached.catalog_revisions.clone(), cached.binding.upgrade()?)));
current
.connections
.validate_tool_call(server, tool, environment_id)?;
let attribution = if wait_for_server
&& current.connections.wait_for_server_startup(server).await
&& let Some(revisions) = current
.connections
.stable_catalog_revisions(
/*required_servers*/ &[],
/*required_plugins*/ &HashSet::new(),
)
.await
{
Self::binding_from_published_runtime(
Arc::clone(&current),
/*required_servers*/ &[],
/*required_plugins*/ &HashSet::new(),
)
.await
.map(|binding| (revisions, binding))
} else {
current
.cached_binding
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.as_ref()
.and_then(|cached| {
Some((cached.catalog_revisions.clone(), cached.binding.upgrade()?))
})
};
let result = current
.connections
.call_tool(
@@ -694,7 +718,7 @@ impl McpRuntime {
// Check after dispatch so a refresh before or during the call invalidates attribution.
// Revisions include client identity and increase on catalog changes. Reading them
// does not start servers or refresh catalogs; all catalogs matter for name collisions.
let namespace = if let Some((revisions, binding)) = cached
let namespace = if let Some((revisions, binding)) = attribution
&& current
.connections
.stable_catalog_revisions(
+9 -6
View File
@@ -1,5 +1,6 @@
use std::sync::Arc;
use anyhow::Context;
use anyhow::bail;
use codex_hooks::HookMcpCall;
use codex_hooks::HookMcpExecutor;
@@ -25,18 +26,20 @@ impl HookMcpExecutor for CoreHookMcpExecutor {
Value::String(self.thread_id.to_string()),
);
let call_result = self
.runtime
.latest_call_tool(
let call_result = tokio::time::timeout(
call.timeout,
self.runtime.latest_call_tool(
&call.server,
&call.tool,
call.environment_id.as_deref(),
Some(Value::Object(call.input)),
Some(Value::Object(metadata)),
Some(call.timeout),
/*wait_for_server*/ false,
)
.await?;
/*wait_for_server*/ true,
),
)
.await
.context("MCP hook timed out")??;
let result = call_result.result;
let text = result
.content
+231 -10
View File
@@ -5,14 +5,24 @@ use std::time::Duration;
use anyhow::Context;
use anyhow::Result;
use codex_config::McpStartupReadiness;
use codex_config::types::AppToolApproval;
use codex_config::types::ApprovalsReviewer;
use codex_config::types::McpServerConfig;
use codex_config::types::McpServerTransportConfig;
use codex_core::StartThreadOptions;
use codex_core::TurnInputRequest;
use codex_core::config::Config;
use codex_features::Feature;
use codex_protocol::mcp::McpServerConnectionStatus;
use codex_protocol::models::ContentItemMetadata;
use codex_protocol::models::PermissionProfile;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::HookRunStatus;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::user_input::UserInput;
use codex_utils_path_uri::LegacyAppPathString;
use core_test_support::hooks::trust_discovered_hooks;
use core_test_support::responses::ev_assistant_message;
@@ -21,12 +31,14 @@ use core_test_support::responses::ev_function_call_with_namespace;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_sse_once;
use core_test_support::responses::mount_sse_sequence;
use core_test_support::responses::namespace_child_tool;
use core_test_support::responses::sse;
use core_test_support::responses::start_mock_server;
use core_test_support::skip_if_no_network;
use core_test_support::skip_if_wine_exec;
use core_test_support::stdio_server_bin;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use core_test_support::wait_for_mcp_server;
use pretty_assertions::assert_eq;
use serde_json::Value;
@@ -507,8 +519,12 @@ async fn mcp_tool_hook_interpolates_prompt_and_runs_without_tool_approval() -> R
Ok(())
}
#[test_case::test_case(McpStartupReadiness::Connection; "connection")]
#[test_case::test_case(McpStartupReadiness::Catalog; "catalog")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mcp_tool_hook_passes_thread_metadata_to_model_hidden_tools() -> Result<()> {
async fn mcp_tool_hook_starts_cached_model_hidden_tools(
startup_readiness: McpStartupReadiness,
) -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
@@ -523,6 +539,7 @@ async fn mcp_tool_hook_passes_thread_metadata_to_model_hidden_tools() -> Result<
.await;
let rmcp_test_server_bin = stdio_server_bin()?;
let test = test_codex()
.with_model_info_override("gpt-5.5", |model| model.use_responses_lite = true)
.with_pre_build_hook(|home| {
let hooks = json!({
"hooks": {
@@ -547,29 +564,233 @@ async fn mcp_tool_hook_passes_thread_metadata_to_model_hidden_tools() -> Result<
AppToolApproval::Prompt,
/*prefix_mcp_tool_names*/ true,
);
let mut servers = config.mcp_servers.get().clone();
let mcp_server = servers.get_mut(RMCP_SERVER).expect("test MCP server");
mcp_server.enabled_tools = Some(vec!["thread_hint".to_string()]);
mcp_server.startup_readiness = startup_readiness;
let unused_server = mcp_server.clone();
servers.insert("unused".to_string(), unused_server);
config.mcp_servers.set(servers).expect("test MCP servers");
})
.build(&server)
.await?;
wait_for_mcp_server(&test.codex, RMCP_SERVER).await?;
test.submit_turn("load thread context").await?;
let subagent = test
.thread_manager
.start_thread(StartThreadOptions {
session_source: Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: test.session_configured.thread_id,
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
})),
..test.start_thread_options().await
})
.await?;
wait_for_event(&subagent.thread, |event| {
matches!(event, EventMsg::McpStartupComplete(_))
})
.await;
let (mcp_config, _) = subagent
.thread
.current_mcp_config_and_runtime_context()
.await;
assert_eq!(
subagent
.thread
.mcp_connection_statuses(&mcp_config)
.await
.get(RMCP_SERVER),
Some(&McpServerConnectionStatus::NotStarted),
);
subagent
.thread
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "load thread context".to_string(),
text_elements: Vec::new(),
}]))
.await?;
let EventMsg::HookCompleted(hook) = wait_for_event(&subagent.thread, |event| {
matches!(event, EventMsg::HookCompleted(_))
})
.await
else {
unreachable!("event predicate guarantees hook completion");
};
assert_eq!(hook.run.status, HookRunStatus::Completed);
wait_for_event(&subagent.thread, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
assert_eq!(
subagent
.thread
.mcp_connection_statuses(&mcp_config)
.await
.get("unused"),
Some(&McpServerConnectionStatus::NotStarted),
"a hook must leave unrelated cached servers dormant",
);
let request = response.single_request();
let expected_context = format!(
"manual history hint for thread {}\nunstructured notes/thread_hint fixture result",
test.session_configured.thread_id,
subagent.thread_id,
);
assert!(
request
.message_input_texts("developer")
.contains(&expected_context)
let hint = request
.input()
.into_iter()
.find(|item| item["content"][0]["text"] == expected_context)
.expect("hook context in the model request");
assert_eq!(
hint["internal_chat_message_metadata_passthrough"]["content_item_metadata"],
json!([ContentItemMetadata::tool(Some(
RMCP_PREFIXED_NAMESPACE.to_string().into()
))]),
);
let tools = request.inputs_of_type("additional_tools");
assert_eq!(tools.len(), 1);
assert!(
request
.tool_by_name(RMCP_PREFIXED_NAMESPACE, "thread_hint")
.is_none(),
namespace_child_tool(&tools[0], RMCP_PREFIXED_NAMESPACE, "thread_hint").is_none(),
"the hook tool must remain hidden from the model"
);
subagent.thread.shutdown_and_wait().await?;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn mcp_tool_hook_timeout_includes_cached_server_startup() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let response = mount_sse_once(
&server,
sse(vec![
ev_response_created("resp-1"),
ev_assistant_message("msg-1", "thread context received"),
ev_completed("resp-1"),
]),
)
.await;
let rmcp_test_server_bin = stdio_server_bin()?;
let test = test_codex()
.with_pre_build_hook(|home| {
let hooks = json!({
"hooks": {
"UserPromptSubmit": [{
"hooks": [{
"type": "mcp_tool",
"server": RMCP_SERVER,
"tool": "thread_hint",
"input": {},
"timeout": 1,
}],
}],
},
});
fs::write(home.join("hooks.json"), hooks.to_string())
.expect("write model-hidden MCP tool hook config");
})
.with_config(move |config| {
enable_hooks_and_rmcp_server(
config,
rmcp_test_server_bin,
AppToolApproval::Prompt,
/*prefix_mcp_tool_names*/ true,
);
let mut servers = config.mcp_servers.get().clone();
let mcp_server = servers.get_mut(RMCP_SERVER).expect("test MCP server");
mcp_server.enabled_tools = Some(vec!["thread_hint".to_string()]);
let McpServerTransportConfig::Stdio { env, .. } = &mut mcp_server.transport else {
unreachable!("test server uses stdio");
};
let barrier_file = config.cwd.join("allow-initialize");
fs::write(&barrier_file, "ready").expect("allow parent server initialization");
*env = Some(HashMap::from([(
"MCP_TEST_INITIALIZE_BARRIER_FILE".to_string(),
barrier_file.to_string_lossy().into_owned(),
)]));
config.mcp_servers.set(servers).expect("test MCP servers");
})
.build(&server)
.await?;
wait_for_mcp_server(&test.codex, RMCP_SERVER).await?;
let barrier_file = test.config.cwd.join("allow-initialize");
fs::remove_file(&barrier_file)?;
let subagent = test
.thread_manager
.start_thread(StartThreadOptions {
session_source: Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: test.session_configured.thread_id,
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
})),
..test.start_thread_options().await
})
.await?;
wait_for_event(&subagent.thread, |event| {
matches!(event, EventMsg::McpStartupComplete(_))
})
.await;
let (mcp_config, _) = subagent
.thread
.current_mcp_config_and_runtime_context()
.await;
assert_eq!(
subagent
.thread
.mcp_connection_statuses(&mcp_config)
.await
.get(RMCP_SERVER),
Some(&McpServerConnectionStatus::NotStarted),
);
subagent
.thread
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "load thread context".to_string(),
text_elements: Vec::new(),
}]))
.await?;
let EventMsg::HookCompleted(hook) = tokio::time::timeout(
Duration::from_secs(5),
wait_for_event(&subagent.thread, |event| {
matches!(event, EventMsg::HookCompleted(_))
}),
)
.await
.context("hook must finish before the server's ten-second startup timeout")?
else {
unreachable!("event predicate guarantees hook completion");
};
assert_eq!(hook.run.status, HookRunStatus::Failed);
assert!(
hook.run
.entries
.iter()
.any(|entry| entry.text.contains("MCP hook timed out"))
);
fs::write(&barrier_file, "ready")?;
wait_for_event(&subagent.thread, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
assert!(
!response
.single_request()
.message_input_texts("developer")
.iter()
.any(|text| text.contains("manual history hint")),
"a timed-out hook must not inject context"
);
subagent.thread.shutdown_and_wait().await?;
Ok(())
}
+38 -1
View File
@@ -950,8 +950,45 @@ async fn cached_mcp_startup_is_eager_for_root_and_lazy_for_subagents() -> anyhow
..StartThreadOptions::new(filtered_config)
})
.await?;
let filtered_response = mount_sse_once(
&responses_server,
responses::sse(vec![
responses::ev_response_created("filtered"),
responses::ev_assistant_message("filtered-message", "done"),
responses::ev_completed("filtered"),
]),
)
.await;
filtered_thread
.start_or_steer_turn(user_turn("Do not use tools."))
.await?;
wait_for_event(&filtered_thread, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
assert!(
responses::namespace_child_tool(
&filtered_response.single_request().body_json(),
NAMESPACE,
"cwd",
)
.is_none()
);
let (mcp_config, _) = filtered_thread
.current_mcp_config_and_runtime_context()
.await;
assert_eq!(
filtered_thread
.mcp_connection_statuses(&mcp_config)
.await
.get(SERVER_NAME),
Some(&McpServerConnectionStatus::NotStarted),
);
let result = filtered_thread
.call_mcp_tool(SERVER_NAME, "cwd", Some(json!({})), /*meta*/ None)
.await?;
assert_ne!(result.is_error, Some(true));
let filtered_pid = wait_for_new_pid(fs.as_ref(), &pid_file, Some(&second_pid)).await?;
wait_for_mcp_server(&filtered_thread, SERVER_NAME).await?;
filtered_thread.shutdown_and_wait().await?;
fs.remove(
&barrier_file,
+76 -12
View File
@@ -1,6 +1,7 @@
use anyhow::Result;
use codex_config::types::McpServerConfig;
use codex_config::types::McpServerTransportConfig;
use codex_core::StartThreadOptions;
use codex_core::TurnInputRequest;
use codex_core::config::Config;
use codex_core::config::TokenBudgetConfig;
@@ -10,6 +11,8 @@ use codex_login::CodexAuth;
use codex_model_provider_info::built_in_model_providers;
use codex_protocol::config_types::AutoCompactTokenLimitScope;
use codex_protocol::items::TurnItem;
use codex_protocol::mcp::McpServerConnectionStatus;
use codex_protocol::models::ContentItemMetadata;
use codex_protocol::openai_models::ModelTokenBudgetConfig;
use codex_protocol::protocol::CONTEXT_WINDOW_CLOSE_TAG;
use codex_protocol::protocol::CONTEXT_WINDOW_GUIDANCE_CLOSE_TAG;
@@ -21,6 +24,8 @@ use codex_protocol::protocol::HookRunStatus;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::ItemStartedEvent;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::ThreadSettingsOverrides;
use codex_protocol::user_input::UserInput;
use codex_skills_extension::SkillsExtensionConfig;
@@ -39,6 +44,7 @@ use core_test_support::responses::ev_function_call;
use core_test_support::responses::ev_response_created;
use core_test_support::responses::mount_sse_once;
use core_test_support::responses::mount_sse_sequence;
use core_test_support::responses::namespace_child_tool;
use core_test_support::responses::sse;
use core_test_support::responses::sse_completed;
use core_test_support::responses::start_mock_server;
@@ -745,13 +751,16 @@ async fn model_token_budget_defaults_do_not_enable_disabled_feature() -> Result<
Ok(())
}
#[test_case(false; "root")]
#[test_case(true; "cached_subagent")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn token_budget_context_injects_plain_thread_hint_text() -> Result<()> {
async fn token_budget_context_injects_plain_thread_hint_text(cached_subagent: bool) -> Result<()> {
skip_if_no_network!(Ok(()));
let server = start_mock_server().await;
let rmcp_test_server_bin = stdio_server_bin()?;
let test = test_codex()
.with_model_info_override("gpt-5.5", |model| model.use_responses_lite = true)
.with_config(move |config| {
config.model_context_window = Some(CONFIGURED_CONTEXT_WINDOW);
config
@@ -781,7 +790,7 @@ async fn token_budget_context_injects_plain_thread_hint_text() -> Result<()> {
startup_timeout_sec: Some(Duration::from_secs(10)),
tool_timeout_sec: None,
default_tools_approval_mode: None,
enabled_tools: None,
enabled_tools: Some(vec!["thread_hint".to_string()]),
disabled_tools: None,
scopes: None,
oauth: None,
@@ -797,19 +806,55 @@ async fn token_budget_context_injects_plain_thread_hint_text() -> Result<()> {
.build(&server)
.await?;
wait_for_mcp_server(&test.codex, "notes").await?;
let responses = mount_sse_sequence(
let (thread, thread_id) = if cached_subagent {
let subagent = test
.thread_manager
.start_thread(StartThreadOptions {
session_source: Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: test.session_configured.thread_id,
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
})),
..test.start_thread_options().await
})
.await?;
wait_for_event(&subagent.thread, |event| {
matches!(event, EventMsg::McpStartupComplete(_))
})
.await;
let (mcp_config, _) = subagent
.thread
.current_mcp_config_and_runtime_context()
.await;
assert_eq!(
subagent
.thread
.mcp_connection_statuses(&mcp_config)
.await
.get("notes"),
Some(&McpServerConnectionStatus::NotStarted),
);
(subagent.thread, subagent.thread_id)
} else {
(Arc::clone(&test.codex), test.session_configured.thread_id)
};
let responses = mount_sse_once(
&server,
vec![sse(vec![
ev_response_created("resp-1"),
ev_completed("resp-1"),
])],
sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]),
)
.await;
test.submit_turn("inject the history hint").await?;
thread
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "inject the history hint".to_string(),
text_elements: Vec::new(),
}]))
.await?;
wait_for_event(&thread, |event| matches!(event, EventMsg::TurnComplete(_))).await;
let request = responses.single_request();
let thread_id = test.session_configured.thread_id;
let token_budgets = token_budget_contexts(&request);
assert_eq!(token_budgets.len(), 1);
let captures = assert_regex_match(
@@ -822,12 +867,31 @@ async fn token_budget_context_injects_plain_thread_hint_text() -> Result<()> {
captures.get(1).expect("first window id capture").as_str(),
captures.get(2).expect("current window id capture").as_str()
);
let tools = request.inputs_of_type("additional_tools");
assert_eq!(tools.len(), 1);
assert!(
!tool_names(&request)
.iter()
.any(|name| name == "mcp__notes__thread_hint"),
namespace_child_tool(&tools[0], "mcp__notes", "thread_hint").is_none(),
"thread_hint should be hidden from model tool exposure"
);
let metadata = request.input().into_iter().find_map(|item| {
let index = item["content"].as_array()?.iter().position(|content| {
content["text"]
.as_str()
.is_some_and(|text| text.starts_with("manual history hint for thread "))
})?;
Some(
item["internal_chat_message_metadata_passthrough"]["content_item_metadata"][index]
.clone(),
)
});
assert_eq!(
metadata,
Some(json!(ContentItemMetadata::tool(Some(
"mcp__notes".to_string().into()
)))),
"the first Notes hint must retain its exact tool namespace"
);
thread.shutdown_and_wait().await?;
Ok(())
}