From 363427b5e3fe1b6d7499e6bc47651f62a5a3b1d2 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Thu, 13 Aug 2026 02:58:29 +0000 Subject: [PATCH] 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 | 1 + codex-rs/core/src/codex_thread.rs | 36 +++++++++++ codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/session/handlers.rs | 9 +++ codex-rs/core/src/session/mod.rs | 22 +++++++ codex-rs/core/src/session/turn_input.rs | 25 ++++++-- codex-rs/core/src/session/turn_input_tests.rs | 58 +++++++++++++++++ .../core/tests/suite/turn_input_submission.rs | 62 +++++++++++++++++++ codex-rs/protocol/src/protocol.rs | 7 +++ codex-rs/protocol/src/turn_input.rs | 11 ++++ 10 files changed, 228 insertions(+), 4 deletions(-) diff --git a/codex-rs/core-api/src/lib.rs b/codex-rs/core-api/src/lib.rs index 0c5f29fe94..930631b418 100644 --- a/codex-rs/core-api/src/lib.rs +++ b/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; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index c5e4289412..5d47188b7f 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/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 { + 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, diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index d1d1f5802a..214a1b188f 100644 --- a/codex-rs/core/src/lib.rs +++ b/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; diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index e3e5084055..e6ae9f1b44 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/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 diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index d74773423d..560744fc52 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/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, + turn_id: String, + ) -> CodexResult { + 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 { diff --git a/codex-rs/core/src/session/turn_input.rs b/codex-rs/core/src/session/turn_input.rs index 1343af3ffe..90e456025e 100644 --- a/codex-rs/core/src/session/turn_input.rs +++ b/codex-rs/core/src/session/turn_input.rs @@ -146,13 +146,24 @@ pub(super) async fn handle( ) -> CodexResult { 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, + thread_settings: ThreadSettingsOverrides, + submission_id: String, +) -> CodexResult { + 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, request: TurnInputRequest, @@ -242,6 +253,7 @@ async fn start_if_idle( session: &Arc, request: TurnInputRequest, submission_id: String, + is_recovery: bool, ) -> CodexResult { 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( diff --git a/codex-rs/core/src/session/turn_input_tests.rs b/codex-rs/core/src/session/turn_input_tests.rs index 4cfeefe2cb..b1eb5f07fe 100644 --- a/codex-rs/core/src/session/turn_input_tests.rs +++ b/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::::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; diff --git a/codex-rs/core/tests/suite/turn_input_submission.rs b/codex-rs/core/tests/suite/turn_input_submission.rs index c40085fc21..4d70eca48b 100644 --- a/codex-rs/core/tests/suite/turn_input_submission.rs +++ b/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("")); +} + /// 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() { diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 56a6aff271..e95dd08caf 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -571,6 +571,12 @@ pub enum Op { reply: oneshot::Sender>, }, + /// Resume an interrupted regular turn. + RecoverTurn { + thread_settings: ThreadSettingsOverrides, + reply: oneshot::Sender>, + }, + /// 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", diff --git a/codex-rs/protocol/src/turn_input.rs b/codex-rs/protocol/src/turn_input.rs index 6506364a37..bc5adbe8b5 100644 --- a/codex-rs/protocol/src/turn_input.rs +++ b/codex-rs/protocol/src/turn_input.rs @@ -38,6 +38,17 @@ pub struct TurnInputRequest { pub trace: Option, } +/// 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, +} + impl TurnInputRequest { /// Creates turn input that can be passed to one of the submission methods. pub fn new(input: TurnInput) -> Self {