Continue interrupted work after managed daemon restarts (#45820)

## Why

Daemon recovery reloads threads but leaves interrupted work unfinished. Resume eligible work automatically from the saved conversation, even without a connected client.

## What changed

- Attempt one new continuation turn immediately after restoration, including in Plan mode. Mark the old turn interrupted and supply recovery context without creating a user message or granting new authorization.
- Require an idle thread, matching permissions, and the same single local environment configured by the thread. Skip completed, aborted, or superseded work and snapshots without environment identity.
- Add `continue_turn_if_idle` with an atomic previous-turn check so newer tasks or standalone settings changes invalidate pending continuation.
- Preserve the output schema, service tier, and root turn ID, and emit a “Resuming interrupted work” warning when continuation starts.

## Testing

Add daemon restart coverage for continuation without a client, Plan mode, permission and environment mismatches, and legacy snapshots. Add core coverage for preserved continuation metadata, absence of user-message events, rejection of superseded continuations, and exclusion of remote execution from recovery snapshots.

GitOrigin-RevId: 2b290b75d9399fc58258bcda2e85c74e9b5b3b09

codex-rs/app-server-transport/src/daemon_recovery.rs

@@ -22,6 +22,9 @@ pub struct InterruptedTurn {
     pub output_schema: Option<serde_json::Value>,
     pub service_tier: Option<String>,
     pub cyber_access_program: Option<codex_protocol::turn_input::CyberAccessProgram>,
+    /// Only local execution with thread-owned configuration can continue automatically.
+    /// Older snapshots without this identity are reloaded without continuation.
+    pub local_environment: Option<codex_app_server_protocol::ThreadEnvironment>,
 }
 
 // Old servers accept the array and skip this non-thread entry during best-effort

codex-rs/app-server/src/daemon_thread_recovery.rs

@@ -34,7 +34,7 @@ pub(crate) async fn start_recovery(
     processor: std::sync::Arc<crate::message_processor::MessageProcessor>,
 ) -> io::Result<tokio::task::JoinHandle<()>> {
     let candidates = tokio::task::spawn_blocking(move || {
-        let candidates = daemon_recovery::read_candidates(&path);
+        let candidates = daemon_recovery::read_snapshot(&path);
         // Even malformed or temporarily unreadable snapshots belong to this generation only.
         match std::fs::remove_file(&path) {
             Ok(()) => candidates,

codex-rs/app-server/src/message_processor.rs

@@ -765,18 +765,18 @@ impl MessageProcessor {
 
     pub(crate) async fn restore_daemon_threads(
         &self,
-        candidates: std::collections::BTreeSet<String>,
+        mut snapshot: codex_app_server_transport::daemon_recovery::RecoverySnapshot,
     ) {
-        for thread_id in candidates {
+        for thread_id in snapshot.loaded {
             let Ok(_permit) = self.turn_admission.admit() else {
                 break;
             };
             if let Err(err) = self
                 .thread_processor
                 .thread_resume(
-                    ThreadResumeTarget::DaemonRecovery,
+                    ThreadResumeTarget::DaemonRecovery(snapshot.interrupted.remove(&thread_id)),
                     codex_app_server_protocol::ThreadResumeParams {
-                        thread_id,
+                        thread_id: thread_id.clone(),
                         exclude_turns: true,
                         ..Default::default()
                     },

codex-rs/app-server/src/request_processors/daemon_continuation.rs

@@ -0,0 +1,135 @@
+//! Attempts one new continuation turn immediately after daemon thread restoration.
+//! Completed or superseded work is discarded; recovery never waits for a later idle event.
+//! Permissions and the saved local executor selection must match the resumed runtime.
+
+use super::ThreadRequestProcessor;
+use codex_app_server_protocol::ThreadEnvironment;
+use codex_app_server_transport::daemon_recovery::InterruptedTurn;
+use codex_core::TurnInput;
+use codex_core::TurnInputRequest;
+use codex_core::TurnInputSubmission;
+use codex_core::TurnStartOptions;
+use codex_core::context::ContextualUserFragment;
+use codex_core::context::InternalContextSource;
+use codex_core::context::InternalModelContextFragment;
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::EnvironmentConfigState;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::TurnAbortReason;
+use codex_protocol::protocol::TurnAbortedEvent;
+use codex_rollout::RolloutItem;
+
+impl ThreadRequestProcessor {
+    pub(crate) async fn continue_daemon_turn(&self, thread_id: &str, saved: InterruptedTurn) {
+        let Ok(thread_id) = ThreadId::from_string(thread_id) else {
+            return;
+        };
+        let Ok(thread) = self.thread_manager.get_thread(thread_id).await else {
+            return;
+        };
+        let Some(path) = thread.rollout_path() else {
+            return;
+        };
+        let history = match codex_rollout::RolloutRecorder::load_rollout_items(&path).await {
+            Ok((history, _, _)) => history,
+            Err(err) => {
+                tracing::warn!(%thread_id, %err, "failed to read interrupted turn history");
+                return;
+            }
+        };
+        // The snapshot has complete recorded input, but the old process may have
+        // completed or replaced its turn during the shutdown grace period.
+        let Some(start) = history
+            .iter()
+            .rposition(|item| matches!(item, RolloutItem::EventMsg(EventMsg::TurnStarted(_))))
+        else {
+            return;
+        };
+        if !matches!(&history[start], RolloutItem::EventMsg(EventMsg::TurnStarted(event))
+            if event.turn_id == saved.turn_id)
+            || history[start..].iter().any(|item| match item {
+                RolloutItem::EventMsg(EventMsg::TurnComplete(event)) => {
+                    event.turn_id == saved.turn_id
+                }
+                RolloutItem::EventMsg(EventMsg::TurnAborted(event)) => event
+                    .turn_id
+                    .as_ref()
+                    .is_none_or(|turn_id| turn_id == &saved.turn_id),
+                _ => false,
+            })
+        {
+            return;
+        }
+        let previous_context = history.iter().rev().find_map(|item| match item {
+            RolloutItem::TurnContext(context)
+                if context.turn_id.as_ref() == Some(&saved.turn_id) =>
+            {
+                Some(context)
+            }
+            _ => None,
+        });
+        let Some(previous_context) = previous_context else {
+            return;
+        };
+        let config = thread.config_snapshot().await;
+        let [environment] = config.environment_selections() else {
+            return;
+        };
+        if environment.environment_id != codex_exec_server::LOCAL_ENVIRONMENT_ID
+            || environment.config != EnvironmentConfigState::FromThread
+            || saved.local_environment.as_ref() != Some(&ThreadEnvironment::from(environment))
+        {
+            return;
+        }
+        // Recovery must not override a stricter saved or newly configured policy.
+        if previous_context.permission_profile() != config.permission_profile {
+            return;
+        }
+        let continuation = ContextualUserFragment::into(InternalModelContextFragment::new(
+            InternalContextSource::from_static("daemon_recovery"),
+            "The server restarted and interrupted the previous turn. Continue the unfinished work from the saved conversation. Check the current state before repeating actions that may already have completed.",
+        ));
+        let request = TurnInputRequest::new(TurnInput::ResponseItem(continuation)).on_start(
+            TurnStartOptions {
+                turn_trigger: Some("daemon_recovery".to_string()),
+                final_output_json_schema: saved.output_schema,
+                service_tier: saved.service_tier,
+                cyber_access_program: saved.cyber_access_program,
+                root_turn_id: previous_context.root_turn_id.clone(),
+                ..Default::default()
+            },
+        );
+        // Close the old turn in persisted history before exposing a new running turn.
+        if let Err(err) = thread
+            .append_rollout_items(&[RolloutItem::EventMsg(EventMsg::TurnAborted(
+                TurnAbortedEvent {
+                    turn_id: Some(saved.turn_id.clone()),
+                    reason: TurnAbortReason::Interrupted,
+                    started_at: None,
+                    completed_at: None,
+                    duration_ms: None,
+                },
+            ))])
+            .await
+        {
+            tracing::warn!(%thread_id, %err, "failed to close interrupted turn history");
+            return;
+        }
+        match thread.continue_turn_if_idle(request, saved.turn_id).await {
+            Ok(TurnInputSubmission::Started { .. }) => {
+                crate::extensions::send_thread_warning(
+                    &self.outgoing,
+                    &self.thread_state_manager,
+                    thread_id,
+                    "Resuming interrupted work".to_string(),
+                )
+                .await;
+            }
+            Ok(TurnInputSubmission::NotSubmitted { reason }) => {
+                tracing::debug!(%thread_id, ?reason, "recovery continuation was not started");
+            }
+            Ok(TurnInputSubmission::Steered { .. }) => unreachable!("continuation cannot steer"),
+            Err(err) => tracing::warn!(%thread_id, %err, "recovery continuation failed"),
+        }
+    }
+}

codex-rs/app-server/src/request_processors/daemon_snapshot.rs

@@ -37,7 +37,7 @@ impl ThreadRequestProcessor {
                     continue;
                 }
                 snapshot.loaded.insert(thread_id.to_string());
-                if let Some((turn_id, options)) = interrupted {
+                if let Some((turn_id, options, environment)) = interrupted {
                     snapshot.interrupted.insert(
                         thread_id.to_string(),
                         InterruptedTurn {
@@ -45,6 +45,7 @@ impl ThreadRequestProcessor {
                             output_schema: options.final_output_json_schema,
                             service_tier: options.service_tier,
                             cyber_access_program: options.cyber_access_program,
+                            local_environment: Some((&environment).into()),
                         },
                     );
                 }

codex-rs/app-server/src/request_processors/thread_processor.rs

@@ -1,3 +1,6 @@
+#[path = "daemon_continuation.rs"]
+mod daemon_continuation;
+
 #[path = "daemon_snapshot.rs"]
 mod daemon_snapshot;
 
@@ -460,7 +463,7 @@ pub(crate) struct ThreadRequestProcessor {
 /// Whether resume attaches a client or restores a cold runtime during daemon startup.
 pub(crate) enum ThreadResumeTarget {
     Client(ConnectionRequestId),
-    DaemonRecovery,
+    DaemonRecovery(Option<codex_app_server_transport::daemon_recovery::InterruptedTurn>),
 }
 
 /// Outcome of trying to satisfy a resume request from an already loaded thread.
@@ -3630,11 +3633,15 @@ impl ThreadRequestProcessor {
                 }
                 RunningThreadResumeResult::NotRunning(stored_thread) => stored_thread,
             },
-            ThreadResumeTarget::DaemonRecovery => {
+            ThreadResumeTarget::DaemonRecovery(saved) => {
                 let thread_id = ThreadId::from_string(&params.thread_id)
                     .map_err(|err| invalid_request(format!("invalid thread id: {err}")))?;
                 // Recheck under the same permit as client resume, including after config loading.
                 if self.thread_manager.get_thread(thread_id).await.is_ok() {
+                    if let Some(saved) = saved {
+                        self.continue_daemon_turn(&params.thread_id, saved.clone())
+                            .await;
+                    }
                     return Ok(ControlFlow::Break(()));
                 }
                 None
@@ -3918,7 +3925,7 @@ impl ThreadRequestProcessor {
                     ThreadResumeTarget::Client(request_id) => {
                         self.request_trace_context(request_id).await
                     }
-                    ThreadResumeTarget::DaemonRecovery => None,
+                    ThreadResumeTarget::DaemonRecovery(_) => None,
                 },
                 client_mcp_extensions,
             )
@@ -3939,6 +3946,10 @@ impl ThreadRequestProcessor {
                     codex_thread
                         .emit_thread_idle_lifecycle_if_idle(ThreadIdleCause::Completed)
                         .await;
+                    if let ThreadResumeTarget::DaemonRecovery(Some(saved)) = target {
+                        self.continue_daemon_turn(&thread_id.to_string(), saved.clone())
+                            .await;
+                    }
                     let state = self.thread_state_manager.thread_state(thread_id).await;
                     self.ensure_listener_task_running(thread_id, Arc::clone(&codex_thread), state)
                         .await?;

codex-rs/app-server/src/request_processors/turn_processor.rs

@@ -1150,7 +1150,9 @@ impl TurnRequestProcessor {
                         None,
                         None,
                     ),
-                    NotSubmittedReason::PendingTriggerTurn | NotSubmittedReason::PlanMode => (
+                    NotSubmittedReason::PendingTriggerTurn
+                    | NotSubmittedReason::PlanMode
+                    | NotSubmittedReason::Superseded => (
                         "no active turn to steer".to_string(),
                         None,
                         Some(AnalyticsJsonRpcError::TurnSteer(

codex-rs/app-server/tests/suite/v2/daemon_update_recovery.rs

@@ -460,6 +460,11 @@ async fn assert_still_running(server: &mut Child, message: &str) {
 }
 
 #[test_case::test_case("running")]
+#[test_case::test_case("plan")]
+#[test_case::test_case("read_only")]
+#[test_case::test_case("stricter_config")]
+#[test_case::test_case("different_environment")]
+#[test_case::test_case("legacy_snapshot")]
 #[test_case::test_case("completed")]
 #[test_case::test_case("canceled")]
 #[test_case::test_case("compacting")]
@@ -467,7 +472,7 @@ async fn assert_still_running(server: &mut Child, message: &str) {
 async fn managed_shutdown_records_interrupted_turn(outcome: &str) -> Result<()> {
     let home = TempDir::new()?;
     let (release, gate) = oneshot::channel();
-    let (_release_compaction, compaction_gate) = oneshot::channel();
+    let (release_compaction, compaction_gate) = oneshot::channel();
     let (mock, _completions) = start_streaming_sse_server(vec![
         vec![stream_chunk(Some(gate), "Original")?],
         vec![stream_chunk(Some(compaction_gate), "Compacted")?],
@@ -476,16 +481,36 @@ async fn managed_shutdown_records_interrupted_turn(outcome: &str) -> Result<()>
     create_config_toml(home.path(), mock.uri(), "never")?;
     let socket = home.path().join("control/server.sock");
     let mut server = spawn_server(home.path(), &socket)?;
-    let mut client = connect_default_daemon_client(&socket).await?;
+    let mut client = connect_daemon_client(
+        &socket,
+        InitializeCapabilities {
+            experimental_api: true,
+            ..Default::default()
+        },
+    )
+    .await?;
     let thread = start_thread(&mut client, /*id*/ 2, json!({})).await?;
+    let local_environment = thread
+        .thread
+        .environments
+        .as_ref()
+        .and_then(|environments| environments.first())
+        .cloned();
     let id = thread.thread.id;
+    let mode = if outcome == "plan" { "plan" } else { "default" };
     let schema = json!({"type":"object","properties":{},"additionalProperties":false});
     let turn = request(
         &mut client,
         /*id*/ 3,
         "turn/start",
         json!({
-            "threadId":id,"input":[{"type":"text","text":"Do the work"}],"outputSchema":schema
+            "threadId":id,"input":[{"type":"text","text":"Do the work"}],"outputSchema":schema,
+            "sandboxPolicy": match outcome {
+                "read_only" => json!({"type":"readOnly"}),
+                "stricter_config" => json!({"type":"workspaceWrite","writableRoots":[],"networkAccess":false,"excludeTmpdirEnvVar":false,"excludeSlashTmp":false}),
+                _ => serde_json::Value::Null,
+            },
+            "collaborationMode":{"mode":mode,"settings":{"model":"mock-model","reasoning_effort":null,"developer_instructions":null}}
         }),
     )
     .await?;
@@ -527,7 +552,12 @@ async fn managed_shutdown_records_interrupted_turn(outcome: &str) -> Result<()>
             .await?;
             wait_for_requests(&mock, /*count*/ 2).await?;
         }
-        "running" => {}
+        "running"
+        | "plan"
+        | "read_only"
+        | "stricter_config"
+        | "different_environment"
+        | "legacy_snapshot" => {}
         _ => unreachable!(),
     }
     request_shutdown(&server, &socket).await?;
@@ -538,14 +568,16 @@ async fn managed_shutdown_records_interrupted_turn(outcome: &str) -> Result<()>
         }
     })
     .await?;
-    let expected = if outcome == "running" {
+    let interrupted = !matches!(outcome, "completed" | "canceled" | "compacting");
+    let expected = if interrupted {
         [(
             id.clone(),
             daemon_recovery::InterruptedTurn {
                 turn_id: turn["turn"]["id"].as_str().context("turn ID")?.to_string(),
-                output_schema: Some(schema),
+                output_schema: Some(schema.clone()),
                 service_tier: Some("default".into()),
                 cyber_access_program: None,
+                local_environment,
             },
         )]
         .into()
@@ -560,11 +592,168 @@ async fn managed_shutdown_records_interrupted_turn(outcome: &str) -> Result<()>
         }
     );
     // Existing binaries can still read the candidate array and ignore the metadata entry.
-    let legacy: std::collections::BTreeSet<String> = serde_json::from_slice(&std::fs::read(path)?)?;
+    let legacy: std::collections::BTreeSet<String> =
+        serde_json::from_slice(&std::fs::read(&path)?)?;
     assert!(legacy.contains(&id));
     if server.try_wait()?.is_none() {
         server.kill().await?;
     }
+    if !interrupted {
+        return Ok(());
+    }
+    if outcome == "running" {
+        // A prior recovery can append an older turn's abort after this turn started.
+        codex_rollout::append_rollout_item_to_path(
+            thread.thread.path.as_ref().context("rollout path")?,
+            &serde_json::from_value(json!({
+                "type":"event_msg",
+                "payload":{"type":"turn_aborted","turn_id":"older-turn","reason":"interrupted"}
+            }))?,
+        )
+        .await?;
+    }
+    if outcome == "read_only" {
+        let config = home.path().join("config.toml");
+        let contents = std::fs::read_to_string(&config)?;
+        std::fs::write(
+            config,
+            contents.replace(
+                "sandbox_mode = \"read-only\"",
+                "default_permissions = \":workspace\"",
+            ),
+        )?;
+    }
+    if matches!(outcome, "different_environment" | "legacy_snapshot") {
+        let mut snapshot = daemon_recovery::read_snapshot(&path)?;
+        let saved = snapshot
+            .interrupted
+            .get_mut(&id)
+            .context("interrupted turn")?;
+        if outcome == "legacy_snapshot" {
+            saved.local_environment = None;
+        } else {
+            saved
+                .local_environment
+                .as_mut()
+                .context("local environment")?
+                .environment_id = "remote".into();
+        }
+        daemon_recovery::write_snapshot(&path, &snapshot)?;
+    }
+    let mut successor = spawn_server(home.path(), &socket)?;
+    if !matches!(outcome, "running" | "plan") {
+        let mut client = connect_default_daemon_client(&socket).await?;
+        timeout(DEFAULT_READ_TIMEOUT, async {
+            loop {
+                let loaded =
+                    request(&mut client, /*id*/ 2, "thread/loaded/list", json!({})).await?;
+                if loaded["data"]
+                    .as_array()
+                    .context("loaded IDs")?
+                    .contains(&json!(id))
+                {
+                    break;
+                }
+                sleep(Duration::from_millis(25)).await;
+            }
+            Ok::<(), anyhow::Error>(())
+        })
+        .await??;
+        // Resume serializes with the ongoing daemon restore, so the recovery attempt has finished.
+        request(
+            &mut client,
+            /*id*/ 3,
+            "thread/resume",
+            json!({"threadId":id,"excludeTurns":true}),
+        )
+        .await?;
+        let response = request(
+            &mut client,
+            /*id*/ 2,
+            "thread/read",
+            json!({"threadId":id,"includeTurns":true}),
+        )
+        .await?;
+        assert_eq!(response["thread"]["status"]["type"], "idle");
+        assert_eq!(mock.requests().await.len(), 1);
+        successor.kill().await?;
+        return Ok(());
+    }
+    // Recovery starts without a client, then its internal message survives reconnect.
+    wait_for_requests(&mock, /*count*/ 2).await?;
+    let mut client = connect_default_daemon_client(&socket).await?;
+    request(
+        &mut client,
+        /*id*/ 2,
+        "thread/resume",
+        json!({"threadId":id,"excludeTurns":true}),
+    )
+    .await?;
+    let response = request(
+        &mut client,
+        /*id*/ 3,
+        "thread/read",
+        json!({"threadId":id,"includeTurns":true}),
+    )
+    .await?;
+    let statuses: Vec<_> = response["thread"]["turns"]
+        .as_array()
+        .context("turns")?
+        .iter()
+        .filter(|turn| turn["id"] != "older-turn")
+        .map(|turn| turn["status"].clone())
+        .collect();
+    assert_eq!(statuses, vec![json!("interrupted"), json!("inProgress")]);
+    release_compaction
+        .send(())
+        .expect("continuation is waiting");
+    timeout(DEFAULT_READ_TIMEOUT, async {
+        loop {
+            let response = request(
+                &mut client,
+                /*id*/ 3,
+                "thread/read",
+                json!({"threadId":id,"includeTurns":true}),
+            )
+            .await?;
+            let turns = response["thread"]["turns"].as_array().context("turns")?;
+            if let Some(continued) = turns.last()
+                && continued["status"] == "completed"
+            {
+                assert_ne!(continued["id"], turn["turn"]["id"]);
+                assert!(
+                    continued["items"]
+                        .as_array()
+                        .context("items")?
+                        .iter()
+                        .all(|item| item["type"] != "userMessage")
+                );
+                break;
+            }
+            sleep(Duration::from_millis(25)).await;
+        }
+        Ok::<(), anyhow::Error>(())
+    })
+    .await??;
+    let requests = mock.requests().await;
+    assert_eq!(requests.len(), 2);
+    let body: serde_json::Value = serde_json::from_slice(&requests[1])?;
+    assert_eq!(body["text"]["format"]["schema"], schema);
+    let input = body["input"].to_string();
+    assert!(
+        input.contains("Do the work") && input.contains("Continue the unfinished work"),
+        "{input}"
+    );
+    let collaboration = input
+        .rsplit("<collaboration_mode>")
+        .next()
+        .context("collaboration instructions")?;
+    assert_eq!(
+        collaboration.contains("# Plan Mode"),
+        mode == "plan",
+        "{input}"
+    );
+    successor.kill().await?;
     Ok(())
 }
 

codex-rs/core/src/codex_thread.rs

@@ -357,6 +357,23 @@ impl CodexThread {
         }
     }
 
+    /// Starts a new internal continuation turn when idle, including in Plan mode.
+    /// Rejects if a newer task has started, even if it has already finished.
+    /// The input must be a response item; it is never treated as user authorization.
+    pub async fn continue_turn_if_idle(
+        &self,
+        request: TurnInputRequest,
+        expected_previous_turn_id: String,
+    ) -> CodexResult<TurnInputSubmission> {
+        self.submit_turn_input_with_mode(
+            request,
+            TurnInputMode::ContinueIfIdle {
+                expected_previous_turn_id,
+            },
+        )
+        .await
+    }
+
     /// Resumes an interrupted regular turn only when the thread is idle.
     ///
     /// Recovery starts no new user input and preserves the turn ID that was
@@ -498,7 +515,9 @@ impl CodexThread {
     }
 
     /// Captures a regular turn only after its input is recorded. The caller must flush the rollout.
-    pub async fn interrupted_turn(&self) -> Option<(String, TurnStartOptions)> {
+    pub async fn interrupted_turn(
+        &self,
+    ) -> Option<(String, TurnStartOptions, TurnEnvironmentSelection)> {
         self.session.interrupted_turn().await
     }
 

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

@@ -1,14 +1,18 @@
-//! Captures recorded, uncanceled regular turns for managed daemon recovery.
+//! Captures recorded, uncanceled regular turns using one thread-configured local executor.
 //! Callers must flush the rollout after capture before persisting the snapshot.
 
 use super::Session;
+use codex_protocol::protocol::EnvironmentConfigState;
+use codex_protocol::protocol::TurnEnvironmentSelection;
 
 /// Turn input and turn-start injections have entered the rollout writer.
 pub(super) struct RecordedTurnInput;
 
 impl Session {
     /// Captures a regular turn only after its input is recorded. The caller must flush the rollout.
-    pub(crate) async fn interrupted_turn(&self) -> Option<(String, crate::TurnStartOptions)> {
+    pub(crate) async fn interrupted_turn(
+        &self,
+    ) -> Option<(String, crate::TurnStartOptions, TurnEnvironmentSelection)> {
         let active = self.active_turn.lock().await;
         let task = active.as_ref()?.task.as_ref()?;
         if task.kind != crate::state::TaskKind::Regular || task.cancellation_token.is_cancelled() {
@@ -16,6 +20,14 @@ impl Session {
         }
         let context = &task.turn_context;
         context.extension_data.get::<RecordedTurnInput>()?;
+        // Remote identities/configuration are not persisted across daemon restarts.
+        if context.environments.environments.len() != 1 {
+            return None;
+        }
+        let environment = context.environments.single_local_environment()?.selection();
+        if environment.config != EnvironmentConfigState::FromThread {
+            return None;
+        }
         Some((
             context.sub_id.clone(),
             crate::TurnStartOptions {
@@ -34,6 +46,7 @@ impl Session {
                 cyber_access_program: context.cyber_access_program,
                 ..Default::default()
             },
+            environment,
         ))
     }
 }

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

@@ -1445,6 +1445,12 @@ impl Session {
                     &session_configuration.session_source,
                 ),
             );
+            state.last_started_turn_id = initial_history.get_rollout_items().iter().rev().find_map(|item| {
+                match item {
+                    RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => Some(event.turn_id.clone()),
+                    _ => None,
+                }
+            });
             state.base_instructions_provenance = base_instructions_provenance.clone();
             state.active_disabled_plugin_ids = session_configuration.disabled_plugin_ids.clone();
             let managed_network_requirements_configured = config

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

@@ -50,6 +50,9 @@ pub(super) async fn update(
                 }),
             })
             .await;
+    } else {
+        // Standalone settings changes supersede a pending automatic continuation.
+        session.state.lock().await.last_started_turn_id = None;
     }
 }
 

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

@@ -216,7 +216,31 @@ pub(super) async fn handle(
                 | SubmittedTurnInput::ResponseItem(_)
                 | SubmittedTurnInput::InterAgentCommunication(_) => TurnStartKind::Automatic,
             };
-            start_if_idle(session, request, submission_id, kind).await
+            start_if_idle(
+                session,
+                request,
+                submission_id,
+                kind,
+                /*expected_previous_turn_id*/ None,
+            )
+            .await
+        }
+        TurnInputMode::ContinueIfIdle {
+            expected_previous_turn_id,
+        } => {
+            if !matches!(&request.input, SubmittedTurnInput::ResponseItem(_)) {
+                return Err(CodexErr::InvalidRequest(
+                    "continuation requires internal response input".to_string(),
+                ));
+            }
+            start_if_idle(
+                session,
+                request,
+                submission_id,
+                TurnStartKind::Recovery,
+                Some(expected_previous_turn_id),
+            )
+            .await
         }
         TurnInputMode::Steer { expected_turn_id } => {
             steer(session, request, expected_turn_id, submission_id).await
@@ -236,7 +260,14 @@ pub(super) async fn handle_recovery(
             turn_trigger: Some("retry".to_string()),
             ..start_options
         });
-    start_if_idle(session, request, submission_id, TurnStartKind::Recovery).await
+    start_if_idle(
+        session,
+        request,
+        submission_id,
+        TurnStartKind::Recovery,
+        /*expected_previous_turn_id*/ None,
+    )
+    .await
 }
 
 async fn start_or_steer(
@@ -335,11 +366,16 @@ async fn start_or_steer(
     }
 }
 
+#[expect(
+    clippy::await_holding_invalid_type,
+    reason = "the previous turn check and idle reservation must be atomic"
+)]
 async fn start_if_idle(
     session: &Arc<Session>,
     request: TurnInputRequest,
     submission_id: String,
     kind: TurnStartKind,
+    expected_previous_turn_id: Option<String>,
 ) -> CodexResult<TurnInputSubmission> {
     let TurnInputRequest {
         input,
@@ -392,6 +428,13 @@ async fn start_if_idle(
                 reason: NotSubmittedReason::NotIdle,
             });
         }
+        if let Some(expected) = expected_previous_turn_id
+            && session.state.lock().await.last_started_turn_id.as_ref() != Some(&expected)
+        {
+            return Ok(TurnInputSubmission::NotSubmitted {
+                reason: NotSubmittedReason::Superseded,
+            });
+        }
         let active_turn = active_turn.get_or_insert_with(ActiveTurn::default);
         Arc::clone(&active_turn.turn_state)
     };
@@ -445,7 +488,7 @@ async fn start_if_idle(
             }
             task_input.push(pending_turn_input(session, input).await);
         }
-        TurnStartKind::Automatic => {
+        TurnStartKind::Automatic | TurnStartKind::Recovery => {
             // Empty automatic user input resumes sampling without a new message.
             if !matches!(&input, SubmittedTurnInput::UserInput { .. }) {
                 session
@@ -457,9 +500,6 @@ async fn start_if_idle(
                     .await;
             }
         }
-        TurnStartKind::Recovery => {
-            // Recovery resumes an existing turn without a new empty user message.
-        }
     }
     session
         .start_task(turn_context, task_input, RegularTask::new())
@@ -508,6 +548,11 @@ async fn steer(
 }
 
 impl Session {
+    /// Called under the active-turn lock before running any task or lifecycle callback.
+    pub(crate) async fn record_started_turn(&self, turn_id: &str) {
+        self.state.lock().await.last_started_turn_id = Some(turn_id.to_string());
+    }
+
     pub(crate) async fn route_realtime_text_input(
         self: &Arc<Self>,
         text: String,

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

@@ -84,6 +84,9 @@ pub(crate) struct SessionState {
     /// model/realtime handling on subsequent regular turns (including full-context
     /// reinjection after resume or `/compact`).
     previous_turn_settings: Option<PreviousTurnSettings>,
+    /// Latest task admitted in this runtime, retained across completion and history edits.
+    /// Cleared by standalone settings changes to invalidate pending continuation.
+    pub(crate) last_started_turn_id: Option<String>,
     /// Runtime accounting state for the active auto-compaction window.
     auto_compact_window: AutoCompactWindow,
     /// Original request effort for the current model while configuration updates remain active.
@@ -127,6 +130,7 @@ impl SessionState {
             mcp_dependency_prompted: HashSet::new(),
             additional_context: AdditionalContextStore::default(),
             previous_turn_settings: None,
+            last_started_turn_id: None,
             auto_compact_window: AutoCompactWindow::new_with_ids(auto_compact_window_ids),
             reasoning_effort_pin: ReasoningEffortPin::Unset,
             startup_prewarm: None,

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

@@ -278,6 +278,10 @@ impl Session {
         self.start_task(turn_context, input, task).await;
     }
 
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "record the started turn atomically with its active reservation"
+    )]
     pub(crate) async fn start_task<T: SessionTask>(
         self: &Arc<Self>,
         turn_context: Arc<TurnContext>,
@@ -309,6 +313,7 @@ impl Session {
         let (pending_items, _) = self.input_queue.drain_mailbox_input_items().await;
         let turn_state = {
             let mut active = self.active_turn.lock().await;
+            self.record_started_turn(&turn_context.sub_id).await;
             let turn = active.get_or_insert_with(ActiveTurn::default);
             debug_assert!(turn.task.is_none());
             Arc::clone(&turn.turn_state)

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

@@ -8,6 +8,9 @@ use codex_core::TurnInputRequest;
 use codex_core::TurnInputSubmission;
 use codex_core::TurnStartOptions;
 use codex_core::config::Constrained;
+use codex_core::context::ContextualUserFragment;
+use codex_core::context::InternalContextSource;
+use codex_core::context::InternalModelContextFragment;
 use codex_extension_api::ExtensionRegistryBuilder;
 use codex_extension_api::TurnStartAdmission;
 use codex_protocol::AgentPath;
@@ -516,6 +519,125 @@ async fn recover_turn_if_idle_preserves_id_and_resumes_plan_mode() {
     assert!(user_input_groups[0][0].starts_with("<environment_context>"));
 }
 
+/// Internal continuation creates a new turn without adding user authorization.
+#[tokio::test]
+async fn continue_turn_if_idle_starts_new_turn_with_internal_input() {
+    let server = responses::start_mock_server().await;
+    let test = test_codex()
+        .with_model("gpt-5.6-sol")
+        .with_config(|config| {
+            config
+                .features
+                .enable(codex_features::Feature::FastMode)
+                .unwrap();
+        })
+        .build_with_auto_env(&server)
+        .await
+        .unwrap();
+    responses::mount_sse_once(&server, responses::sse_completed("original")).await;
+    let TurnInputSubmission::Started {
+        turn_id: previous_turn_id,
+    } = test
+        .codex
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
+            text: "Do the work".to_string(),
+            text_elements: Vec::new(),
+        }]))
+        .await
+        .unwrap()
+    else {
+        panic!("original turn did not start")
+    };
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+    let mock = responses::mount_sse_once(&server, responses::sse_completed("continued")).await;
+    let input = ContextualUserFragment::into(InternalModelContextFragment::new(
+        InternalContextSource::from_static("daemon_recovery"),
+        "Continue the interrupted work.",
+    ));
+    let schema = serde_json::json!({"type":"object","properties":{},"additionalProperties":false});
+    let submission = test
+        .codex
+        .continue_turn_if_idle(
+            TurnInputRequest::new(TurnInput::ResponseItem(input.clone())).on_start(
+                TurnStartOptions {
+                    final_output_json_schema: Some(schema.clone()),
+                    service_tier: Some("priority".to_string()),
+                    root_turn_id: Some("originating-turn".to_string()),
+                    ..Default::default()
+                },
+            ),
+            previous_turn_id.clone(),
+        )
+        .await
+        .unwrap();
+    let TurnInputSubmission::Started { turn_id } = submission else {
+        panic!("continuation did not start")
+    };
+    wait_for_event(&test.codex, |event| {
+        assert!(!matches!(event, EventMsg::UserMessage(_)));
+        assert!(!matches!(event, EventMsg::ItemCompleted(event)
+            if matches!(event.item, codex_protocol::items::TurnItem::UserMessage(_))));
+        matches!(event, EventMsg::TurnComplete(_))
+    })
+    .await;
+    let request = mock.single_request();
+    assert!(request.has_content_kinds(&["daemon_recovery.internal_context"]));
+    let metadata: Value = serde_json::from_str(
+        &request
+            .header("x-codex-turn-metadata")
+            .expect("turn metadata"),
+    )
+    .unwrap();
+    assert_eq!(metadata["root_turn_id"], "originating-turn");
+    let body = request.body_json();
+    assert_eq!(body["text"]["format"]["schema"], schema);
+    assert_eq!(body["service_tier"], "priority");
+    // The continuation has completed, but the saved previous ID is still stale.
+    assert_eq!(
+        test.codex
+            .continue_turn_if_idle(
+                TurnInputRequest::new(TurnInput::ResponseItem(ContextualUserFragment::into(
+                    InternalModelContextFragment::new(
+                        InternalContextSource::from_static("daemon_recovery"),
+                        "Continue."
+                    ),
+                ))),
+                previous_turn_id,
+            )
+            .await
+            .unwrap(),
+        TurnInputSubmission::NotSubmitted {
+            reason: NotSubmittedReason::Superseded
+        }
+    );
+    assert_eq!(mock.requests().len(), 1);
+    submit_thread_settings(
+        &test.codex,
+        ThreadSettingsOverrides {
+            approval_policy: Some(AskForApproval::OnRequest),
+            ..Default::default()
+        },
+    )
+    .await
+    .expect("update permissions before continuation admission");
+    assert_eq!(
+        test.codex
+            .continue_turn_if_idle(
+                TurnInputRequest::new(TurnInput::ResponseItem(input)),
+                turn_id,
+            )
+            .await
+            .unwrap(),
+        TurnInputSubmission::NotSubmitted {
+            reason: NotSubmittedReason::Superseded
+        }
+    );
+    assert_eq!(mock.requests().len(), 1);
+}
+
 /// 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() {
@@ -833,17 +955,32 @@ async fn start_or_steer_turn_requires_matching_active_output_schema() {
     server.shutdown().await;
 }
 
-#[test_case(Vec::new(); "automatic")]
-#[test_case(vec![UserInput::Text { text: "Do the work".into(), text_elements: Vec::new() }]; "user")]
+#[test_case(Vec::new(), "local"; "automatic")]
+#[test_case(vec![UserInput::Text { text: "Do the work".into(), text_elements: Vec::new() }], "local"; "user")]
+#[cfg_attr(unix, test_case(Vec::new(), "remote"; "remote_stays_idle"))]
 #[tokio::test]
-async fn sampling_is_ready_for_daemon_recovery(input: Vec<UserInput>) -> anyhow::Result<()> {
+async fn sampling_is_ready_for_daemon_recovery(
+    input: Vec<UserInput>,
+    executor: &str,
+) -> anyhow::Result<()> {
     let (release, gate) = oneshot::channel();
     let (server, _completions) = start_streaming_sse_server(vec![vec![StreamingSseChunk {
         gate: Some(gate),
         body: responses::sse_completed("automatic"),
     }]])
     .await;
-    let test = test_codex().build_with_streaming_server(&server).await?;
+    #[cfg(unix)]
+    let remote = if executor == "remote" {
+        Some(super::multi_exec_server_sandbox::ExecServerProcess::start().await?)
+    } else {
+        None
+    };
+    let mut builder = test_codex();
+    #[cfg(unix)]
+    if let Some(remote) = &remote {
+        builder = builder.with_exec_server_url(&remote.websocket_url);
+    }
+    let test = builder.build_with_streaming_server(&server).await?;
     let StartIfIdleSubmission::Started { turn_id } = test
         .codex
         .start_turn_if_idle(TurnInputRequest::user_input(input))
@@ -857,7 +994,10 @@ async fn sampling_is_ready_for_daemon_recovery(input: Vec<UserInput>) -> anyhow:
     )
     .await?;
     let active = test.codex.interrupted_turn().await;
-    assert_eq!(active.map(|(id, _)| id), Some(turn_id));
+    assert_eq!(
+        active.map(|(id, _, _)| id),
+        (executor == "local").then_some(turn_id)
+    );
     release.send(()).expect("sampling is waiting");
     wait_for_event(&test.codex, |event| {
         matches!(event, EventMsg::TurnComplete(_))

codex-rs/protocol/src/turn_input.rs

@@ -135,6 +135,9 @@ pub enum TurnInputMode {
     StartOrSteer,
     /// Start only when the thread is idle.
     StartIfIdle,
+    /// Start an internal continuation when idle.
+    /// Reject if another task has started since the expected previous turn.
+    ContinueIfIdle { expected_previous_turn_id: String },
     /// Steer only if this exact turn is active.
     Steer { expected_turn_id: String },
 }
@@ -215,6 +218,9 @@ pub enum SteerSubmission {
 /// Why Core did not accept submitted turn input for turn processing.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub enum NotSubmittedReason {
+    /// New work superseded the expected previous turn of an internal continuation.
+    Superseded,
+
     /// The host is draining and no longer permits new regular turns.
     ServerDraining,
 

codex-rs/tui/src/app/agents_overview_tests.rs

@@ -1564,8 +1564,10 @@ async fn root_switch_loads_local_preferences_from_disk() -> Result<()> {
 async fn root_switch_preserves_idle_root_with_running_subagent() -> Result<()> {
     let mut app = make_test_app().await;
     trust_fixture_folders(&mut app);
-    let mut app_server =
-        crate::start_embedded_app_server_for_picker(app.chat_widget.config_ref()).await?;
+    let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker(
+        app.chat_widget.config_ref(),
+    ))
+    .await?;
     let previous = app_server.start_thread(&app.config).await?;
     let previous_root_id = previous.session.thread_id;
     app.enqueue_primary_thread_session(previous.session, previous.turns)
@@ -1662,7 +1664,8 @@ async fn overview_selection_applies_user_permissions_only_to_unloaded_threads()
             .expect("create historical session"),
         )?);
     }
-    let mut app_server = crate::start_embedded_app_server_for_picker(&server_config).await?;
+    let mut app_server =
+        Box::pin(crate::start_embedded_app_server_for_picker(&server_config)).await?;
     let loaded = app_server
         .resume_thread(
             &crate::local_settings::LocalSettings::from(&server_config),

codex-rs/tui/src/app/tests.rs

@@ -7952,12 +7952,12 @@ async fn remembered_current_cwd_stays_at_launch_across_in_app_resumes() -> Resul
     let state_db =
         crate::init_state_db_for_app_server_target(&config, &crate::AppServerTarget::Embedded)
             .await?;
-    let mut app_server = crate::start_app_server_for_picker(
+    let mut app_server = Box::pin(crate::start_app_server_for_picker(
         &config,
         &crate::AppServerTarget::Embedded,
         state_db.clone(),
         Arc::new(EnvironmentManager::default_for_tests()),
-    )
+    ))
     .await?;
     let (mut app, _app_event_rx, _op_rx) = make_test_app_with_channels().await;
     app.config = config;
@@ -7968,9 +7968,8 @@ async fn remembered_current_cwd_stays_at_launch_across_in_app_resumes() -> Resul
     let mut tui = crate::tui::test_support::make_test_tui()?;
 
     for target_session in targets {
-        let control = app
-            .resume_target_session(&mut tui, &mut app_server, target_session)
-            .await?;
+        let control =
+            Box::pin(app.resume_target_session(&mut tui, &mut app_server, target_session)).await?;
 
         assert!(matches!(control, AppRunControl::Continue));
         assert!(!crate::session_resume::cwds_differ(

codex-rs/tui/src/app/tests/session_lifecycle_requests.rs

@@ -805,11 +805,11 @@ fn spawn_approved_task_tool_call(
 #[tokio::test]
 async fn external_transport_registers_dynamic_tools_and_finds_task_mentions() -> Result<()> {
     let (app, _codex_home) = make_history_test_app().await?;
-    let (mut app_server, requests, proxy) = start_recording_app_server(
+    let (mut app_server, requests, proxy) = Box::pin(start_recording_app_server(
         &app.config,
         /*blocked_thread_list*/ None,
         /*failed_thread_name*/ None,
-    )
+    ))
     .await?;
 
     let started = app_server.start_thread(&app.config).await?;
@@ -858,25 +858,27 @@ async fn external_transport_registers_dynamic_tools_and_finds_task_mentions() ->
     app_server.shutdown().await?;
     proxy.await??;
     let (mut restarted_app_server, _restarted_requests, restarted_proxy) =
-        start_recording_app_server(
+        Box::pin(start_recording_app_server(
             &app.config,
             /*blocked_thread_list*/ None,
             /*failed_thread_name*/ None,
-        )
-        .await?;
-    let resumed = restarted_app_server
-        .resume_thread(
-            &app.local_settings,
-            app.config.clone(),
-            target_id,
-            crate::app_server_session::ResumeModelSettings::RestoreFromThread,
-        )
+        ))
         .await?;
+    let resumed = Box::pin(restarted_app_server.resume_thread(
+        &app.local_settings,
+        app.config.clone(),
+        target_id,
+        crate::app_server_session::ResumeModelSettings::RestoreFromThread,
+    ))
+    .await?;
     assert!(resumed.task_tools_available);
     assert!(restarted_app_server.task_tools_available(target_id));
-    let forked = restarted_app_server
-        .fork_thread(&app.local_settings, app.config.clone(), target_id)
-        .await?;
+    let forked = Box::pin(restarted_app_server.fork_thread(
+        &app.local_settings,
+        app.config.clone(),
+        target_id,
+    ))
+    .await?;
     assert!(forked.task_tools_available);
     assert!(restarted_app_server.task_tools_available(forked.session.thread_id));
     restarted_app_server

codex-rs/tui/src/app/tests/startup.rs

@@ -1377,7 +1377,7 @@ async fn owned_subagent_approval_before_thread_started_is_preserved() -> Result<
     let codex_home = tempdir()?;
     app.config.codex_home = codex_home.path().to_path_buf().abs();
     app.config.sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs());
-    let mut app_server = crate::start_embedded_app_server_for_picker(&app.config).await?;
+    let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?;
     let parent = app_server.start_thread(&app.config).await?;
     let parent_thread_id = parent.session.thread_id;
     app.enqueue_primary_thread_session(parent.session, parent.turns)