diff --git a/codex-rs/core/src/state/turn.rs b/codex-rs/core/src/state/turn.rs index 28ae8fec65..41b0d4bec2 100644 --- a/codex-rs/core/src/state/turn.rs +++ b/codex-rs/core/src/state/turn.rs @@ -149,11 +149,3 @@ impl TurnState { !self.pending_input.is_empty() } } - -impl ActiveTurn { - /// Clear any pending approvals and input buffered for the current turn. - pub(crate) async fn clear_pending(&self) { - let mut ts = self.turn_state.lock().await; - ts.clear_pending(); - } -} diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index a8feeb5133..7921280f54 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -31,12 +31,14 @@ use crate::protocol::TurnCompleteEvent; use crate::state::ActiveTurn; use crate::state::RunningTask; use crate::state::TaskKind; +use crate::state::TurnState; use codex_protocol::items::TurnItem; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::RolloutItem; use codex_protocol::user_input::UserInput; +use tokio::sync::Mutex; use crate::features::Feature; pub(crate) use compact::CompactTask; @@ -192,7 +194,16 @@ impl Session { } pub async fn abort_all_tasks(self: &Arc, reason: TurnAbortReason) { - for task in self.take_all_running_tasks().await { + let (tasks, turn_state) = self.take_all_running_tasks().await; + for task in &tasks { + task.cancellation_token.cancel(); + } + if let Some(turn_state) = turn_state { + // Drop pending approvals only after all running tasks observe cancellation, so + // interrupted approval waits resolve as aborts instead of synthetic denials. + turn_state.lock().await.clear_pending(); + } + for task in tasks { self.handle_task_abort(task, reason.clone()).await; } if reason == TurnAbortReason::Interrupted { @@ -328,15 +339,14 @@ impl Session { *active = Some(turn); } - async fn take_all_running_tasks(&self) -> Vec { + async fn take_all_running_tasks(&self) -> (Vec, Option>>) { let mut active = self.active_turn.lock().await; match active.take() { Some(mut at) => { - at.clear_pending().await; - - at.drain_tasks() + let turn_state = Arc::clone(&at.turn_state); + (at.drain_tasks(), Some(turn_state)) } - None => Vec::new(), + None => (Vec::new(), None), } } @@ -349,10 +359,6 @@ impl Session { async fn handle_task_abort(self: &Arc, task: RunningTask, reason: TurnAbortReason) { let sub_id = task.turn_context.sub_id.clone(); - if task.cancellation_token.is_cancelled() { - return; - } - trace!(task_kind = ?task.kind, sub_id, "aborting running task"); task.cancellation_token.cancel(); task.turn_context