From c4f509101b91c07991a23d27d808b3153e8f9fd8 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Fri, 29 May 2026 14:09:52 +0100 Subject: [PATCH] Make goal idle continuation start idle turns only --- codex-rs/core/src/codex_thread.rs | 10 ++-- codex-rs/core/src/session/inject.rs | 75 +++++++++++++++++------------ codex-rs/core/src/session/tests.rs | 28 +++++++++++ codex-rs/ext/goal/src/runtime.rs | 4 +- 4 files changed, 81 insertions(+), 36 deletions(-) diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 75bb15828d..5fb321174a 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/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) { - 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, + ) -> Result<(), Vec> { + self.codex.session.try_start_turn_if_idle(items).await } pub async fn set_app_server_client_info( diff --git a/codex-rs/core/src/session/inject.rs b/codex-rs/core/src/session/inject.rs index 3b495742b1..b94cfe4de7 100644 --- a/codex-rs/core/src/session/inject.rs +++ b/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, input: Vec) { + /// 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, + input: Vec, + ) -> Result<(), Vec> { 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>) { + 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. diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index aa2586cae4..e203f57e11 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/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::::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; diff --git a/codex-rs/ext/goal/src/runtime.rs b/codex-rs/ext/goal/src/runtime.rs index dd18d5d6cc..fcb4124952 100644 --- a/codex-rs/ext/goal/src/runtime.rs +++ b/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