diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs index 7cd03930d8..367a4a64c8 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread.rs @@ -685,7 +685,7 @@ pub struct ThreadIncrementElicitationParams { #[ts(export_to = "v2/")] pub struct ThreadIncrementElicitationResponse { /// Current out-of-band elicitation count after the increment. - pub count: u64, + pub count: i64, /// Whether timeout accounting is paused after applying the increment. pub paused: bool, } @@ -705,7 +705,7 @@ pub struct ThreadDecrementElicitationParams { #[ts(export_to = "v2/")] pub struct ThreadDecrementElicitationResponse { /// Current out-of-band elicitation count after the decrement. - pub count: u64, + pub count: i64, /// Whether timeout accounting remains paused after applying the decrement. pub paused: bool, } diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index c76e3c3c08..2143e3b577 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -142,6 +142,7 @@ impl McpConnectionManager { tool_plugin_provenance: ToolPluginProvenance, auth: Option<&CodexAuth>, elicitation_reviewer: Option, + elicitation_lifecycle: Option, elicitation_router: ElicitationRequestRouter, ) -> Self { let mut required_servers = mcp_servers @@ -157,6 +158,7 @@ impl McpConnectionManager { approval_policy.value(), initial_permission_profile, elicitation_reviewer, + elicitation_lifecycle, elicitation_router, ); let tool_plugin_provenance = Arc::new(tool_plugin_provenance); @@ -360,6 +362,7 @@ impl McpConnectionManager { approval_policy.value(), permission_profile.clone(), /*reviewer*/ None, + /*lifecycle*/ None, ElicitationRequestRouter::default(), ), startup_cancellation_token: CancellationToken::new(), diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index 1f61ab700e..41fc19a594 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -2,6 +2,7 @@ use super::*; use crate::codex_apps_cache::CodexAppsToolsCache; use crate::codex_apps_cache::CodexAppsToolsCacheContext; use crate::declared_openai_file_input_param_names; +use crate::elicitation::ElicitationLifecycle; use crate::elicitation::ElicitationRequestManager; use crate::elicitation::ElicitationRequestRouter; use crate::elicitation::elicitation_is_rejected_by_policy; @@ -323,6 +324,7 @@ async fn disabled_permissions_auto_accept_elicitation_with_empty_form_schema() { AskForApproval::Never, PermissionProfile::Disabled, /*reviewer*/ None, + /*lifecycle*/ None, ElicitationRequestRouter::default(), ); let (tx_event, _rx_event) = async_channel::bounded(1); @@ -359,6 +361,7 @@ async fn disabled_permissions_do_not_auto_accept_elicitation_with_requested_fiel AskForApproval::Never, PermissionProfile::Disabled, /*reviewer*/ None, + /*lifecycle*/ None, ElicitationRequestRouter::default(), ); let (tx_event, _rx_event) = async_channel::bounded(1); @@ -395,17 +398,35 @@ async fn disabled_permissions_do_not_auto_accept_elicitation_with_requested_fiel #[tokio::test] async fn shared_elicitation_router_targets_the_exact_pending_request() { + struct Registration(Arc); + + impl Drop for Registration { + fn drop(&mut self) { + self.0.fetch_sub(1, std::sync::atomic::Ordering::SeqCst); + } + } + let router = ElicitationRequestRouter::default(); + let outstanding = Arc::new(AtomicUsize::new(0)); + let lifecycle = ElicitationLifecycle::new({ + let outstanding = outstanding.clone(); + move || { + outstanding.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Registration(outstanding.clone()) + } + }); let manager_a = ElicitationRequestManager::new( AskForApproval::OnRequest, PermissionProfile::default(), /*reviewer*/ None, + Some(lifecycle.clone()), router.clone(), ); let manager_b = ElicitationRequestManager::new( AskForApproval::OnRequest, PermissionProfile::default(), /*reviewer*/ None, + Some(lifecycle), router, ); let (tx_event, rx_event) = async_channel::bounded(2); @@ -435,6 +456,7 @@ async fn shared_elicitation_router_targets_the_exact_pending_request() { else { panic!("expected elicitation request"); }; + assert_eq!(outstanding.load(std::sync::atomic::Ordering::SeqCst), 2); let ( codex_protocol::mcp::RequestId::String(request_a_id), codex_protocol::mcp::RequestId::String(request_b_id), @@ -485,6 +507,7 @@ async fn shared_elicitation_router_targets_the_exact_pending_request() { .expect("request B response"), response_b ); + assert_eq!(outstanding.load(std::sync::atomic::Ordering::SeqCst), 0); } #[test] @@ -1539,6 +1562,7 @@ async fn no_local_runtime_fails_local_stdio_but_keeps_local_http_server() { ToolPluginProvenance::default(), /*auth*/ None, /*elicitation_reviewer*/ None, + /*elicitation_lifecycle*/ None, ElicitationRequestRouter::default(), ) .await; diff --git a/codex-rs/codex-mcp/src/elicitation.rs b/codex-rs/codex-mcp/src/elicitation.rs index e4fcf49d22..b6bb7fa791 100644 --- a/codex-rs/codex-mcp/src/elicitation.rs +++ b/codex-rs/codex-mcp/src/elicitation.rs @@ -52,6 +52,33 @@ pub trait ElicitationReviewer: Send + Sync { pub type ElicitationReviewerHandle = Arc; +/// Holds an owner-provided registration while an MCP elicitation is waiting for a response. +#[derive(Clone)] +pub struct ElicitationLifecycle { + register: Arc Box + Send + Sync>, +} + +impl ElicitationLifecycle { + pub fn new(register: impl Fn() -> T + Send + Sync + 'static) -> Self + where + T: Send + Sync + 'static, + { + Self { + register: Arc::new(move || Box::new(register())), + } + } + + fn start(&self) -> ActiveElicitation { + ActiveElicitation { + _registration: (self.register)(), + } + } +} + +struct ActiveElicitation { + _registration: Box, +} + /// Routes model-visible elicitation response tokens to their exact pending responders. /// /// One router is shared by every MCP runtime created for a thread. The public response token is @@ -86,6 +113,7 @@ pub(crate) struct ElicitationRequestManager { pub(crate) permission_profile: Arc>, auto_deny: Arc>, reviewer: Option, + lifecycle: Option, } impl ElicitationRequestManager { @@ -93,6 +121,7 @@ impl ElicitationRequestManager { approval_policy: AskForApproval, permission_profile: PermissionProfile, reviewer: Option, + lifecycle: Option, router: ElicitationRequestRouter, ) -> Self { Self { @@ -101,6 +130,7 @@ impl ElicitationRequestManager { permission_profile: Arc::new(StdMutex::new(permission_profile)), auto_deny: Arc::new(StdMutex::new(false)), reviewer, + lifecycle, } } @@ -140,6 +170,7 @@ impl ElicitationRequestManager { let permission_profile = self.permission_profile.clone(); let auto_deny = self.auto_deny.clone(); let reviewer = self.reviewer.clone(); + let lifecycle = self.lifecycle.clone(); Box::new(move |id, elicitation| { let router = router.clone(); let tx_event = tx_event.clone(); @@ -148,6 +179,7 @@ impl ElicitationRequestManager { let permission_profile = permission_profile.clone(); let auto_deny = auto_deny.clone(); let reviewer = reviewer.clone(); + let lifecycle = lifecycle.clone(); async move { let auto_deny = auto_deny .lock() @@ -249,6 +281,7 @@ impl ElicitationRequestManager { }, }; let (tx, rx) = oneshot::channel(); + let _active_elicitation = lifecycle.as_ref().map(ElicitationLifecycle::start); { let mut lock = router.requests.lock().await; lock.insert((server_name.clone(), routed_request_id), tx); diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index eda7ac2399..77128d7970 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -1,5 +1,6 @@ pub use connection_manager::McpConnectionManager; pub use connection_manager::tool_is_model_visible; +pub use elicitation::ElicitationLifecycle; pub use elicitation::ElicitationRequestRouter; pub use elicitation::ElicitationReviewRequest; pub use elicitation::ElicitationReviewer; diff --git a/codex-rs/codex-mcp/src/mcp/mod.rs b/codex-rs/codex-mcp/src/mcp/mod.rs index ffb797f26e..87ea25dd9f 100644 --- a/codex-rs/codex-mcp/src/mcp/mod.rs +++ b/codex-rs/codex-mcp/src/mcp/mod.rs @@ -339,6 +339,7 @@ pub async fn read_mcp_resource( tool_plugin_provenance(config), auth, /*elicitation_reviewer*/ None, + /*elicitation_lifecycle*/ None, crate::elicitation::ElicitationRequestRouter::default(), ) .await; @@ -416,6 +417,7 @@ pub async fn collect_mcp_server_status_snapshot_with_detail( tool_plugin_provenance, auth, /*elicitation_reviewer*/ None, + /*elicitation_lifecycle*/ None, crate::elicitation::ElicitationRequestRouter::default(), ) .await; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 4a562092f1..c80268e2f9 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -1,5 +1,6 @@ use crate::agent::AgentStatus; use crate::config::ConstraintResult; +use crate::elicitation::ElicitationRegistration; use crate::session::Codex; use crate::session::SessionSettingsUpdate; use crate::session::SteerInputError; @@ -162,7 +163,13 @@ pub struct CodexThread { pub(crate) session_source: SessionSource, session_configured: SessionConfiguredEvent, rollout_path: Option, - out_of_band_elicitation_count: Mutex, + out_of_band_elicitations: Mutex, +} + +#[derive(Default)] +struct OutOfBandElicitations { + count: i64, + registration: Option, } #[derive(Debug, Eq, PartialEq)] @@ -187,7 +194,7 @@ impl CodexThread { session_source, session_configured, rollout_path, - out_of_band_elicitation_count: Mutex::new(0), + out_of_band_elicitations: Mutex::new(OutOfBandElicitations::default()), } } @@ -655,38 +662,30 @@ impl CodexThread { self.codex.enabled(feature) } - pub async fn increment_out_of_band_elicitation_count(&self) -> CodexResult { - let mut guard = self.out_of_band_elicitation_count.lock().await; - let was_zero = *guard == 0; - *guard = guard.checked_add(1).ok_or_else(|| { + pub async fn increment_out_of_band_elicitation_count(&self) -> CodexResult { + let mut elicitations = self.out_of_band_elicitations.lock().await; + let incremented = elicitations.count.checked_add(1).ok_or_else(|| { CodexErr::Fatal("out-of-band elicitation count overflowed".to_string()) })?; - - if was_zero { - self.codex - .session - .set_out_of_band_elicitation_pause_state(/*paused*/ true); + if elicitations.count == 0 { + elicitations.registration = Some(self.codex.session.services.elicitations.register()); } - - Ok(*guard) + elicitations.count = incremented; + Ok(incremented) } - pub async fn decrement_out_of_band_elicitation_count(&self) -> CodexResult { - let mut guard = self.out_of_band_elicitation_count.lock().await; - if *guard == 0 { + pub async fn decrement_out_of_band_elicitation_count(&self) -> CodexResult { + let mut elicitations = self.out_of_band_elicitations.lock().await; + if elicitations.count == 0 { return Err(CodexErr::InvalidRequest( "out-of-band elicitation count is already zero".to_string(), )); } - *guard -= 1; - let now_zero = *guard == 0; - if now_zero { - self.codex - .session - .set_out_of_band_elicitation_pause_state(/*paused*/ false); + elicitations.count -= 1; + if elicitations.count == 0 { + elicitations.registration = None; } - - Ok(*guard) + Ok(elicitations.count) } } diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index b45cc841a1..805d56b648 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -283,6 +283,7 @@ pub async fn list_accessible_connectors_from_mcp_tools_with_mcp_manager( ToolPluginProvenance::default(), auth.as_ref(), /*elicitation_reviewer*/ None, + /*elicitation_lifecycle*/ None, codex_mcp::ElicitationRequestRouter::default(), ) .await; diff --git a/codex-rs/core/src/elicitation.rs b/codex-rs/core/src/elicitation.rs new file mode 100644 index 0000000000..b29207af46 --- /dev/null +++ b/codex-rs/core/src/elicitation.rs @@ -0,0 +1,100 @@ +use std::sync::Arc; +use std::sync::Mutex; + +use tokio::sync::watch; + +/// Coordinates user elicitations that pause tool-result delivery for a session. +/// +/// Registrations are counted so concurrent elicitations keep the session paused until all of them +/// finish. Consumers can subscribe to pause timeout progress or wait before returning an already +/// captured result. +#[derive(Clone)] +pub(crate) struct ElicitationService { + inner: Arc, +} + +struct Inner { + state: Mutex, + paused: watch::Sender, +} + +#[derive(Default)] +struct State { + outstanding: i64, +} + +pub(crate) struct ElicitationRegistration { + service: ElicitationService, +} + +impl ElicitationService { + pub(crate) fn new() -> Self { + let (paused, _paused_rx) = watch::channel(false); + Self { + inner: Arc::new(Inner { + state: Mutex::new(State::default()), + paused, + }), + } + } + + pub(crate) fn register(&self) -> ElicitationRegistration { + self.increment(); + ElicitationRegistration { + service: self.clone(), + } + } + + fn increment(&self) { + let mut state = self + .inner + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let was_clear = state.outstanding == 0; + assert_ne!( + state.outstanding, + i64::MAX, + "outstanding elicitation count overflowed" + ); + state.outstanding += 1; + if was_clear { + self.inner.paused.send_replace(true); + } + } + + pub(crate) fn subscribe(&self) -> watch::Receiver { + self.inner.paused.subscribe() + } + + pub(crate) async fn wait_until_clear(&self) { + let mut paused = self.subscribe(); + let _ = paused.wait_for(|paused| !*paused).await; + } + + fn decrement(&self) { + let mut state = self + .inner + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + assert!( + state.outstanding > 0, + "elicitation registration count underflowed" + ); + state.outstanding -= 1; + if state.outstanding == 0 { + self.inner.paused.send_replace(false); + } + } +} + +impl Drop for ElicitationRegistration { + fn drop(&mut self) { + self.service.decrement(); + } +} + +#[cfg(test)] +#[path = "elicitation_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/elicitation_tests.rs b/codex-rs/core/src/elicitation_tests.rs new file mode 100644 index 0000000000..eefe1f5394 --- /dev/null +++ b/codex-rs/core/src/elicitation_tests.rs @@ -0,0 +1,19 @@ +use super::*; + +#[tokio::test] +async fn wait_until_clear_waits_for_every_registration() { + let service = ElicitationService::new(); + let first = service.register(); + let second = service.register(); + let waiting = tokio::spawn({ + let service = service.clone(); + async move { service.wait_until_clear().await } + }); + + drop(first); + tokio::task::yield_now().await; + assert!(!waiting.is_finished()); + + drop(second); + waiting.await.expect("elicitation waiter should complete"); +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index df8753a386..91ccbd6e63 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -40,6 +40,7 @@ pub mod connectors; pub mod context; mod context_manager; mod current_time; +mod elicitation; mod environment_selection; pub mod exec; pub mod exec_env; diff --git a/codex-rs/core/src/mcp_tool_call_tests.rs b/codex-rs/core/src/mcp_tool_call_tests.rs index 6b642b1cd0..f68d32ff4e 100644 --- a/codex-rs/core/src/mcp_tool_call_tests.rs +++ b/codex-rs/core/src/mcp_tool_call_tests.rs @@ -1439,6 +1439,7 @@ async fn host_owned_codex_apps_manager( codex_mcp::ToolPluginProvenance::default(), auth.as_ref(), /*elicitation_reviewer*/ None, + /*elicitation_lifecycle*/ None, codex_mcp::ElicitationRequestRouter::default(), ) .await; diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index 4cfb0a36de..3f5e92eca7 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -195,6 +195,11 @@ impl Session { Arc::new(GuardianMcpElicitationReviewer::new(self)) } + pub(crate) fn mcp_elicitation_lifecycle(&self) -> codex_mcp::ElicitationLifecycle { + let elicitations = self.services.elicitations.clone(); + codex_mcp::ElicitationLifecycle::new(move || elicitations.register()) + } + #[expect( clippy::await_holding_invalid_type, reason = "active turn checks and turn state updates must remain atomic" @@ -222,6 +227,7 @@ impl Session { }; } + let _elicitation = self.services.elicitations.register(); let (tx_response, rx_response) = oneshot::channel(); let prev_entry = { let mut active = self.active_turn.lock().await; @@ -377,6 +383,7 @@ impl Session { tool_plugin_provenance, auth.as_ref(), elicitation_reviewer, + Some(self.mcp_elicitation_lifecycle()), current_runtime.manager().elicitation_router(), ) .await; diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 560dcf5985..ec3fa12f39 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1141,12 +1141,8 @@ impl Session { state.session_configuration.codex_home().clone() } - pub(crate) fn subscribe_out_of_band_elicitation_pause_state(&self) -> watch::Receiver { - self.out_of_band_elicitation_paused.subscribe() - } - - pub(crate) fn set_out_of_band_elicitation_pause_state(&self, paused: bool) { - self.out_of_band_elicitation_paused.send_replace(paused); + pub(crate) fn subscribe_elicitation_pause_state(&self) -> watch::Receiver { + self.services.elicitations.subscribe() } pub(crate) fn get_tx_event(&self) -> Sender { diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 48a9ee447e..2261eaa495 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -30,7 +30,6 @@ pub(crate) struct Session { pub(crate) installation_id: String, pub(super) tx_event: Sender, pub(super) agent_status: watch::Sender, - pub(super) out_of_band_elicitation_paused: watch::Sender, pub(super) state: Mutex, /// Serializes rebuild/apply cycles for the running proxy; each cycle /// rebuilds from the current SessionState while holding this lock. @@ -1057,6 +1056,7 @@ impl Session { unified_exec_manager: UnifiedExecProcessManager::new( config.background_terminal_max_timeout, ), + elicitations: crate::elicitation::ElicitationService::new(), shell_zsh_path: config.zsh_path.clone(), main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(), analytics_events_client, @@ -1125,15 +1125,11 @@ impl Session { tool_search_handler_cache: Default::default(), turn_environments: Arc::clone(&turn_environments), }; - let (out_of_band_elicitation_paused, _out_of_band_elicitation_paused_rx) = - watch::channel(false); - let sess = Arc::new(Session { thread_id, installation_id, tx_event: tx_event.clone(), agent_status, - out_of_band_elicitation_paused, state: Mutex::new(state), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), @@ -1219,6 +1215,7 @@ impl Session { tool_plugin_provenance, auth, Some(sess.mcp_elicitation_reviewer()), + Some(sess.mcp_elicitation_lifecycle()), codex_mcp::ElicitationRequestRouter::default(), ) .instrument(info_span!( diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 9ae29a0917..582a58f2ab 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -5384,6 +5384,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { unified_exec_manager: UnifiedExecProcessManager::new( config.background_terminal_max_timeout, ), + elicitations: crate::elicitation::ElicitationService::new(), shell_zsh_path: None, main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(), analytics_events_client: AnalyticsEventsClient::new( @@ -5494,7 +5495,6 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { installation_id: "11111111-1111-4111-8111-111111111111".to_string(), tx_event, agent_status: agent_status_tx, - out_of_band_elicitation_paused: watch::channel(false).0, state: Mutex::new(state), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), @@ -7510,6 +7510,7 @@ where unified_exec_manager: UnifiedExecProcessManager::new( config.background_terminal_max_timeout, ), + elicitations: crate::elicitation::ElicitationService::new(), shell_zsh_path: None, main_execve_wrapper_exe: config.main_execve_wrapper_exe.clone(), analytics_events_client: AnalyticsEventsClient::new( @@ -7620,7 +7621,6 @@ where installation_id: "11111111-1111-4111-8111-111111111111".to_string(), tx_event, agent_status: agent_status_tx, - out_of_band_elicitation_paused: watch::channel(false).0, state: Mutex::new(state), managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1), features: config.features.clone(), diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index c4b67a629b..43f411e31b 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -10,6 +10,7 @@ use crate::client::ModelClient; use crate::config::NetworkProxyAuditMetadata; use crate::config::StartedNetworkProxy; use crate::current_time::TimeProvider; +use crate::elicitation::ElicitationService; use crate::environment_selection::ThreadEnvironments; use crate::exec_policy::ExecPolicyManager; use crate::guardian::GuardianRejection; @@ -55,6 +56,7 @@ pub(crate) struct SessionServices { pub(crate) mcp_projection_lock: Mutex<()>, pub(crate) mcp_startup_cancellation_token: Mutex, pub(crate) unified_exec_manager: UnifiedExecProcessManager, + pub(crate) elicitations: ElicitationService, #[cfg_attr(not(unix), allow(dead_code))] pub(crate) shell_zsh_path: Option, #[cfg_attr(not(unix), allow(dead_code))] diff --git a/codex-rs/core/src/tools/code_mode/execute_handler.rs b/codex-rs/core/src/tools/code_mode/execute_handler.rs index 2e46c1bd6b..8374ae1ca4 100644 --- a/codex-rs/core/src/tools/code_mode/execute_handler.rs +++ b/codex-rs/core/src/tools/code_mode/execute_handler.rs @@ -85,6 +85,7 @@ impl CodeModeExecuteHandler { .code_mode_service .finish_cell_dispatch(&cell_id); } + exec.session.services.elicitations.wait_until_clear().await; handle_runtime_response(&exec, response, args.max_output_tokens, started_at) .await .map_err(FunctionCallError::RespondToModel) diff --git a/codex-rs/core/src/tools/code_mode/wait_handler.rs b/codex-rs/core/src/tools/code_mode/wait_handler.rs index 953391a573..c3c867f3ef 100644 --- a/codex-rs/core/src/tools/code_mode/wait_handler.rs +++ b/codex-rs/core/src/tools/code_mode/wait_handler.rs @@ -120,6 +120,7 @@ impl CodeModeWaitHandler { .code_mode_service .finish_cell_dispatch(runtime_cell_id); } + exec.session.services.elicitations.wait_until_clear().await; handle_runtime_response(&exec, wait_response.into(), args.max_tokens, started_at) .await .map(boxed_tool_output) diff --git a/codex-rs/core/src/unified_exec/mod_tests.rs b/codex-rs/core/src/unified_exec/mod_tests.rs index 54ee788e3b..81403c8e52 100644 --- a/codex-rs/core/src/unified_exec/mod_tests.rs +++ b/codex-rs/core/src/unified_exec/mod_tests.rs @@ -160,7 +160,7 @@ async fn exec_command_with_tty( &output_closed, &output_closed_notify, &cancellation_token, - Some(session.subscribe_out_of_band_elicitation_pause_state()), + Some(session.subscribe_elicitation_pause_state()), deadline, ) .await; @@ -515,12 +515,11 @@ async fn unified_exec_pause_blocks_yield_timeout() -> anyhow::Result<()> { skip_if_sandbox!(Ok(())); let (session, turn) = test_session_and_turn().await; - session.set_out_of_band_elicitation_pause_state(/*paused*/ true); + let elicitation = session.services.elicitations.register(); - let paused_session = Arc::clone(&session); tokio::spawn(async move { tokio::time::sleep(Duration::from_secs(2)).await; - paused_session.set_out_of_band_elicitation_pause_state(/*paused*/ false); + drop(elicitation); }); let started = tokio::time::Instant::now(); diff --git a/codex-rs/core/src/unified_exec/process_manager.rs b/codex-rs/core/src/unified_exec/process_manager.rs index ef22918cd2..a895708f79 100644 --- a/codex-rs/core/src/unified_exec/process_manager.rs +++ b/codex-rs/core/src/unified_exec/process_manager.rs @@ -493,11 +493,7 @@ impl UnifiedExecProcessManager { &output_closed, &output_closed_notify, &cancellation_token, - Some( - context - .session - .subscribe_out_of_band_elicitation_pause_state(), - ), + Some(context.session.subscribe_elicitation_pause_state()), deadline, ) .await; @@ -839,7 +835,7 @@ impl UnifiedExecProcessManager { let pause_state = entry .session .upgrade() - .map(|session| session.subscribe_out_of_band_elicitation_pause_state()); + .map(|session| session.subscribe_elicitation_pause_state()); let session = entry.session.upgrade(); Ok(PreparedProcessHandles { diff --git a/codex-rs/core/tests/suite/code_mode.rs b/codex-rs/core/tests/suite/code_mode.rs index 26c041d8f7..9293b506ef 100644 --- a/codex-rs/core/tests/suite/code_mode.rs +++ b/codex-rs/core/tests/suite/code_mode.rs @@ -521,6 +521,76 @@ text(JSON.stringify(await tools.exec_command({ cmd: "printf code_mode_exec_marke Ok(()) } +#[tokio::test] +async fn code_mode_exec_holds_captured_result_during_elicitation() -> Result<()> { + let server = responses::start_mock_server().await; + let mut builder = test_codex() + .with_model("test-gpt-5.1-codex") + .with_config(|config| { + let _ = config.features.enable(Feature::CodeMode); + }); + let test = builder.build(&server).await?; + + let first_mock = responses::mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + ev_custom_tool_call("call-1", "exec", "text('captured');"), + ev_completed("resp-1"), + ]), + ) + .await; + let second_mock = responses::mount_sse_once( + &server, + sse(vec![ + ev_assistant_message("msg-1", "done"), + ev_completed("resp-2"), + ]), + ) + .await; + + assert_eq!( + test.codex.increment_out_of_band_elicitation_count().await?, + 1 + ); + assert_eq!( + test.codex.increment_out_of_band_elicitation_count().await?, + 2 + ); + let release_elicitation = async { + tokio::time::timeout(Duration::from_secs(5), async { + while first_mock.requests().is_empty() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("initial response request should arrive"); + tokio::time::sleep(Duration::from_millis(250)).await; + assert!( + second_mock.requests().is_empty(), + "captured exec result should not return during an elicitation" + ); + assert_eq!( + test.codex.decrement_out_of_band_elicitation_count().await?, + 1 + ); + tokio::time::sleep(Duration::from_millis(100)).await; + assert!( + second_mock.requests().is_empty(), + "captured exec result should wait for every elicitation" + ); + assert_eq!( + test.codex.decrement_out_of_band_elicitation_count().await?, + 0 + ); + Ok::<(), anyhow::Error>(()) + }; + + tokio::try_join!(test.submit_turn("run code mode"), release_elicitation)?; + second_mock.single_request(); + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn code_mode_only_restricts_prompt_tools() -> Result<()> { skip_if_no_network!(Ok(()));