Add conditional turn interruption that preserves pending input (#47340)

## Why

Callers need to interrupt a specific active turn without discarding queued input. Tool callbacks must be able to await the interruption decision without blocking task cancellation.

## What changed

- Add `CodexThread::interrupt_if_no_pending_input` and `Op::InterruptIfNoPendingInput`.
- Check the turn identity and pending user input or mailbox items before taking the active task. Leave the turn running for stale or abandoned requests.
- Acknowledge the decision before joining cancellation, and unblock pending submissions when the active task is cancelled.

## Testing

Add regression coverage for successful interruption, stale turns, preservation of queued user input and agent mail, abandoned requests, and cancellation while the submission queue is full.

GitOrigin-RevId: 42e31c09369fb070ba1b4b3c64ef45334dd872d8

Author: pmccrary-oai

codex-rs/core/src/session/extension_interruption.rs

@@ -1,13 +1,95 @@
 //! Applies extension-requested interruption only if the selected turn is still active.
-//! The caller selects the warning; the host owns task cancellation and idle notification.
+//! Conditional interruption preserves queued input and acknowledges before joining the task.
 
 use super::session::Session;
+use crate::codex_thread::CodexThread;
 use codex_extension_api::ThreadIdleCause;
+use codex_protocol::error::CodexErr;
+use codex_protocol::error::Result as CodexResult;
 use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::Op;
 use codex_protocol::protocol::TurnAbortReason;
 use std::sync::Arc;
+use tokio::sync::oneshot;
+use tokio::sync::oneshot::error::TryRecvError;
+
+#[cfg(test)]
+#[path = "extension_interruption_tests.rs"]
+mod tests;
+
+impl CodexThread {
+    /// Interrupts the named active turn unless it has queued input.
+    /// Returns the decision before joining cancellation, so tool callbacks can await it.
+    /// Like [`Op::Interrupt`], this does not notify idle contributors; the caller owns wakeup.
+    pub async fn interrupt_if_no_pending_input(&self, turn_id: &str) -> CodexResult<bool> {
+        let cancellation_token = {
+            let active = self.session.active_turn.lock().await;
+            let Some(task) = active.as_ref().and_then(|turn| turn.task.as_ref()) else {
+                return Ok(false);
+            };
+            if task.turn_context.sub_id != turn_id {
+                return Ok(false);
+            }
+            task.cancellation_token.clone()
+        };
+        let (reply, mut result) = oneshot::channel();
+        tokio::select! {
+            biased;
+            result = async {
+                self.submit(Op::InterruptIfNoPendingInput {
+                    turn_id: turn_id.to_owned(),
+                    reply,
+                }).await?;
+                (&mut result).await.map_err(|_| CodexErr::InternalAgentDied)
+            } => result,
+            // An earlier interrupt may already be waiting for this tool callback to finish.
+            // Cancellation can also race with a decision sent by this request.
+            _ = cancellation_token.cancelled() => match result.try_recv() {
+                Ok(decision) => Ok(decision),
+                Err(TryRecvError::Empty | TryRecvError::Closed) => Ok(false),
+            },
+        }
+    }
+}
 
 impl Session {
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "the turn identity and pending input check must be atomic with taking the task"
+    )]
+    pub(crate) async fn interrupt_turn_if_no_pending_input(
+        self: &Arc<Self>,
+        turn_id: &str,
+        reply: oneshot::Sender<bool>,
+    ) {
+        let active_turn = {
+            let mut active = self.active_turn.lock().await;
+            let Some(turn) = active.as_ref().filter(|turn| {
+                turn.task
+                    .as_ref()
+                    .is_some_and(|task| task.turn_context.sub_id == turn_id)
+            }) else {
+                let _ = reply.send(false);
+                return;
+            };
+            if !turn.turn_state.lock().await.pending_input.is_empty()
+                || self.input_queue.has_pending_mailbox_items().await
+                || reply.is_closed()
+            {
+                let _ = reply.send(false);
+                return;
+            }
+            self.mark_interrupted();
+            active.take()
+        };
+        // The caller may be inside the task that cancellation must join.
+        let _ = reply.send(active_turn.is_some());
+        if let Some(active_turn) = active_turn {
+            self.finish_turn_abort(active_turn, TurnAbortReason::Interrupted)
+                .await;
+        }
+    }
+
     pub(crate) async fn interrupt_turn_with_warning(
         self: &Arc<Self>,
         turn_id: &str,

codex-rs/core/src/session/extension_interruption_tests.rs

@@ -0,0 +1,93 @@
+//! Regression coverage for cancellation while submitting a conditional interrupt.
+
+use super::*;
+use crate::session::SessionIo;
+use crate::session::completed_session_loop_termination;
+use crate::session::tests::HeldStepTask;
+use crate::session::tests::make_session_and_context_with_rx;
+use crate::state::TaskKind;
+use crate::thread_startup_metadata::ThreadStartupMetadata;
+use codex_protocol::config_types::ApprovalsReviewer;
+use codex_protocol::protocol::AgentStatus;
+use codex_protocol::protocol::AskForApproval;
+use codex_protocol::protocol::SessionConfiguredEvent;
+use codex_protocol::protocol::SessionSource;
+use pretty_assertions::assert_eq;
+use std::time::Duration;
+use tokio::sync::Notify;
+use tokio::sync::watch;
+
+#[tokio::test]
+async fn interrupt_if_no_pending_input_handles_cancelled_submission() {
+    let (session, turn_context, rx_event) = make_session_and_context_with_rx().await;
+    session
+        .spawn_task(
+            Arc::clone(&turn_context),
+            Vec::new(),
+            HeldStepTask {
+                kind: TaskKind::Regular,
+                finish: Arc::new(Notify::new()),
+            },
+        )
+        .await;
+    let cancellation_token = session
+        .active_turn
+        .lock()
+        .await
+        .as_ref()
+        .expect("active turn")
+        .task
+        .as_ref()
+        .expect("active task")
+        .cancellation_token
+        .clone();
+    let (tx_sub, rx_sub) = async_channel::bounded(/*cap*/ 1);
+    let thread = CodexThread::new(
+        Arc::clone(&session),
+        SessionIo {
+            tx_sub,
+            rx_event,
+            agent_status: watch::channel(AgentStatus::PendingInit).1,
+            session_loop_termination: completed_session_loop_termination(),
+        },
+        ThreadStartupMetadata::from(&SessionConfiguredEvent {
+            session_id: session.session_id(),
+            thread_id: session.thread_id(),
+            forked_from_id: None,
+            parent_thread_id: None,
+            thread_source: None,
+            thread_name: None,
+            model: "test".to_string(),
+            model_provider_id: "test".to_string(),
+            service_tier: None,
+            approval_policy: AskForApproval::Never,
+            approvals_reviewer: ApprovalsReviewer::User,
+            permission_profile: turn_context.permission_profile(),
+            active_permission_profile: None,
+            cwd: turn_context.config.cwd.clone(),
+            reasoning_effort: None,
+            initial_messages: None,
+            network_proxy: None,
+            rollout_path: None,
+        }),
+        /*rollout_path*/ None,
+        SessionSource::Cli,
+    );
+    thread.submit(Op::Interrupt).await.expect("fill the queue");
+
+    let interrupt = thread.interrupt_if_no_pending_input(&turn_context.sub_id);
+    tokio::pin!(interrupt);
+    assert!(futures::poll!(interrupt.as_mut()).is_pending());
+    // The pending submission still owns its reply sender when cancellation wins.
+    cancellation_token.cancel();
+    assert_eq!(
+        tokio::time::timeout(Duration::from_secs(/*secs*/ 1), interrupt)
+            .await
+            .expect("cancellation should unblock the submission")
+            .expect("cancelling a submission does not mean the agent died"),
+        false,
+    );
+    assert!(!rx_sub.is_closed());
+    assert_eq!(rx_sub.len(), 1);
+    session.abort_all_tasks(TurnAbortReason::Interrupted).await;
+}

codex-rs/core/src/session/handlers.rs

@@ -428,6 +428,11 @@ pub(super) async fn submission_loop(
                     interrupt(&sess).await;
                     false
                 }
+                Op::InterruptIfNoPendingInput { turn_id, reply } => {
+                    sess.interrupt_turn_if_no_pending_input(&turn_id, reply)
+                        .await;
+                    false
+                }
                 Op::CleanBackgroundTerminals => {
                     clean_background_terminals(&sess).await;
                     false

codex-rs/core/src/session/input_queue.rs

@@ -333,7 +333,7 @@ impl InputQueue {
                 Some(active_turn) => {
                     let turn_state = active_turn.turn_state.lock().await;
                     (
-                        !turn_state.pending_input.items.is_empty(),
+                        !turn_state.pending_input.is_empty(),
                         turn_state.accepts_mailbox_delivery_for_current_turn(),
                     )
                 }
@@ -351,6 +351,10 @@ impl InputQueue {
 }
 
 impl TurnInputQueue {
+    pub(crate) fn is_empty(&self) -> bool {
+        self.items.is_empty()
+    }
+
     fn has_pending_input(&self) -> bool {
         self.items.iter().any(|input| {
             matches!(

codex-rs/core/src/tasks/mod.rs

@@ -586,10 +586,19 @@ impl Session {
                 None
             }
         };
-        let Some(mut active_turn) = active_turn else {
+        let Some(active_turn) = active_turn else {
             return false;
         };
 
+        self.finish_turn_abort(active_turn, reason).await;
+        true
+    }
+
+    pub(crate) async fn finish_turn_abort(
+        self: &Arc<Self>,
+        mut active_turn: ActiveTurn,
+        reason: TurnAbortReason,
+    ) {
         let task = active_turn.task.take();
         let turn_context = task.as_ref().map(|task| Arc::clone(&task.turn_context));
         if let Some(task) = task {
@@ -607,8 +616,6 @@ impl Session {
         if reason == TurnAbortReason::Interrupted {
             self.maybe_start_turn_for_pending_work().await;
         }
-
-        true
     }
 
     pub async fn on_task_finished(

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

@@ -1162,6 +1162,145 @@ async fn user_input_does_not_preempt_after_reasoning_item() {
     server.shutdown().await;
 }
 
+#[derive(Clone, Copy)]
+enum ConditionalInterruptCase {
+    CurrentTurn,
+    StaleTurn,
+    PendingUserInput,
+    PendingMailbox,
+    AbandonedRequest,
+}
+
+#[test_case(ConditionalInterruptCase::CurrentTurn; "current_turn_without_pending_input")]
+#[test_case(ConditionalInterruptCase::StaleTurn; "stale_turn")]
+#[test_case(ConditionalInterruptCase::PendingUserInput; "pending_user_input")]
+#[test_case(ConditionalInterruptCase::PendingMailbox; "pending_mailbox")]
+#[test_case(ConditionalInterruptCase::AbandonedRequest; "abandoned_request")]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn interrupt_if_no_pending_input_checks_turn_and_queue(
+    case: ConditionalInterruptCase,
+) -> anyhow::Result<()> {
+    const INITIAL_PROMPT: &str = "first prompt";
+    const PENDING_PROMPT: &str = "preserve this pending input";
+    let (release_response, response_gate) = oneshot::channel();
+    let first_chunks = vec![
+        chunk(ev_response_created("resp-1")),
+        chunk(ev_reasoning_item_added("reason-1", &["thinking"])),
+        gated_chunk(
+            response_gate,
+            vec![
+                ev_reasoning_item("reason-1", &["thinking"], &[]),
+                ev_completed("resp-1"),
+            ],
+        ),
+    ];
+    let (server, _completions) =
+        start_streaming_sse_server(vec![first_chunks, response_completed_chunks("resp-2")]).await;
+    let config_server = responses::start_mock_server().await;
+    let base_url = format!("{}/v1", server.uri());
+    let test = test_codex()
+        .with_model("gpt-5.4")
+        .with_config(move |config| {
+            config.model_provider.base_url = Some(base_url);
+            let _ = config.features.disable(Feature::EnableRequestCompression);
+        })
+        .build_with_auto_env(&config_server)
+        .await?;
+    let codex = &test.codex;
+    let TurnInputSubmission::Started { turn_id } = codex
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
+            text: INITIAL_PROMPT.to_string(),
+            text_elements: Vec::new(),
+        }]))
+        .await?
+    else {
+        panic!("initial input should start a turn");
+    };
+    wait_for_reasoning_item_started(codex).await;
+    if matches!(case, ConditionalInterruptCase::PendingUserInput) {
+        steer_user_input(codex, PENDING_PROMPT).await;
+    }
+    if matches!(case, ConditionalInterruptCase::PendingMailbox) {
+        submit_queue_only_agent_mail(codex, PENDING_PROMPT).await;
+    }
+    let expected_turn_id = match case {
+        ConditionalInterruptCase::StaleTurn | ConditionalInterruptCase::AbandonedRequest => {
+            format!("stale-{turn_id}")
+        }
+        ConditionalInterruptCase::CurrentTurn
+        | ConditionalInterruptCase::PendingUserInput
+        | ConditionalInterruptCase::PendingMailbox => turn_id.clone(),
+    };
+    if matches!(case, ConditionalInterruptCase::AbandonedRequest) {
+        let (reply, result) = oneshot::channel();
+        drop(result);
+        codex
+            .submit(Op::InterruptIfNoPendingInput { turn_id, reply })
+            .await?;
+        // Wait for the abandoned request to be handled before releasing the response.
+        let (reply, result) = oneshot::channel();
+        codex
+            .submit(Op::InterruptIfNoPendingInput {
+                turn_id: expected_turn_id.clone(),
+                reply,
+            })
+            .await?;
+        assert!(!tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 10), result).await??);
+    }
+    let should_abort = matches!(case, ConditionalInterruptCase::CurrentTurn);
+    assert_eq!(
+        tokio::time::timeout(
+            std::time::Duration::from_secs(/*secs*/ 10),
+            codex.interrupt_if_no_pending_input(&expected_turn_id),
+        )
+        .await??,
+        should_abort,
+    );
+
+    if should_abort {
+        wait_for_event(codex, |event| {
+            assert!(!matches!(event, EventMsg::TurnComplete(_)));
+            matches!(event, EventMsg::TurnAborted(_))
+        })
+        .await;
+        let _ = release_response.send(());
+    } else {
+        release_response.send(()).expect("release model response");
+        wait_for_event(codex, |event| {
+            assert!(!matches!(event, EventMsg::TurnAborted(_)));
+            matches!(event, EventMsg::TurnComplete(_))
+        })
+        .await;
+    }
+    let requests = server.requests().await;
+    if matches!(case, ConditionalInterruptCase::PendingUserInput) {
+        assert_eq!(requests.len(), 2);
+        let second: Value = from_slice(&requests[1])?;
+        let prompts = message_input_texts(&second, "user")
+            .into_iter()
+            .filter(|text| text == INITIAL_PROMPT || text == PENDING_PROMPT)
+            .collect::<Vec<_>>();
+        assert_eq!(prompts, vec![INITIAL_PROMPT, PENDING_PROMPT]);
+    } else if matches!(case, ConditionalInterruptCase::PendingMailbox) {
+        assert_eq!(requests.len(), 2);
+        let second: Value = from_slice(&requests[1])?;
+        let mail = second["input"]
+            .as_array()
+            .expect("model input")
+            .iter()
+            .find(|item| item["type"] == "agent_message")
+            .expect("pending mailbox input");
+        assert_eq!(
+            mail["content"],
+            json!([{ "type": "input_text", "text": PENDING_PROMPT }])
+        );
+    } else {
+        assert_eq!(requests.len(), 1);
+    }
+    server.shutdown().await;
+    Ok(())
+}
+
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum CompactionFailurePoint {
     PreTurn,