Add interrupted turn recovery (#38303)

## What changed

- Add `RecoverTurnRequest` and `CodexThread::recover_turn_if_idle` to resume an interrupted regular turn with its existing turn ID and updated thread settings.
- Treat recovery separately from automatic idle work so it can resume in Plan mode without injecting an empty user message.
- Reject recovery while another turn is active without applying settings or queuing input.

## Testing

- Cover successful Plan-mode recovery, turn ID preservation, and the absence of an empty user message.
- Cover active-turn rejection and verify that it leaves settings and pending input unchanged.

GitOrigin-RevId: 2376fdc78b5d48b571633981960741ba269ab1ef

codex-rs/core-api/src/lib.rs

@@ -33,6 +33,7 @@ pub use codex_core::LoadedAgentsMd;
 pub use codex_core::McpManager;
 pub use codex_core::NewThread;
 pub use codex_core::NotSubmittedReason;
+pub use codex_core::RecoverTurnRequest;
 pub use codex_core::StartIfIdleSubmission;
 pub use codex_core::StartThreadOptions;
 pub use codex_core::StateDbHandle;

codex-rs/core/src/codex_thread.rs

@@ -41,6 +41,7 @@ use codex_protocol::protocol::TokenUsageInfo;
 use codex_protocol::protocol::TurnEnvironmentSelection;
 use codex_protocol::protocol::TurnEnvironmentSelections;
 use codex_protocol::protocol::W3cTraceContext;
+use codex_protocol::turn_input::RecoverTurnRequest;
 use codex_protocol::turn_input::StartIfIdleSubmission;
 use codex_protocol::turn_input::SteerSubmission;
 use codex_protocol::turn_input::TurnInputMode;
@@ -294,6 +295,41 @@ impl CodexThread {
         }
     }
 
