Complete dynamic tool lifecycles on cancellation (#51556)

## Why

Interrupting a turn or terminating a Code Mode cell can drop a pending dynamic tool handler before it publishes its completion. Cancellation needs to settle the call in live events and persisted history without allowing a late response into model context.

## What changed

- Let dynamic tool handlers observe cancellation and publish their terminal item before dispatch finishes. Calls cancelled while awaiting a response complete with a failed item and a cancellation error.
- Keep readiness, dispatch admission, registry lock waits, hooks, and output processing cancellable so cancellation can finish even when a hook is blocked.
- Report an aborted tool lifecycle outcome when cancellation interrupts dispatch, including before handler execution.

## Testing

Add regression tests for turn interruption with pending calls or blocked hooks, cell termination with persisted failure and late-response exclusion, cancellation before dispatch admission, and exactly one completion when a response races cancellation.

GitOrigin-RevId: 0112fc74c8652c91503ed5bbfb334ae2e26930ef

codex-rs/core/src/function_tool.rs

@@ -1 +1,24 @@
+use std::future::Future;
+
+use codex_async_utils::OrCancelExt;
 pub use codex_tools::FunctionCallError;
+use tokio_util::sync::CancellationToken;
+
+/// Maps cancellation to the standard tool-call error while preserving the future's output.
+pub(crate) trait OrCancelToolExt: OrCancelExt + Send {
+    fn or_cancel_tool(
+        self,
+        token: &CancellationToken,
+    ) -> impl Future<Output = Result<Self::Output, FunctionCallError>> + Send;
+}
+
+impl<F: OrCancelExt + Send> OrCancelToolExt for F {
+    async fn or_cancel_tool(
+        self,
+        token: &CancellationToken,
+    ) -> Result<Self::Output, FunctionCallError> {
+        self.or_cancel(token)
+            .await
+            .map_err(|_| FunctionCallError::RespondToModel("tool call cancelled".to_string()))
+    }
+}

codex-rs/core/src/tools/handlers/dynamic.rs

@@ -9,6 +9,7 @@ use crate::tools::handlers::parse_arguments;
 use crate::tools::registry::CoreToolRuntime;
 use crate::tools::registry::ToolExecutor;
 use crate::tools::registry::ToolExposure;
+use codex_async_utils::OrCancelExt;
 use codex_protocol::dynamic_tools::DynamicToolFunctionSpec;
 use codex_protocol::dynamic_tools::DynamicToolNamespaceSpec;
 use codex_protocol::dynamic_tools::DynamicToolResponse;
@@ -27,6 +28,7 @@ use codex_tools::dynamic_tool_to_responses_api_tool;
 use serde_json::Value;
 use std::time::Instant;
 use tokio::sync::oneshot;
+use tokio_util::sync::CancellationToken;
 use tracing::warn;
 
 pub struct DynamicToolHandler {
@@ -123,6 +125,7 @@ impl DynamicToolHandler {
             turn,
             call_id,
             payload,
+            cancellation_token,
             ..
         } = invocation;
 
@@ -142,6 +145,7 @@ impl DynamicToolHandler {
             call_id,
             self.tool_name.clone(),
             args,
+            cancellation_token,
         )
         .await
         .ok_or_else(|| {
@@ -166,6 +170,10 @@ impl DynamicToolHandler {
 }
 
 impl CoreToolRuntime for DynamicToolHandler {
+    fn finishes_on_cancellation(&self) -> bool {
+        true
+    }
+
     fn is_third_party_tool(&self) -> bool {
         true
     }
@@ -181,19 +189,26 @@ async fn request_dynamic_tool(
     call_id: String,
     tool_name: ToolName,
     arguments: Value,
+    cancellation_token: CancellationToken,
 ) -> Option<DynamicToolResponse> {
+    if cancellation_token.is_cancelled() {
+        return None;
+    }
     let namespace = tool_name.namespace;
     let tool = tool_name.name;
     let (tx_response, rx_response) = oneshot::channel();
     let event_id = call_id.clone();
     let prev_entry = {
         let mut active = session.active_turn.lock().await;
+        if cancellation_token.is_cancelled() {
+            return None;
+        }
         match active.as_mut() {
             Some(at) => {
                 let mut ts = at.turn_state.lock().await;
                 ts.insert_pending_dynamic_tool(call_id.clone(), tx_response)
             }
-            None => None,
+            None => return None,
         }
     };
     if prev_entry.is_some() {
@@ -217,7 +232,11 @@ async fn request_dynamic_tool(
             }),
         )
         .await;
-    let response = rx_response.await.ok();
+    let response = rx_response
+        .or_cancel(&cancellation_token)
+        .await
+        .ok()
+        .and_then(Result::ok);
 
     let item = match &response {
         Some(response) => DynamicToolCallItem {
@@ -253,3 +272,7 @@ async fn request_dynamic_tool(
 
     response
 }
+
+#[cfg(test)]
+#[path = "dynamic_tests.rs"]
+mod tests;

codex-rs/core/src/tools/handlers/dynamic_tests.rs

@@ -0,0 +1,76 @@
+//! Dynamic-tool handlers finish their item lifecycle when cancellation is signalled.
+
+use super::*;
+use crate::state::ActiveTurn;
+use codex_protocol::protocol::Event;
+use codex_protocol::protocol::EventMsg;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use std::sync::Arc;
+use std::time::Duration;
+
+async fn next_item(receiver: &async_channel::Receiver<Event>) -> DynamicToolCallItem {
+    tokio::time::timeout(Duration::from_secs(10), async {
+        loop {
+            let item = match receiver.recv().await.expect("event channel open").msg {
+                EventMsg::ItemStarted(event) => event.item,
+                EventMsg::ItemCompleted(event) => event.item,
+                _ => continue,
+            };
+            if let TurnItem::DynamicToolCall(item) = item {
+                return item;
+            }
+        }
+    })
+    .await
+    .expect("dynamic tool item should be emitted")
+}
+
+#[tokio::test]
+async fn response_cancel_race_completes_once() {
+    let (session, turn, receiver) = crate::session::tests::make_session_and_context_with_rx().await;
+    *session.active_turn.lock().await = Some(ActiveTurn::default());
+    let cancellation = CancellationToken::new();
+    let lifecycle = tokio::spawn({
+        let session = Arc::clone(&session);
+        let cancellation = cancellation.clone();
+        async move {
+            request_dynamic_tool(
+                &session,
+                &turn,
+                "call-1".to_string(),
+                ToolName::plain("gate"),
+                json!({}),
+                cancellation,
+            )
+            .await
+        }
+    });
+    next_item(&receiver).await;
+    let response = DynamicToolResponse {
+        content_items: Vec::new(),
+        success: true,
+    };
+    session
+        .notify_dynamic_tool_response("call-1", response.clone())
+        .await;
+    cancellation.cancel();
+    let result = lifecycle.await.expect("lifecycle joined");
+    assert!(result.is_none() || result == Some(response));
+    let item = next_item(&receiver).await;
+    assert_eq!(
+        (item.status, item.success, item.error),
+        if result.is_some() {
+            (DynamicToolCallStatus::Completed, Some(true), None)
+        } else {
+            (
+                DynamicToolCallStatus::Failed,
+                Some(false),
+                Some("dynamic tool call was cancelled before receiving a response".to_string()),
+            )
+        }
+    );
+    while let Ok(event) = receiver.try_recv() {
+        assert!(!matches!(event.msg, EventMsg::ItemCompleted(_)));
+    }
+}

codex-rs/core/src/tools/parallel.rs

@@ -14,6 +14,7 @@ use tracing::instrument;
 use tracing::trace_span;
 
 use crate::function_tool::FunctionCallError;
+use crate::function_tool::OrCancelToolExt;
 use crate::session::session::Session;
 use crate::session::step_context::StepContext;
 use crate::tools::call_trace;
@@ -139,6 +140,9 @@ impl ToolCallRuntime {
         let router = &step_context.tool_router;
         let supports_parallel = router.tool_supports_parallel(&call);
         let tool_runtime = router.tool_runtime(&call.tool_name);
+        let finishes_on_cancellation = tool_runtime
+            .as_ref()
+            .is_some_and(|runtime| runtime.finishes_on_cancellation());
         let router = Arc::clone(router);
         let session = Arc::clone(&self.session);
         let turn = Arc::clone(&step_context.turn);
@@ -196,17 +200,22 @@ impl ToolCallRuntime {
         let mut dispatch_handle = AbortOnDropHandle::new(tokio::spawn(
             async move {
                 let _message_admission = message_admission?;
-                if let Some(tool_runtime) = tool_runtime
-                    && let Some(readiness) = tool_runtime.wait_until_ready(&session)
-                {
-                    readiness.await;
-                }
+                // Readiness and admission remain cancellable even when the handler must finish.
+                let guard = async {
+                    if let Some(tool_runtime) = tool_runtime
+                        && let Some(readiness) = tool_runtime.wait_until_ready(&session)
+                    {
+                        readiness.await;
+                    }
 
-                let guard = if supports_parallel {
-                    Either::Left(lock.read().await)
-                } else {
-                    Either::Right(lock.write().await)
-                };
+                    if supports_parallel {
+                        Either::Left(lock.read().await)
+                    } else {
+                        Either::Right(lock.write().await)
+                    }
+                }
+                .or_cancel_tool(&invocation_cancellation_token)
+                .await?;
                 // Admission through the parallel-execution gate marks the end
                 // of dispatch waiting and the start of handler execution.
                 if let Some(execution_started_at) = execution_started_at {
@@ -249,7 +258,11 @@ impl ToolCallRuntime {
             let mut result = tokio::select! {
                 res = &mut dispatch_handle => res.map_err(Self::tool_task_join_error)?,
                 _ = cancellation_token.cancelled() => {
-                    if call_state.terminal_outcome_reached.load(Ordering::Acquire) || dispatch_handle.is_finished() {
+                    // Dynamic handlers observe the token and publish their terminal item themselves.
+                    if finishes_on_cancellation
+                        || call_state.terminal_outcome_reached.load(Ordering::Acquire)
+                        || dispatch_handle.is_finished()
+                    {
                         dispatch_handle.await.map_err(Self::tool_task_join_error)?
                     } else {
                         let secs = started.elapsed().as_secs_f32().max(0.1);
@@ -266,12 +279,13 @@ impl ToolCallRuntime {
                                     &call.call_id,
                                     trace_source,
                                 );
+                                call_state.terminal_outcome_reached.store(true, Ordering::Release);
                                 notify_tool_aborted(
                                     abort_session.as_ref(),
                                     abort_turn.as_ref(),
                                     call.call_id.as_str(),
                                     &call.tool_name,
-                                    abort_source,
+                                    abort_source.clone(),
                                 )
                                 .await;
                                 Ok(response)
@@ -281,6 +295,25 @@ impl ToolCallRuntime {
                     }
                 },
             };
+            // Cancellation outside the handler can return before the registry reports an outcome.
+            if cancellation_token.is_cancelled()
+                && !call_state
+                    .terminal_outcome_reached
+                    .swap(true, Ordering::AcqRel)
+            {
+                notify_tool_aborted(
+                    abort_session.as_ref(),
+                    abort_turn.as_ref(),
+                    call.call_id.as_str(),
+                    &call.tool_name,
+                    abort_source,
+                )
+                .await;
+                result = Ok(Self::aborted_response(
+                    &call,
+                    started.elapsed().as_secs_f32().max(0.1),
+                ));
+            }
             // Use one completion measurement for logging and response formatting.
             // Measuring inside a handler would omit routing and output processing.
             if let Some(timing) = tool_call_timing_guard.as_mut()
@@ -568,16 +601,33 @@ mod tests {
         });
     }
 
+    #[test_case::test_case(false; "ordinary_handler")]
+    #[test_case::test_case(true; "dynamic_handler")]
     #[tokio::test]
-    async fn cancellation_before_dispatch_admission_logs_dispatch_only_timing() -> anyhow::Result<()>
-    {
+    async fn cancellation_before_dispatch_admission_logs_dispatch_only_timing(
+        dynamic: bool,
+    ) -> anyhow::Result<()> {
         let (session, turn_context) = crate::session::tests::make_session_and_context().await;
         let session = Arc::new(session);
         let turn_context = Arc::new(turn_context);
         let tool_name = codex_tools::ToolName::plain("test_tool");
-        let handler = Arc::new(ImmediateHandler {
-            tool_name: tool_name.clone(),
-        }) as Arc<dyn CoreToolRuntime>;
+        let handler: Arc<dyn CoreToolRuntime> = if dynamic {
+            Arc::new(
+                crate::tools::handlers::DynamicToolHandler::new(
+                    &codex_protocol::dynamic_tools::DynamicToolFunctionSpec {
+                        name: tool_name.name.clone(),
+                        description: "test".to_string(),
+                        input_schema: serde_json::json!({"type": "object"}),
+                        defer_loading: false,
+                    },
+                )
+                .expect("dynamic handler"),
+            )
+        } else {
+            Arc::new(ImmediateHandler {
+                tool_name: tool_name.clone(),
+            })
+        };
         let step_context = StepContext::for_test(Arc::clone(&turn_context));
         let router = Arc::new(ToolRouter::from_parts(
             ToolRegistry::from_tools([handler]),

codex-rs/core/src/tools/registry.rs

@@ -4,6 +4,7 @@ use std::sync::atomic::Ordering;
 use std::time::Duration;
 
 use crate::function_tool::FunctionCallError;
+use crate::function_tool::OrCancelToolExt;
 use crate::hook_runtime::PreToolUseHookResult;
 use crate::hook_runtime::record_additional_contexts;
 use crate::hook_runtime::run_post_tool_use_hooks;
@@ -54,6 +55,11 @@ pub use codex_tools::ToolExposure;
 /// Implementers provide the shared `ToolExecutor` behavior plus optional
 /// core-owned metadata for hooks, telemetry, tool search, and argument diffs.
 pub(crate) trait CoreToolRuntime: ToolExecutor<ToolInvocation> {
+    /// The handler observes its cancellation token and must finish before dispatch is aborted.
+    fn finishes_on_cancellation(&self) -> bool {
+        false
+    }
+
     /// Whether this built-in control tool needs a structured tool-call event.
     fn is_builtin_control_tool(&self) -> bool {
         false
@@ -535,6 +541,7 @@ impl ToolRegistry {
         mut invocation: ToolInvocation,
         call_state: Option<Arc<ToolCallState>>,
     ) -> Result<AnyToolResult, FunctionCallError> {
+        let cancellation_token = invocation.cancellation_token.clone();
         let tool_name = invocation.tool_name.clone();
         let call_id_owned = invocation.call_id.clone();
         let otel = invocation
@@ -547,9 +554,18 @@ impl ToolRegistry {
         let sandbox_tags = invocation.turn.turn_metadata_state.sandbox_tags;
 
         {
-            let mut active = invocation.session.active_turn.lock().await;
+            let mut active = invocation
+                .session
+                .active_turn
+                .lock()
+                .or_cancel_tool(&cancellation_token)
+                .await?;
             if let Some(active_turn) = active.as_mut() {
-                let mut turn_state = active_turn.turn_state.lock().await;
+                let mut turn_state = active_turn
+                    .turn_state
+                    .lock()
+                    .or_cancel_tool(&cancellation_token)
+                    .await?;
                 turn_state.tool_calls = turn_state.tool_calls.saturating_add(1);
             }
         }
@@ -614,7 +630,8 @@ impl ToolRegistry {
                 &pre_tool_use_payload.tool_name,
                 &pre_tool_use_payload.tool_input,
             )
-            .await
+            .or_cancel_tool(&cancellation_token)
+            .await?
             {
                 PreToolUseHookResult::Blocked(message) => {
                     if tool.is_builtin_control_tool() {
@@ -661,7 +678,9 @@ impl ToolRegistry {
         }
 
         if tool.mcp_server_name().is_none() {
-            notify_tool_start(&invocation, /*mcp_tool*/ None).await;
+            notify_tool_start(&invocation, /*mcp_tool*/ None)
+                .or_cancel_tool(&cancellation_token)
+                .await?;
         }
         let mut control_tool_analytics = tool
             .is_builtin_control_tool()
@@ -732,7 +751,8 @@ impl ToolRegistry {
                     post_tool_use_payload.tool_input,
                     post_tool_use_payload.tool_response,
                 )
-                .await,
+                .or_cancel_tool(&cancellation_token)
+                .await?,
             )
         } else {
             None
@@ -743,11 +763,17 @@ impl ToolRegistry {
                 &invocation.turn,
                 outcome.additional_contexts.clone(),
             )
-            .await;
+            .or_cancel_tool(&cancellation_token)
+            .await?;
         }
 
         // A PostToolUse block rejects the result, not the already-completed tool execution.
         let lifecycle_outcome = match &result {
+            _ if tool.finishes_on_cancellation()
+                && invocation.cancellation_token.is_cancelled() =>
+            {
+                ToolCallOutcome::Aborted
+            }
             Ok(_) => ToolCallOutcome::Completed { success },
             Err(_) => ToolCallOutcome::Failed {
                 handler_executed: true,
@@ -832,7 +858,8 @@ async fn handle_any_tool(
             invocation.session.thread_id,
             "tool_output",
         )
-        .await;
+        .or_cancel_tool(&invocation.cancellation_token)
+        .await?;
     }
     Ok(result)
 }

codex-rs/core/tests/suite/dynamic_tool_cancellation.rs

@@ -0,0 +1,348 @@
+//! Dynamic-call cancellation settles live events, persisted history, and model context.
+
+use anyhow::Result;
+use codex_core::StartThreadOptions;
+use codex_core::TurnInputRequest;
+use codex_extension_api::ExtensionFuture;
+use codex_extension_api::ExtensionRegistryBuilder;
+use codex_extension_api::ToolCallOutcome;
+use codex_extension_api::ToolFinishInput;
+use codex_extension_api::ToolLifecycleContributor;
+use codex_features::Feature;
+use codex_history::RolloutItem;
+use codex_protocol::dynamic_tools::DynamicToolCallOutputContentItem;
+use codex_protocol::dynamic_tools::DynamicToolFunctionSpec;
+use codex_protocol::dynamic_tools::DynamicToolResponse;
+use codex_protocol::dynamic_tools::DynamicToolSpec;
+use codex_protocol::items::DynamicToolCallStatus;
+use codex_protocol::items::TurnItem;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::HookEventName;
+use codex_protocol::protocol::Op;
+use codex_protocol::protocol::ThreadHistoryMode;
+use codex_protocol::user_input::UserInput;
+use core_test_support::hooks::trust_discovered_hooks;
+use core_test_support::responses;
+use core_test_support::responses::ev_completed;
+use core_test_support::responses::ev_custom_tool_call;
+use core_test_support::responses::ev_function_call;
+use core_test_support::responses::ev_response_created;
+use core_test_support::responses::sse;
+use core_test_support::skip_if_no_network;
+use core_test_support::test_codex::test_codex;
+use core_test_support::wait_for_event;
+use core_test_support::wait_for_event_match;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use std::sync::Arc;
+use std::sync::Mutex;
+
+#[derive(Default)]
+struct ToolOutcomes(Mutex<Vec<ToolCallOutcome>>);
+
+impl ToolLifecycleContributor for ToolOutcomes {
+    fn on_tool_finish<'a>(&'a self, input: ToolFinishInput<'a>) -> ExtensionFuture<'a, ()> {
+        Box::pin(async move {
+            self.0
+                .lock()
+                .expect("tool outcomes lock poisoned")
+                .push(input.outcome);
+        })
+    }
+}
+
+#[test_case::test_case(None; "handler")]
+#[test_case::test_case(Some(HookEventName::PreToolUse); "pre_hook")]
+#[test_case::test_case(Some(HookEventName::PostToolUse); "post_hook")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn interrupt_completes_cancelled_dynamic_call(hook: Option<HookEventName>) -> Result<()> {
+    skip_if_no_network!(Ok(()));
+    let server = responses::start_mock_server().await;
+    let outcomes = Arc::new(ToolOutcomes::default());
+    let mut extensions = ExtensionRegistryBuilder::new();
+    extensions.tool_lifecycle_contributor(outcomes.clone());
+    let test =
+        test_codex()
+            .with_extensions(Arc::new(extensions.build()))
+            .with_pre_build_hook(move |home| {
+                if let Some(hook) = hook {
+                    let name = match hook {
+                        HookEventName::PreToolUse => "PreToolUse",
+                        HookEventName::PostToolUse => "PostToolUse",
+                        _ => unreachable!(),
+                    };
+                    std::fs::write(home.join("hooks.json"), json!({"hooks": {
+                    name: [{"matcher": "^gate$", "hooks": [{"type": "command",
+                        "command": "python3 -c 'import threading; threading.Event().wait()'"}]}]
+                }}).to_string()).expect("write blocking hook fixture");
+                }
+            })
+            .with_config(move |config| {
+                if hook.is_some() {
+                    trust_discovered_hooks(config);
+                }
+            })
+            .build_with_auto_env(&server)
+            .await?;
+    let thread = test
+        .thread_manager
+        .start_thread(StartThreadOptions {
+            dynamic_tools: vec![DynamicToolSpec::Function(DynamicToolFunctionSpec {
+                name: "gate".to_string(),
+                description: "Wait for a host reply.".to_string(),
+                input_schema: json!({"type": "object"}),
+                defer_loading: false,
+            })],
+            ..StartThreadOptions::new(test.config.clone())
+        })
+        .await?
+        .thread;
+    responses::mount_sse_once(
+        &server,
+        sse(vec![
+            ev_response_created("run"),
+            ev_function_call("gate-call", "gate", "{}"),
+            ev_completed("run"),
+        ]),
+    )
+    .await;
+    thread
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
+            text: "start a call".to_string(),
+            text_elements: Vec::new(),
+        }]))
+        .await?;
+    if hook != Some(HookEventName::PreToolUse) {
+        wait_for_event(&thread, |event| {
+            matches!(event, EventMsg::DynamicToolCallRequest(_))
+        })
+        .await;
+    }
+    if hook == Some(HookEventName::PostToolUse) {
+        thread
+            .submit(Op::DynamicToolResponse {
+                id: "gate-call".to_string(),
+                response: DynamicToolResponse {
+                    content_items: Vec::new(),
+                    success: true,
+                },
+            })
+            .await?;
+    }
+    if let Some(hook) = hook {
+        wait_for_event(
+            &thread,
+            |event| matches!(event, EventMsg::HookStarted(event) if event.run.event_name == hook),
+        )
+        .await;
+    }
+    thread.submit(Op::Interrupt).await?;
+    if hook.is_none() {
+        let item = wait_for_event_match(&thread, |event| match event {
+            EventMsg::ItemCompleted(event) => match &event.item {
+                TurnItem::DynamicToolCall(item) => Some(item.clone()),
+                _ => None,
+            },
+            EventMsg::TurnAborted(_) => panic!("turn aborted before the dynamic call completed"),
+            _ => None,
+        })
+        .await;
+        assert_eq!(
+            (item.id.as_str(), item.status, item.success, item.error),
+            (
+                "gate-call",
+                DynamicToolCallStatus::Failed,
+                Some(false),
+                Some("dynamic tool call was cancelled before receiving a response".to_string())
+            )
+        );
+    }
+    wait_for_event(&thread, |event| matches!(event, EventMsg::TurnAborted(_))).await;
+    // Forced turn abort would drop dispatch without its tool callback. Cancellation
+    // must finish before that fallback even while the hook stays blocked.
+    assert_eq!(
+        *outcomes.0.lock().expect("tool outcomes lock poisoned"),
+        vec![ToolCallOutcome::Aborted]
+    );
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn terminated_cell_persists_failure_and_excludes_late_reply_from_context() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+    let server = responses::start_mock_server().await;
+    let test = test_codex()
+        .with_model("test-gpt-5.1-codex")
+        .with_config(|config| {
+            let _ = config.features.enable(Feature::CodeMode);
+        })
+        .build_with_auto_env(&server)
+        .await?;
+    let thread = test
+        .thread_manager
+        .start_thread(StartThreadOptions {
+            environments: Some(vec![test.executor_environment().request()]),
+            dynamic_tools: vec![DynamicToolSpec::Function(DynamicToolFunctionSpec {
+                name: "lookup".to_string(),
+                description: "Look up a value.".to_string(),
+                input_schema: json!({"type": "object"}),
+                defer_loading: false,
+            })],
+            history_mode: Some(ThreadHistoryMode::Paginated),
+            ..StartThreadOptions::new(test.config.clone())
+        })
+        .await?
+        .thread;
+    let setup = responses::mount_sse_sequence(
+        &server,
+        vec![
+            sse(vec![
+                ev_custom_tool_call(
+                    "exec-call",
+                    "exec",
+                    "const pending = tools.lookup({}); yield_control(); text(await pending);",
+                ),
+                ev_completed("exec"),
+            ]),
+            sse(vec![
+                ev_function_call("before", "lookup", "{}"),
+                ev_completed("before"),
+            ]),
+        ],
+    )
+    .await;
+    thread
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
+            text: "Start a lookup, then terminate its cell.".to_string(),
+            text_elements: Vec::new(),
+        }]))
+        .await?;
+    // A direct call holds the turn open until the nested call has registered.
+    let mut calls = Vec::new();
+    for _ in 0..2 {
+        calls.push(
+            wait_for_event_match(&thread, |event| match event {
+                EventMsg::DynamicToolCallRequest(request) => Some(request.call_id.clone()),
+                _ => None,
+            })
+            .await,
+        );
+    }
+    assert!(calls.iter().any(|id| id == "before"));
+    let original = calls.into_iter().find(|id| id != "before").unwrap();
+    let output = setup
+        .last_request()
+        .unwrap()
+        .custom_tool_call_output("exec-call");
+    let running = output["output"]
+        .as_str()
+        .or_else(|| output["output"][0]["text"].as_str())
+        .unwrap();
+    let cell_id = running
+        .strip_prefix("Script running with cell ID ")
+        .and_then(|rest| rest.lines().next())
+        .unwrap();
+    let finish = responses::mount_sse_sequence(
+        &server,
+        vec![
+            sse(vec![
+                ev_function_call(
+                    "terminate",
+                    "wait",
+                    &json!({"cell_id": cell_id, "terminate": true}).to_string(),
+                ),
+                ev_completed("terminate"),
+            ]),
+            sse(vec![
+                ev_function_call("after", "lookup", "{}"),
+                ev_completed("after"),
+            ]),
+            sse(vec![ev_completed("done")]),
+        ],
+    )
+    .await;
+    let barrier_reply = DynamicToolResponse {
+        content_items: Vec::new(),
+        success: true,
+    };
+    thread
+        .submit(Op::DynamicToolResponse {
+            id: "before".into(),
+            response: barrier_reply.clone(),
+        })
+        .await?;
+    // Completion publication can race the next direct call; collect both.
+    let mut completed = None;
+    let mut after = false;
+    wait_for_event(&thread, |event| {
+        match event {
+            EventMsg::ItemCompleted(event) if event.item.id() == original => {
+                completed = Some(event.clone())
+            }
+            EventMsg::DynamicToolCallRequest(request) if request.call_id == "after" => after = true,
+            _ => {}
+        }
+        completed.is_some() && after
+    })
+    .await;
+    let completed = completed.unwrap();
+    let TurnItem::DynamicToolCall(item) = &completed.item else {
+        panic!("dynamic completion")
+    };
+    assert_eq!(
+        (item.status, item.success, item.error.as_deref()),
+        (
+            DynamicToolCallStatus::Failed,
+            Some(false),
+            Some("dynamic tool call was cancelled before receiving a response")
+        )
+    );
+    thread
+        .submit(Op::DynamicToolResponse {
+            id: original.clone(),
+            response: DynamicToolResponse {
+                content_items: vec![DynamicToolCallOutputContentItem::InputText {
+                    text: "LATE_RESULT_MUST_NOT_REACH_MODEL".into(),
+                }],
+                success: true,
+            },
+        })
+        .await?;
+    thread
+        .submit(Op::DynamicToolResponse {
+            id: "after".into(),
+            response: barrier_reply,
+        })
+        .await?;
+    wait_for_event(&thread, |event| matches!(event, EventMsg::TurnComplete(_))).await;
+    let requests = finish.requests();
+    assert_eq!(requests.len(), 3);
+    assert!(
+        requests[1]
+            .function_call_output("terminate")
+            .to_string()
+            .contains("Script terminated")
+    );
+    assert!(!requests[2].input().iter().any(|item| {
+        item.to_string()
+            .contains("LATE_RESULT_MUST_NOT_REACH_MODEL")
+    }));
+    thread.flush_rollout().await?;
+    let rollout = std::fs::read_to_string(thread.rollout_path().unwrap())?;
+    let completions: Vec<_> = rollout
+        .lines()
+        .map(codex_rollout::parse_rollout_line)
+        .collect::<Result<Vec<_>, _>>()?
+        .into_iter()
+        .filter_map(|line| match line.item {
+            RolloutItem::EventMsg(EventMsg::ItemCompleted(event))
+                if event.item.id() == original =>
+            {
+                Some(event)
+            }
+            _ => None,
+        })
+        .collect();
+    assert_eq!(serde_json::to_value(completions)?, json!([completed]));
+    Ok(())
+}

codex-rs/core/tests/suite/mod.rs

@@ -80,6 +80,7 @@ mod cyber_exec_policy;
 mod daybreak_access;
 mod deprecation_notice;
 mod direct_tool_metadata;
+mod dynamic_tool_cancellation;
 mod exec;
 mod exec_policy;
 #[cfg(not(target_os = "windows"))]