Guard against racing turn starts

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;

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),