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

Enable executor cleanup hooks with V2 capability discovery (#56044)

Browse Source 
## Why

Executor hooks depended on V1 discovery snapshots, so plugins discovered through V2 could not run their cleanup hooks.

## What changed

Prepare executor hook sources during step setup from either V1 or V2 discovery and capture them in `StepContext` for stop, interrupt, and memory consolidation hooks. Restrict V2 sources to selected plugin installations while preserving plugin enablement, hook admission, MCP routing, and app tool policy checks.

## Testing

Add unit coverage for V2 hook admission, environment routing, disabled plugins, and failed discovery. Add integration coverage showing that `Stop` and `Interrupt` cleanup hooks run with V2 discovery alone and remain silent when the plugin is disabled.

GitOrigin-RevId: a4c0e9c56bdebcdf22cea63f02f8fb4c5eed358e
This commit is contained in:
Anthony Tafoya
2026-10-22 01:42:11 +00:00
committed by copyberry
parent 45179176e5
commit 0769c6aeeb
9 changed files with 383 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core-plugins/src/executor_hooks.rs
+48 -13
View File
@@ -3,7 +3,9 @@
use codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID;
use codex_config::HookHandlerConfig;
use codex_config::types::PluginsConfigToml;
use codex_exec_server::DiscoveredPluginFiles;
use codex_exec_server::ExecutorCapabilityDiscoverySnapshot;
use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY;
use codex_mcp::ToolInfo;
@@ -12,6 +14,7 @@ use codex_plugin::PluginId;
use codex_plugin::is_allowlisted_bundled_cleanup_hook;
use codex_plugin::manifest::PluginManifestHooks;
use codex_protocol::capabilities::CapabilityRootLocation;
use codex_utils_path_uri::PathUri;
use serde_json::Map;
use serde_json::Value;
@@ -32,23 +35,55 @@ pub fn executor_plugin_hook_sources<'a>(
plugins_config: &PluginsConfigToml,
lookup_enabled_tool: impl Fn(&str, &str) -> Option<&'a ToolInfo>,
) -> Vec<ExecutorPluginHookSource> {
let mut sources = Vec::new();
for entry in snapshot.roots() {
if !plugins_config.allows_plugin(&entry.selected_root.id) {
continue;
}
let Ok(plugin_id) = PluginId::parse(&entry.selected_root.id) else {
continue;
};
let plugins = snapshot.roots().iter().filter_map(|entry| {
let CapabilityRootLocation::Environment {
environment_id,
path: plugin_root,
} = &entry.selected_root.location;
let Ok(discovery) = &entry.result else {
let plugin = entry.result.as_ref().ok()?.plugin.as_ref()?;
Some((
entry.selected_root.id.as_str(),
environment_id.as_str(),
plugin_root,
plugin,
))
});
admitted_hook_sources(plugins, plugins_config, lookup_enabled_tool)
}
/// Applies the same inline-hook admission and MCP routing policy to V2 inventories.
/// Callers must additionally restrict sources to their accepted plugin installations.
pub fn executor_plugin_hook_sources_v2<'a>(
snapshots: &[ExecutorCapabilityDiscoveryV2Snapshot],
plugins_config: &PluginsConfigToml,
lookup_enabled_tool: impl Fn(&str, &str) -> Option<&'a ToolInfo>,
) -> Vec<ExecutorPluginHookSource> {
let plugins = snapshots.iter().flat_map(|snapshot| {
snapshot.result.iter().flat_map(move |inventory| {
inventory.plugins.iter().map(move |plugin| {
(
plugin.id.as_str(),
snapshot.environment_id.as_str(),
&plugin.root,
&plugin.files,
)
})
})
});
admitted_hook_sources(plugins, plugins_config, lookup_enabled_tool)
}
fn admitted_hook_sources<'a, 'b>(
plugins: impl Iterator<Item = (&'b str, &'b str, &'b PathUri, &'b DiscoveredPluginFiles)>,
plugins_config: &PluginsConfigToml,
lookup_enabled_tool: impl Fn(&str, &str) -> Option<&'a ToolInfo>,
) -> Vec<ExecutorPluginHookSource> {
let mut sources = Vec::new();
for (plugin_id, environment_id, plugin_root, plugin) in plugins {
if !plugins_config.allows_plugin(plugin_id) {
continue;
};
let Some(plugin) = &discovery.plugin else {
}
let Ok(plugin_id) = PluginId::parse(plugin_id) else {
continue;
};
let Ok(manifest) = parse_plugin_manifest_uri(
@@ -72,7 +107,7 @@ pub fn executor_plugin_hook_sources<'a>(
sources.push(ExecutorPluginHookSource {
plugin_id: plugin_id.clone(),
environment_id: environment_id.clone(),
environment_id: environment_id.to_string(),
mcp_environment_id: None,
mcp_metadata: None,
plugin_root: plugin_root.clone(),
codex-rs/core-plugins/src/executor_hooks_tests.rs
+56
View File
@@ -499,3 +499,59 @@ fn ignores_file_backed_executor_plugin_hooks() {
Vec::<ExecutorPluginHookSource>::new()
);
}
#[test]
fn v2_hooks_preserve_admission_policy_and_environment_routing() {
let manifest = cleanup_hook_manifest();
let files = DiscoveredPluginFiles {
manifest: CapabilityTextFile {
path: PathUri::parse("file:///plugins/computer-use/.codex-plugin/plugin.json").unwrap(),
contents: manifest.to_string(),
},
mcp_config: None,
apps_config: None,
};
let plugin = codex_exec_server::ExecutorPlugin {
id: "computer-use@openai-bundled".to_string(),
display_name: "Computer use".to_string(),
remote_plugin_id: None,
version: "local".to_string(),
root: PathUri::parse("file:///plugins/computer-use").unwrap(),
files,
skills: Vec::new(),
};
let mut untrusted = plugin.clone();
untrusted.id = "computer-use@untrusted".to_string();
let inventory = Arc::new(codex_exec_server::DiscoverV2CapabilitiesResponse {
plugins: vec![plugin, untrusted],
..Default::default()
});
let snapshots = [
ExecutorCapabilityDiscoveryV2Snapshot {
environment_id: "executor-a".to_string(),
result: Ok(Arc::clone(&inventory)),
},
ExecutorCapabilityDiscoveryV2Snapshot {
environment_id: "executor-b".to_string(),
result: Ok(inventory),
},
ExecutorCapabilityDiscoveryV2Snapshot {
environment_id: "failed".to_string(),
result: Err("discovery failed".to_string()),
},
];
let mut second = expected_source(/*index*/ 0);
second.environment_id = "executor-b".to_string();
assert_eq!(
executor_plugin_hook_sources_v2(&snapshots, &PluginsConfigToml::default(), |_, _| None),
vec![expected_source(/*index*/ 0), second]
);
let disabled = toml::from_str(
"[_default]\nenabled = true\n[\"computer-use@openai-bundled\"]\nenabled = false",
)
.unwrap();
assert_eq!(
executor_plugin_hook_sources_v2(&snapshots, &disabled, |_, _| None),
Vec::new()
);
}
codex-rs/core-plugins/src/lib.rs
+1
View File
@@ -76,6 +76,7 @@ pub use discoverable::ToolSuggestDiscoverablePlugin;
pub use discoverable::ToolSuggestPluginDiscoveryInput;
pub use executor_catalog::catalog_from_snapshots;
pub use executor_hooks::executor_plugin_hook_sources;
pub use executor_hooks::executor_plugin_hook_sources_v2;
pub use executor_provider::ExecutorPluginProvider;
pub use executor_provider::ExecutorPluginProviderError;
pub use executor_provider::PluginRootOwnership;
codex-rs/core/src/hook_runtime.rs
+7 -50
View File
@@ -1,3 +1,7 @@
mod executor;
pub(crate) use executor::prepare_executor_hook_sources;
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;
@@ -5,9 +9,6 @@ use std::time::Duration;
use codex_analytics::CompactionTrigger;
use codex_analytics::HookRunFact;
use codex_analytics::build_track_events_context;
use codex_connectors::AppToolPolicyEvaluator;
use codex_connectors::AppToolPolicyInput;
use codex_core_plugins::executor_plugin_hook_sources;
use codex_hooks::InterruptRequest;
use codex_hooks::PermissionRequestDecision;
use codex_hooks::PermissionRequestOutcome;
@@ -25,10 +26,8 @@ use codex_hooks::UserPromptSubmitOutcome;
use codex_hooks::UserPromptSubmitRequest;
use codex_hooks::hook_execution_mode_label;
use codex_hooks::hook_handler_type_label;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_otel::HOOK_RUN_DURATION_METRIC;
use codex_otel::HOOK_RUN_METRIC;
use codex_plugin::ExecutorPluginHookSource;
use codex_protocol::items::FunctionCallOutputItem;
use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
@@ -327,48 +326,6 @@ pub(crate) async fn run_post_tool_use_hooks(
outcome
}
fn executor_hook_sources_for_step(step_context: &StepContext) -> Vec<ExecutorPluginHookSource> {
step_context
.executor_capability_discovery
.as_deref()
.map(|snapshot| {
let config = step_context.mcp.config();
let app_tool_policy = AppToolPolicyEvaluator::new(&config.config_layer_stack);
executor_plugin_hook_sources(snapshot, &config.plugins, |server, tool| {
step_context
.mcp
.tool_info(server, tool)
.filter(|tool_info| {
if server != CODEX_APPS_MCP_SERVER_NAME {
return true;
}
let annotations = tool_info.tool.annotations.as_ref();
app_tool_policy
.policy(AppToolPolicyInput {
connector_id: tool_info.connector_id.as_deref(),
link_id: None,
tool_name: &tool_info.tool.name,
tool_title: tool_info.tool.title.as_deref(),
destructive_hint: annotations
.and_then(|annotations| annotations.destructive_hint),
open_world_hint: annotations
.and_then(|annotations| annotations.open_world_hint),
})
.enabled
})
})
.into_iter()
.filter(|source| {
!step_context
.turn
.disabled_plugin_ids
.contains(&source.plugin_id.as_key())
})
.collect()
})
.unwrap_or_default()
}
fn build_request_metadata(
step_context: Option<&StepContext>,
turn_context: &TurnContext,
@@ -458,7 +415,7 @@ pub(crate) async fn run_turn_stop_hooks(
last_assistant_message,
target,
};
let executor_hook_sources = executor_hook_sources_for_step(step_context);
let executor_hook_sources = step_context.executor_hook_sources.clone();
let hooks = sess.hooks().with_executor_hooks(executor_hook_sources);
emit_hook_started_events(sess, turn_context, hooks.preview_stop(&request)).await;
@@ -512,7 +469,7 @@ pub(crate) async fn run_turn_interrupt_hooks(
let last_known_step_context = turn_state.lock().await.last_known_step_context.clone();
let executor_hook_sources = last_known_step_context
.as_deref()
.map(executor_hook_sources_for_step)
.map(|step| step.executor_hook_sources.clone())
.unwrap_or_default();
let has_executor_hooks = !executor_hook_sources.is_empty();
let hooks = sess.hooks().with_executor_hooks(executor_hook_sources);
@@ -1082,7 +1039,7 @@ pub(crate) fn hooks_for_model_request(
match step.turn.session_source {
SessionSource::Internal(InternalSessionSource::MemoryConsolidation)
| SessionSource::SubAgent(SubAgentSource::MemoryConsolidation) => {
Arc::new(hooks.for_memory_consolidation(executor_hook_sources_for_step(step)))
Arc::new(hooks.for_memory_consolidation(step.executor_hook_sources.clone()))
}
_ => hooks,
}
codex-rs/core/src/hook_runtime/executor.rs
+66
View File
@@ -0,0 +1,66 @@
//! Captures admitted executor hooks from this step's discovery and MCP policy.
use codex_connectors::AppToolPolicyEvaluator;
use codex_connectors::AppToolPolicyInput;
use codex_core_plugins::executor_plugin_hook_sources;
use codex_core_plugins::executor_plugin_hook_sources_v2;
use codex_exec_server::ExecutorCapabilityDiscoverySnapshot;
use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot;
use codex_extension_api::SelectedPluginSnapshot;
use codex_mcp::CODEX_APPS_MCP_SERVER_NAME;
use codex_mcp::McpBinding;
use codex_plugin::ExecutorPluginHookSource;
pub(crate) fn prepare_executor_hook_sources(
mcp: &McpBinding,
selected_plugins: &SelectedPluginSnapshot,
discovery: Option<&ExecutorCapabilityDiscoverySnapshot>,
discovery_v2: Option<&[ExecutorCapabilityDiscoveryV2Snapshot]>,
) -> Vec<ExecutorPluginHookSource> {
let config = mcp.config();
let app_tool_policy = AppToolPolicyEvaluator::new(&config.config_layer_stack);
let lookup_enabled_tool = |server: &str, tool: &str| {
mcp.tool_info(server, tool).filter(|tool_info| {
if server != CODEX_APPS_MCP_SERVER_NAME {
return true;
}
let annotations = tool_info.tool.annotations.as_ref();
app_tool_policy
.policy(AppToolPolicyInput {
connector_id: tool_info.connector_id.as_deref(),
link_id: None,
tool_name: &tool_info.tool.name,
tool_title: tool_info.tool.title.as_deref(),
destructive_hint: annotations
.and_then(|annotations| annotations.destructive_hint),
open_world_hint: annotations
.and_then(|annotations| annotations.open_world_hint),
})
.enabled
})
};
let sources = if let Some(snapshots) = discovery_v2 {
executor_plugin_hook_sources_v2(snapshots, &config.plugins, lookup_enabled_tool)
.into_iter()
.filter(|source| {
let root_id = format!("{}:{}", source.environment_id, source.plugin_id.as_key());
selected_plugins
.plugins
.iter()
.any(|plugin| plugin.selected_root_id.as_deref() == Some(root_id.as_str()))
})
.collect()
} else if let Some(snapshot) = discovery {
executor_plugin_hook_sources(snapshot, &config.plugins, lookup_enabled_tool)
} else {
Vec::new()
};
sources
.into_iter()
.filter(|source| {
!selected_plugins
.disabled_plugin_ids
.contains(&source.plugin_id.as_key())
})
.collect()
}
codex-rs/core/src/session/mod.rs
+9 -10
View File
@@ -3995,8 +3995,6 @@ impl Session {
.executor_capability_discovery_for_step_v2(&turn_context.config, &environments)
.await;
let v2_snapshots = executor_capability_discovery_v2.as_ref();
// V1 discovery is still retained on StepContext for executor Stop/Interrupt hooks.
// TODO: Prepare hook sources at capture time, including V2, instead.
let (executor_capability_discovery, discovery_recovered) = if v2_snapshots.is_some() {
(None, false)
} else {
@@ -4071,6 +4069,12 @@ impl Session {
.or_cancel(cancellation_token)
.await?;
}
let executor_hook_sources = crate::hook_runtime::prepare_executor_hook_sources(
&mcp,
&selected_plugins,
executor_capability_discovery.as_deref(),
executor_capability_discovery_v2.as_deref(),
);
extension_data.insert(selected_plugins);
let tool_router = turn::built_tools(
self.as_ref(),
@@ -4082,12 +4086,7 @@ impl Session {
prepared_recommendations,
)
.await?;
Ok::<_, CodexErr>((
executor_capability_discovery,
mcp,
tool_router,
extension_data,
))
Ok::<_, CodexErr>((executor_hook_sources, mcp, tool_router, extension_data))
});
// Returned warnings must finish delivery even if tools fail or preparation is cancelled.
let (loaded_agents_md, prepared_tools) =
@@ -4096,7 +4095,7 @@ impl Session {
if cancellation_token.is_cancelled() {
return Err(CodexErr::TurnAborted);
}
let (executor_capability_discovery, mcp, tool_router, extension_data) = prepared_tools??;
let (executor_hook_sources, mcp, tool_router, extension_data) = prepared_tools??;
Ok(Arc::new(StepContext {
preempt: turn_context
.config
@@ -4111,7 +4110,7 @@ impl Session {
extension_data,
environments,
inherited_environments,
executor_capability_discovery,
executor_hook_sources,
mcp,
tool_router,
loaded_agents_md,
codex-rs/core/src/session/step_context.rs
+3 -3
View File
@@ -10,11 +10,11 @@ use crate::session::step_settings::ResolvedStepSettings;
use crate::session::turn_context::TurnContext;
use crate::session::turn_context::TurnEnvironment;
use crate::tools::router::ToolRouter;
use codex_exec_server::ExecutorCapabilityDiscoverySnapshot;
use codex_extension_api::ExtensionData;
use codex_file_system::EnvironmentAccess;
use codex_mcp::McpBinding;
use codex_otel::SessionTelemetry;
use codex_plugin::ExecutorPluginHookSource;
use codex_protocol::items::ModelInvocationContext;
use codex_protocol::protocol::TurnContextItem;
use tokio_util::sync::CancellationToken;
@@ -35,8 +35,8 @@ pub(crate) struct StepContext {
pub(crate) environments: TurnEnvironmentSnapshot,
/// Child bindings captured with this step; configured-only when opted in.
pub(crate) inherited_environments: TurnEnvironmentSnapshot,
/// Legacy discovery retained for executor hook materialization in this exact step.
pub(crate) executor_capability_discovery: Option<Arc<ExecutorCapabilityDiscoverySnapshot>>,
/// Admitted executor hooks bound to this step's discovery, plugin policy, and MCP routing.
pub(crate) executor_hook_sources: Vec<ExecutorPluginHookSource>,
/// Keeps the extension inputs used to build this step's tools for its World State as well.
pub(crate) extension_data: ExtensionData,
/// The exact MCP connections, configuration, and catalog captured for this step.
codex-rs/core/src/session/tests.rs
+1 -1
View File
@@ -291,7 +291,7 @@ impl StepContext {
turn: Arc::clone(&turn),
inherited_environments: environments.clone(),
environments,
executor_capability_discovery: None,
executor_hook_sources: Vec::new(),
extension_data: codex_extension_api::ExtensionData::new(turn.sub_id.clone()),
mcp: Arc::new(codex_mcp::McpBinding::empty(mcp_config_for_test(
&turn.config,
codex-rs/core/tests/suite/executor_discovery_v2.rs
+192
View File
@@ -2,6 +2,7 @@
use anyhow::Context;
use anyhow::Result;
use codex_core::TurnInputRequest;
use codex_core::config::Config;
use codex_core::windows_sandbox::WindowsSandboxLevelExt;
use codex_core_plugins::ExecutorPluginProvider;
@@ -36,7 +37,11 @@ use codex_mcp::McpResourceClient;
use codex_mcp_extension::PluginProviders;
use codex_mcp_extension::install_plugin_providers;
use codex_protocol::config_types::WindowsSandboxLevel;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::SKILLS_INSTRUCTIONS_OPEN_TAG;
use codex_protocol::protocol::TurnEnvironmentSelections;
use codex_protocol::user_input::UserInput;
use codex_skills_extension::ExecutorSkillProvider;
use codex_skills_extension::SkillProviders;
use codex_skills_extension::SkillsExtensionConfig;
@@ -1196,3 +1201,190 @@ async fn auth_rotation_refreshes_plugin_acceptance_before_merged_skills() -> Res
);
Ok(())
}
#[test_case::test_case("Stop"; "stop")]
#[test_case::test_case("Interrupt"; "interrupt")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn v2_cleanup_hooks_follow_plugin_selection_without_legacy_roots(
hook_event: &'static str,
) -> Result<()> {
skip_if_no_network!(Ok(()));
let model = responses::start_mock_server().await;
let hook_calls = Arc::new(AtomicUsize::new(0));
let recorded = Arc::clone(&hook_calls);
wiremock::Mock::given(wiremock::matchers::path("/hooks"))
.respond_with(move |request: &wiremock::Request| {
let request: Value =
serde_json::from_slice(&request.body).expect("MCP request should be valid JSON");
let result = match request["method"].as_str() {
Some("initialize") => json!({
"protocolVersion": request["params"]["protocolVersion"],
"capabilities": {"tools": {}}, "serverInfo": {"name": "hooks", "version": "1"}
}),
Some("notifications/initialized") => return wiremock::ResponseTemplate::new(202),
Some("tools/list") => json!({"tools": [{
"name": "turn_ended", "inputSchema": {"type": "object"}
}]}),
Some("tools/call") => {
assert_eq!(request["params"]["name"], "turn_ended");
assert_eq!(
request["params"]["arguments"]["hook_event_name"],
hook_event
);
recorded.fetch_add(1, Ordering::SeqCst);
json!({"content": []})
}
other => panic!("unexpected MCP request: {other:?}"),
};
wiremock::ResponseTemplate::new(200)
.set_body_json(json!({"jsonrpc": "2.0", "id": request["id"], "result": result}))
})
.mount(&model)
.await;
let manifest = json!({"name": "computer-use", "hooks": {"hooks": {(hook_event): [{"hooks": [{
"type": "mcp_tool", "server": "node_repl", "tool": "turn_ended",
"input": {"hook_event_name": "${hook_event_name}", "session_id": "${session_id}", "turn_id": "${turn_id}"}
}]}]}}});
let inventory = json!({"plugins": [{
"id": "computer-use@openai-bundled", "displayName": "Computer use",
"remotePluginId": null, "version": "1", "root": "file:///plugins/computer-use",
"files": {"manifest": {"path": "file:///plugins/computer-use/.codex-plugin/plugin.json",
"contents": manifest.to_string()}, "mcpConfig": null, "appsConfig": null}, "skills": []
}], "skills": [], "warnings": []});
let executor = Executor::start(/*supports_v2*/ true, DiscoveryFailure::None).await?;
*executor.inventory.lock().await = Some(inventory);
let hook_url = format!("{}/hooks", model.uri());
let hook_server = serde_json::from_value(json!({
"url": hook_url, "environment_id": codex_exec_server::REMOTE_ENVIRONMENT_ID
}))?;
let mut extensions = ExtensionRegistryBuilder::new();
codex_mcp_extension::install_plugins(
&mut extensions,
Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()),
);
let home = Arc::new(tempfile::TempDir::new()?);
std::fs::write(
home.path().join("config.toml"),
"[plugins.\"computer-use@openai-bundled\"]\nenabled = true\n",
)?;
let test = test_codex()
.with_home(home)
.with_extensions(Arc::new(extensions.build()))
.with_exec_server_url(executor.url.clone())
.with_config(move |config| {
config
.features
.enable(Feature::Plugins)
.expect("enable plugins");
config
.features
.enable(Feature::ExecutorCapabilityDiscoveryV2)
.expect("enable V2 discovery");
config
.features
.disable(Feature::ExecutorCapabilityDiscovery)
.expect("disable V1 discovery");
config
.mcp_servers
.set(
[(String::from("node_repl"), hook_server)]
.into_iter()
.collect(),
)
.expect("configure the executor hook server");
})
.build_with_auto_env(&model)
.await?;
core_test_support::wait_for_mcp_server(&test.codex, "node_repl").await?;
let mut selection = test.executor_environment().selection().clone();
selection.environment_id = codex_exec_server::REMOTE_ENVIRONMENT_ID.to_string();
selection.selected_capability_roots = Default::default();
for disabled in [false, true] {
core_test_support::submit_thread_settings(
&test.codex,
codex_protocol::protocol::ThreadSettingsOverrides {
environments: Some(
TurnEnvironmentSelections::new(
test.config.cwd.clone(),
vec![selection.clone()],
)
.into_requests(),
),
disabled_plugin_ids: Some(if disabled {
vec!["computer-use@openai-bundled".to_string()]
} else {
Vec::new()
}),
..Default::default()
},
)
.await?;
let mut response = responses::sse_response(responses::sse(vec![
responses::ev_response_created("response"),
responses::ev_assistant_message("message", "Done."),
responses::ev_completed("response"),
]));
if hook_event == "Interrupt" {
response = response.set_delay(std::time::Duration::from_secs(60));
}
let requests = responses::mount_response_once(&model, response).await;
test.codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "Finish this turn.".to_string(),
text_elements: Vec::new(),
}]))
.await?;
if hook_event == "Interrupt" {
tokio::time::timeout(std::time::Duration::from_secs(10), async {
while requests.requests().is_empty() {
tokio::task::yield_now().await;
}
})
.await
.context("interrupted turn should reach inference with its prepared step")?;
test.codex.submit(Op::Interrupt).await?;
}
core_test_support::wait_for_event(&test.codex, |event| {
if hook_event == "Interrupt" {
matches!(event, EventMsg::TurnAborted(_))
} else {
matches!(event, EventMsg::TurnComplete(_))
}
})
.await;
if !disabled {
tokio::time::timeout(std::time::Duration::from_secs(10), async {
while hook_calls.load(Ordering::SeqCst) == 0 {
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await
.context("V2 cleanup hook should run")?;
} else {
// Cleanup is asynchronous; turn completion alone cannot prove disablement.
assert!(
tokio::time::timeout(std::time::Duration::from_secs(1), async {
while hook_calls.load(Ordering::SeqCst) == 1 {
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
})
.await
.is_err(),
"disabled cleanup hook must remain silent"
);
}
assert_eq!(hook_calls.load(Ordering::SeqCst), 1);
}
let requests = executor.requests.lock().await;
assert!(
requests
.iter()
.any(|request| request["method"] == "capabilities/discoverV2")
);
assert!(
!requests
.iter()
.any(|request| request["method"] == "capabilityRoots/discoverV1")
);
Ok(())
}