Make goal idle continuation start idle turns only

codex-rs/core/src/codex_thread.rs

@@ -276,10 +276,12 @@ impl CodexThread {
         self.codex.session.inject_if_running(items).await
     }
 
-    /// Injects model-visible items into active work, or starts a regular turn
-    /// with them if the thread is idle.
-    pub async fn inject_or_start_turn(&self, items: Vec<ResponseItem>) {
-        self.codex.session.inject_or_start_turn(items).await
+    /// Starts a regular turn with model-visible items only if the thread is idle.
+    pub async fn try_start_turn_if_idle(
+        &self,
+        items: Vec<ResponseItem>,
+    ) -> Result<(), Vec<ResponseItem>> {
+        self.codex.session.try_start_turn_if_idle(items).await
     }
 
     pub async fn set_app_server_client_info(

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

@@ -2,6 +2,7 @@ use super::input_queue::TurnInput;
 use super::session::Session;
 use super::turn_context::TurnContext;
 use crate::state::ActiveTurn;
+use crate::state::TurnState;
 use crate::tasks::RegularTask;
 use codex_protocol::config_types::ModeKind;
 use codex_protocol::models::ResponseItem;
@@ -32,38 +33,34 @@ impl Session {
         }
     }
 
-    /// Injects items into active work, or starts a regular turn with them.
-    pub(crate) async fn inject_or_start_turn(self: &Arc<Self>, input: Vec<ResponseItem>) {
+    /// Starts a regular turn with the provided items only if the session is idle.
+    pub(crate) async fn try_start_turn_if_idle(
+        self: &Arc<Self>,
+        input: Vec<ResponseItem>,
+    ) -> Result<(), Vec<ResponseItem>> {
         if input.is_empty() {
-            return;
+            return Ok(());
         }
-        let input = match self.inject_if_running(input).await {
-            Ok(()) => return,
-            Err(input) => input,
-        };
         if self.collaboration_mode().await.mode == ModeKind::Plan {
-            let _ = self.inject_if_running(input).await;
-            return;
+            return Err(input);
+        }
+        if self.input_queue.has_trigger_turn_mailbox_items().await {
+            return Err(input);
         }
 
-        let (turn_state, should_start_turn) = {
+        let turn_state = {
             let mut active_turn = self.active_turn.lock().await;
-            if let Some(active_turn) = active_turn.as_ref() {
-                (Arc::clone(&active_turn.turn_state), false)
-            } else {
-                let active_turn = active_turn.get_or_insert_with(ActiveTurn::default);
-                (Arc::clone(&active_turn.turn_state), true)
+            if active_turn.is_some() {
+                return Err(input);
             }
+            let active_turn = active_turn.get_or_insert_with(ActiveTurn::default);
+            Arc::clone(&active_turn.turn_state)
         };
 
-        self.input_queue
-            .extend_pending_input_for_turn_state(
-                turn_state.as_ref(),
-                input.into_iter().map(TurnInput::ResponseItem).collect(),
-            )
-            .await;
-        if !should_start_turn {
-            return;
+        if self.input_queue.has_trigger_turn_mailbox_items().await {
+            self.clear_reserved_idle_turn(&turn_state).await;
+            self.maybe_start_turn_for_pending_work().await;
+            return Err(input);
         }
 
         let turn_context = self
@@ -71,6 +68,11 @@ impl Session {
             .await;
         self.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref())
             .await;
+        if self.input_queue.has_trigger_turn_mailbox_items().await {
+            self.clear_reserved_idle_turn(&turn_state).await;
+            self.maybe_start_turn_for_pending_work().await;
+            return Err(input);
+        }
         let still_reserved = {
             let active_turn = self.active_turn.lock().await;
             active_turn.as_ref().is_some_and(|active_turn| {
@@ -78,18 +80,29 @@ impl Session {
             })
         };
         if !still_reserved {
-            let mut active_turn_guard = self.active_turn.lock().await;
-            if let Some(active_turn) = active_turn_guard.as_ref()
-                && active_turn.task.is_none()
-                && Arc::ptr_eq(&active_turn.turn_state, &turn_state)
-            {
-                *active_turn_guard = None;
-            }
-            return;
+            self.clear_reserved_idle_turn(&turn_state).await;
+            return Err(input);
         }
 
+        self.input_queue
+            .extend_pending_input_for_turn_state(
+                turn_state.as_ref(),
+                input.into_iter().map(TurnInput::ResponseItem).collect(),
+            )
+            .await;
         self.start_task(turn_context, Vec::new(), RegularTask::new())
             .await;
+        Ok(())
+    }
+
+    async fn clear_reserved_idle_turn(&self, turn_state: &Arc<tokio::sync::Mutex<TurnState>>) {
+        let mut active_turn_guard = self.active_turn.lock().await;
+        if let Some(active_turn) = active_turn_guard.as_ref()
+            && active_turn.task.is_none()
+            && Arc::ptr_eq(&active_turn.turn_state, turn_state)
+        {
+            *active_turn_guard = None;
+        }
     }
 
     /// Injects items into active work, or records them without starting a turn.

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

@@ -8379,6 +8379,34 @@ async fn thread_idle_lifecycle_waits_for_trigger_turn_mailbox_work() {
     assert_eq!(0, calls.load(std::sync::atomic::Ordering::SeqCst));
 }
 
+#[tokio::test]
+async fn try_start_turn_if_idle_rejects_active_turn_without_injecting() {
+    let (sess, tc, _rx) = make_session_and_context_with_rx().await;
+    sess.spawn_task(
+        Arc::clone(&tc),
+        Vec::new(),
+        NeverEndingTask {
+            kind: TaskKind::Regular,
+            listen_to_cancellation_token: true,
+        },
+    )
+    .await;
+
+    let item = user_message("synthetic idle input");
+    let err = sess
+        .try_start_turn_if_idle(vec![item.clone()])
+        .await
+        .expect_err("active turn should reject idle-only input");
+
+    assert_eq!(vec![item], err);
+    assert_eq!(
+        Vec::<TurnInput>::new(),
+        sess.input_queue.get_pending_input(&sess.active_turn).await
+    );
+
+    sess.abort_all_tasks(TurnAbortReason::Interrupted).await;
+}
+
 #[tokio::test]
 async fn steer_input_requires_active_turn() {
     let (sess, _tc, _rx) = make_session_and_context_with_rx().await;

codex-rs/ext/goal/src/runtime.rs

@@ -308,7 +308,9 @@ impl GoalRuntimeHandle {
             return Ok(());
         };
 
-        thread.inject_or_start_turn(vec![item]).await;
+        if thread.try_start_turn_if_idle(vec![item]).await.is_err() {
+            tracing::debug!("skipping goal continuation because the thread is no longer idle");
+        }
 
         let current_turn_is_goal_active = self
             .inner