Add goal extension idle continuation (#25060)

## Why

The goal extension needs a way to resume an active goal after the thread
becomes idle, but the old core goal runtime should not be refactored as
part of this step. The missing piece is a small core-owned turn-start
primitive: let an extension ask for a normal model turn only when the
thread is idle, and otherwise fail without injecting into whatever is
currently active.

## What Changed

- Adds `CodexThread::try_start_turn_if_idle(...)` as the narrow
extension-facing primitive for synthetic idle work.
- Implements the session side so it refuses to start when:
  - the provided input is empty,
  - the session is in plan mode,
  - a turn is already active, or
  - trigger-turn mailbox work is pending.
- Gives trigger-turn mailbox work priority if it appears while the idle
turn is being prepared.
- Wires `GoalExtension::on_thread_idle` to read the active persisted
goal and submit the continuation prompt through this idle-only
primitive.
- Keeps the legacy core goal continuation implementation in place
instead of folding it into this PR.

## Behavior

This is intentionally best-effort. If `try_start_turn_if_idle` observes
that the thread is not idle, or that higher-priority mailbox work should
run first, it returns the input to the caller. The goal extension drops
that continuation prompt and waits for a future idle opportunity instead
of injecting stale synthetic goal text into an active turn.

## Validation

- `just test -p codex-core
try_start_turn_if_idle_rejects_active_turn_without_injecting`
- `just test -p codex-goal-extension`

codex-rs/core/src/codex_thread.rs

@@ -278,6 +278,14 @@ impl CodexThread {
         self.codex.session.inject_if_running(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(
         &self,
         app_server_client_name: Option<String>,

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

@@ -1,7 +1,12 @@
 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;
+use std::sync::Arc;
 
 impl Session {
     /// Returns the input if there is no active turn to inject into.
@@ -28,6 +33,78 @@ impl Session {
         }
     }
 
+    /// 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 Ok(());
+        }
+        if self.collaboration_mode().await.mode == ModeKind::Plan {
+            return Err(input);
+        }
+        if self.input_queue.has_trigger_turn_mailbox_items().await {
+            return Err(input);
+        }
+
+        let turn_state = {
+            let mut active_turn = self.active_turn.lock().await;
+            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)
+        };
+
+        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
+            .new_default_turn_with_sub_id(uuid::Uuid::new_v4().to_string())
+            .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| {
+                active_turn.task.is_none() && Arc::ptr_eq(&active_turn.turn_state, &turn_state)
+            })
+        };
+        if !still_reserved {
+            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.
     pub(crate) async fn inject_no_new_turn(
         &self,

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

@@ -8420,6 +8420,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;