diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 22bc33ca43..15d9047222 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -8238,6 +8238,36 @@ async fn queued_response_items_for_next_turn_move_into_next_active_turn() { ); } +#[tokio::test] +async fn start_task_does_not_overwrite_racing_active_task() { + let (sess, first_context, _rx) = make_session_and_context_with_rx().await; + let task = NeverEndingTask { + kind: TaskKind::Regular, + listen_to_cancellation_token: true, + }; + sess.start_task(Arc::clone(&first_context), Vec::new(), task) + .await; + + let second_context = sess + .new_default_turn_with_sub_id("racing-turn".to_string()) + .await; + sess.start_task(second_context, Vec::new(), task).await; + + let active_turn_id = { + let active_turn = sess.active_turn.lock().await; + active_turn + .as_ref() + .and_then(|active_turn| active_turn.task.as_ref()) + .map(|task| task.turn_context.sub_id.clone()) + }; + assert_eq!( + active_turn_id.as_deref(), + Some(first_context.sub_id.as_str()) + ); + + sess.abort_all_tasks(TurnAbortReason::Interrupted).await; +} + #[tokio::test] async fn idle_interrupt_does_not_wake_queued_next_turn_items() { let (sess, _tc, _rx) = make_session_and_context_with_rx().await; diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 72c6639da1..6af3d3013a 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -339,17 +339,19 @@ impl Session { .await .clear_turn(&turn_context.sub_id); + let turn_state = { + let mut active = self.active_turn.lock().await; + let turn = active.get_or_insert_with(ActiveTurn::default); + if turn.task.is_some() { + return; + } + Arc::clone(&turn.turn_state) + }; let queued_response_items = self .input_queue .take_queued_response_items_for_next_turn() .await; let mailbox_items = self.input_queue.get_pending_input(&self.active_turn).await; - let turn_state = { - let mut active = self.active_turn.lock().await; - let turn = active.get_or_insert_with(ActiveTurn::default); - debug_assert!(turn.task.is_none()); - Arc::clone(&turn.turn_state) - }; turn_state.lock().await.token_usage_at_turn_start = token_usage_at_turn_start.clone(); let mut pending_items = queued_response_items .into_iter() @@ -365,7 +367,9 @@ impl Session { let turn_extension_data = Arc::clone(&turn_context.extension_data); let mut active = self.active_turn.lock().await; let turn = active.get_or_insert_with(ActiveTurn::default); - debug_assert!(turn.task.is_none()); + if turn.task.is_some() { + return; + } let done_clone = Arc::clone(&done); let session_ctx = Arc::new(SessionTaskContext::new( Arc::clone(self),