+    /// Resumes an interrupted regular turn only when the thread is idle.
+    ///
+    /// Recovery starts no new user input and preserves the turn ID that was
+    /// already recorded for the interrupted turn.
+    pub async fn recover_turn_if_idle(
+        &self,
+        request: RecoverTurnRequest,
+    ) -> CodexResult<StartIfIdleSubmission> {
+        self.session
+            .services
+            .agent_control
+            .ensure_execution_capacity_for_turn_start(self)
+            .await?;
+        let RecoverTurnRequest {
+            turn_id,
+            thread_settings,
+            trace,
+        } = request;
+        match self
+            .io
+            .submit_recover_turn(thread_settings, trace, turn_id)
+            .await?
+        {
+            TurnInputSubmission::Started { turn_id } => {
+                Ok(StartIfIdleSubmission::Started { turn_id })
+            }
+            TurnInputSubmission::NotSubmitted { reason } => {
+                Ok(StartIfIdleSubmission::NotSubmitted { reason })
+            }
+            TurnInputSubmission::Steered { .. } => {
+                unreachable!("recovered turn submission cannot steer")
+            }
+        }
+    }
+
     /// Steers only if `expected_turn_id` is still the active regular turn.
     pub async fn steer_turn(
         &self,

codex-rs/core/src/lib.rs

@@ -16,6 +16,7 @@ mod responses_metadata;
 mod responses_retry;
 pub(crate) mod session;
 pub use codex_protocol::turn_input::NotSubmittedReason;
+pub use codex_protocol::turn_input::RecoverTurnRequest;
 pub use codex_protocol::turn_input::StartIfIdleSubmission;
 pub use codex_protocol::turn_input::SteerSubmission;
 pub use codex_protocol::turn_input::TurnInput;

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

@@ -565,6 +565,15 @@ pub(super) async fn submission_loop(
                     let _ = reply.send(result);
                     false
                 }
+                Op::RecoverTurn {
+                    thread_settings,
+                    reply,
+                } => {
+                    let result =
+                        turn_input::handle_recovery(&sess, thread_settings, sub.id.clone()).await;
+                    let _ = reply.send(result);
+                    false
+                }
                 Op::ThreadSettings { thread_settings } => {
                     thread_settings::update(&sess, sub.id.clone(), thread_settings).await;
                     false

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

@@ -120,6 +120,7 @@ use codex_protocol::protocol::RawResponseItemEvent;
 use codex_protocol::protocol::SessionSource;
 use codex_protocol::protocol::SubAgentSource;
 use codex_protocol::protocol::ThreadHistoryMode;
+use codex_protocol::protocol::ThreadSettingsOverrides;
 use codex_protocol::protocol::ThreadSource;
 use codex_protocol::protocol::TurnAbortReason;
 use codex_protocol::protocol::TurnContextItem;
@@ -856,6 +857,27 @@ impl SessionIo {
         reply_rx.await.unwrap_or(Err(CodexErr::InternalAgentDied))
     }
 
+    pub(crate) async fn submit_recover_turn(
+        &self,
+        thread_settings: ThreadSettingsOverrides,
+        trace: Option<W3cTraceContext>,
+        turn_id: String,
+    ) -> CodexResult<TurnInputSubmission> {
+        let (reply_tx, reply_rx) = oneshot::channel();
+        self.submit_with_id(Submission {
+            id: turn_id,
+            op: Op::RecoverTurn {
+                thread_settings,
+                reply: reply_tx,
+            },
+            trace,
+            parent_turn_id: None,
+            root_turn_id: None,
+        })
+        .await?;
+        reply_rx.await.unwrap_or(Err(CodexErr::InternalAgentDied))
+    }
+
     pub(crate) async fn shutdown_and_wait(&self) -> CodexResult<()> {
         let session_loop_termination = self.session_loop_termination.clone();
         match self.submit(Op::Shutdown).await {

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

@@ -146,13 +146,24 @@ pub(super) async fn handle(
 ) -> CodexResult<TurnInputSubmission> {
     match mode {
         TurnInputMode::StartOrSteer => start_or_steer(session, request, submission_id).await,
-        TurnInputMode::StartIfIdle => start_if_idle(session, request, submission_id).await,
+        TurnInputMode::StartIfIdle => {
+            start_if_idle(session, request, submission_id, /*is_recovery*/ false).await
+        }
         TurnInputMode::Steer { expected_turn_id } => {
             steer(session, request, expected_turn_id, submission_id).await
         }
     }
 }
 
+pub(super) async fn handle_recovery(
+    session: &Arc<Session>,
+    thread_settings: ThreadSettingsOverrides,
+    submission_id: String,
+) -> CodexResult<TurnInputSubmission> {
+    let request = TurnInputRequest::user_input(Vec::new()).with_thread_settings(thread_settings);
+    start_if_idle(session, request, submission_id, /*is_recovery*/ true).await
+}
+
 async fn start_or_steer(
     session: &Arc<Session>,
     request: TurnInputRequest,
@@ -242,6 +253,7 @@ async fn start_if_idle(
     session: &Arc<Session>,
     request: TurnInputRequest,
     submission_id: String,
+    is_recovery: bool,
 ) -> CodexResult<TurnInputSubmission> {
     let TurnInputRequest {
         input,
@@ -252,13 +264,16 @@ async fn start_if_idle(
         ..
     } = request;
     let has_user_input = has_nonempty_user_input(&input);
+    let is_automatic_idle_work = !has_user_input && !is_recovery;
     let can_start_root_turn = start.parent_turn_id.is_none() && start.root_turn_id.is_none();
     if session.input_queue.has_trigger_turn_mailbox_items().await {
         return Ok(TurnInputSubmission::NotSubmitted {
             reason: NotSubmittedReason::PendingTriggerTurn,
         });
     }
-    if !has_user_input && session.collaboration_mode().await.mode == ModeKind::Plan {
+    // Empty non-recovery starts are automatic wakeups, not explicit user requests.
+    // Do not let them start a Plan turn.
+    if is_automatic_idle_work && session.collaboration_mode().await.mode == ModeKind::Plan {
         return Ok(TurnInputSubmission::NotSubmitted {
             reason: NotSubmittedReason::PlanMode,
         });
@@ -292,7 +307,7 @@ async fn start_if_idle(
     };
     // Automatic work must not use persistent settings to start a turn
     // whose effective collaboration mode is Plan.
-    if !has_user_input && settings.would_enter_plan_mode() {
+    if is_automatic_idle_work && settings.would_enter_plan_mode() {
         session.clear_reserved_idle_turn(&turn_state).await;
         return Ok(TurnInputSubmission::NotSubmitted {
             reason: NotSubmittedReason::PlanMode,
@@ -332,7 +347,9 @@ async fn start_if_idle(
             turn_context.session_telemetry.user_prompt(content);
         }
         task_input.push(pending_turn_input(input));
-    } else {
+    } else if is_automatic_idle_work {
+        // Recovery resumes an existing turn, so it must not queue a new empty
+        // user message for that turn.
         session
             .input_queue
             .extend_pending_input_for_turn_state(

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

@@ -9,6 +9,7 @@ use codex_protocol::config_types::ModeKind;
 use codex_protocol::config_types::Settings;
 use codex_protocol::models::ContentItem;
 use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::AskForApproval;
 use codex_protocol::protocol::InterAgentCommunication;
 use codex_protocol::protocol::ThreadSettingsOverrides;
 use codex_protocol::protocol::TurnAbortReason;
@@ -175,6 +176,63 @@ async fn start_only_rejects_active_turn_without_injecting() {
     session.abort_all_tasks(TurnAbortReason::Interrupted).await;
 }
 
+#[tokio::test]
+async fn recovery_rejects_active_turn_without_injecting_or_applying_settings() {
+    let (session, turn_context, _rx) = make_session_and_context_with_rx().await;
+    let original_approval_policy = session
+        .get_config()
+        .await
+        .permissions
+        .approval_policy
+        .value();
+    session
+        .spawn_task(
+            Arc::clone(&turn_context),
+            Vec::new(),
+            NeverEndingTask {
+                kind: TaskKind::Regular,
+                listen_to_cancellation_token: true,
+            },
+        )
+        .await;
+
+    let submission = handle_recovery(
+        &session,
+        ThreadSettingsOverrides {
+            approval_policy: Some(AskForApproval::Never),
+            ..Default::default()
+        },
+        "recovered-turn".to_string(),
+    )
+    .await
+    .expect("recovery should return a typed rejection");
+
+    assert_eq!(
+        submission,
+        TurnInputSubmission::NotSubmitted {
+            reason: NotSubmittedReason::NotIdle,
+        }
+    );
+    assert_eq!(
+        session
+            .get_config()
+            .await
+            .permissions
+            .approval_policy
+            .value(),
+        original_approval_policy
+    );
+    assert_eq!(
+        session
+            .input_queue
+            .get_pending_input(&session.active_turn)
+            .await,
+        (Vec::<TurnInput>::new(), None, None)
+    );
+
+    session.abort_all_tasks(TurnAbortReason::Interrupted).await;
+}
+
 #[tokio::test]
 async fn start_only_rejects_plan_mode_without_injecting() {
     let (session, _turn_context, _rx) = make_session_and_context_with_rx().await;

codex-rs/core/tests/suite/turn_input_submission.rs

@@ -1,4 +1,5 @@
 use codex_core::NotSubmittedReason;
+use codex_core::RecoverTurnRequest;
 use codex_core::StartIfIdleSubmission;
 use codex_core::SteerSubmission;
 use codex_core::TurnInput;
@@ -84,6 +85,67 @@ async fn start_turn_if_idle_rejects_non_user_input_that_requests_plan_mode() {
     );
 }
 
+#[tokio::test]
+async fn recover_turn_if_idle_preserves_id_and_resumes_plan_mode() {
+    let server = responses::start_mock_server().await;
+    let response_mock = responses::mount_sse_once(
+        &server,
+        responses::sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]),
+    )
+    .await;
+    let test = test_codex()
+        .build_with_auto_env(&server)
+        .await
+        .expect("build recovered turn session");
+    let turn_id = "durable-recovered-turn";
+
+    let submission = test
+        .codex
+        .recover_turn_if_idle(RecoverTurnRequest {
+            turn_id: turn_id.to_string(),
+            thread_settings: ThreadSettingsOverrides {
+                collaboration_mode: Some(CollaborationMode {
+                    mode: ModeKind::Plan,
+                    settings: Settings {
+                        model: test.session_configured.model.clone(),
+                        reasoning_effort: None,
+                        developer_instructions: None,
+                    },
+                }),
+                ..Default::default()
+            },
+            trace: None,
+        })
+        .await
+        .expect("recovered turn should start");
+    assert_eq!(
+        submission,
+        StartIfIdleSubmission::Started {
+            turn_id: turn_id.to_string(),
+        }
+    );
+
+    let started = wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnStarted(_))
+    })
+    .await;
+    let EventMsg::TurnStarted(started) = started else {
+        unreachable!("wait_for_event returned unexpected event");
+    };
+    assert_eq!(started.turn_id, turn_id);
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+
+    let user_input_groups = response_mock
+        .single_request()
+        .message_input_text_groups("user");
+    assert_eq!(user_input_groups.len(), 1);
+    assert_eq!(user_input_groups[0].len(), 1);
+    assert!(user_input_groups[0][0].starts_with("<environment_context>"));
+}
+
 /// Concurrent submissions must start exactly one turn and steer the other message.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn turn_input_submission_reports_started_and_steered_for_concurrent_submissions() {

codex-rs/protocol/src/protocol.rs

@@ -571,6 +571,12 @@ pub enum Op {
         reply: oneshot::Sender<CodexResult<TurnInputSubmission>>,
     },
 
+    /// Resume an interrupted regular turn.
+    RecoverTurn {
+        thread_settings: ThreadSettingsOverrides,
+        reply: oneshot::Sender<CodexResult<TurnInputSubmission>>,
+    },
+
     /// Apply persistent thread-settings overrides without starting a turn.
     ///
     /// This uses the same submission queue as turn starts so app-server can
@@ -864,6 +870,7 @@ impl Op {
             Self::RealtimeConversationClose => "realtime_conversation_close",
             Self::RealtimeConversationListVoices => "realtime_conversation_list_voices",
             Self::TurnInput { .. } => "turn_input",
+            Self::RecoverTurn { .. } => "recover_turn",
             Self::ThreadSettings { .. } => "thread_settings",
             Self::InterAgentCommunication { .. } => "inter_agent_communication",
             Self::ExecApproval { .. } => "exec_approval",

codex-rs/protocol/src/turn_input.rs

@@ -38,6 +38,17 @@ pub struct TurnInputRequest {
     pub trace: Option<W3cTraceContext>,
 }
 
+/// Request to resume sampling for an interrupted regular turn.
+///
+/// Sampling restarts under `turn_id`, which must be the ID already recorded
+/// for that turn.
+#[derive(Clone, Debug)]
+pub struct RecoverTurnRequest {
+    pub turn_id: String,
+    pub thread_settings: ThreadSettingsOverrides,
+    pub trace: Option<W3cTraceContext>,
+}
+
 impl TurnInputRequest {
     /// Creates turn input that can be passed to one of the submission methods.
     pub fn new(input: TurnInput) -> Self {