core: make idle resume decisions atomic

codex-rs/core/src/agent/control.rs

@@ -352,15 +352,6 @@ impl AgentControl {
             .ok_or(CodexErr::ThreadNotFound(agent_id))
     }
 
-    pub(crate) async fn list_live_agent_subtree_thread_ids(
-        &self,
-        agent_id: ThreadId,
-    ) -> CodexResult<Vec<ThreadId>> {
-        let mut thread_ids = vec![agent_id];
-        thread_ids.extend(self.live_thread_spawn_descendants(agent_id).await?);
-        Ok(thread_ids)
-    }
-
     pub(crate) async fn get_agent_config_snapshot(
         &self,
         agent_id: ThreadId,

codex-rs/core/src/agent/control_tests.rs

@@ -2768,6 +2768,7 @@ async fn list_agent_subtree_thread_ids_finds_live_descendants_of_unloaded_root()
         .expect("grandchild spawn should succeed");
 
     manager.remove_thread(&parent_thread_id).await;
+    manager.remove_thread(&child_thread_id).await;
 
     let mut subtree_thread_ids = manager
         .list_agent_subtree_thread_ids(parent_thread_id)

codex-rs/core/src/codex_thread.rs

@@ -9,6 +9,7 @@ use codex_otel::SessionTelemetry;
 use codex_protocol::ThreadId;
 use codex_protocol::config_types::ApprovalsReviewer;
 use codex_protocol::config_types::CollaborationMode;
+use codex_protocol::config_types::ModeKind;
 use codex_protocol::config_types::Personality;
 use codex_protocol::config_types::ReasoningSummary;
 use codex_protocol::config_types::WindowsSandboxLevel;
@@ -81,6 +82,33 @@ pub struct ThreadConfigSnapshot {
     pub originator: String,
 }
 
+/// Runtime configuration captured only while a thread is eligible for idle resume handling.
+pub struct IdleResumeSnapshot {
+    pub thread_config_snapshot: ThreadConfigSnapshot,
+    pub original_config: Arc<crate::config::Config>,
+}
+
+/// Synchronous decision made while an idle thread is reserved against new work.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum IdleResumeDecision {
+    /// Keep using the existing thread.
+    Reuse,
+    /// Shut down the existing thread before replacing it.
+    Shutdown,
+}
+
+/// Result of atomically resolving whether an existing thread can be reused.
+pub enum IdleResumeOutcome {
+    /// Work is active or queued, so the caller should rejoin or retry later.
+    Active,
+    /// The reserved thread can be reused with this captured configuration.
+    Reused(Box<IdleResumeSnapshot>),
+    /// This call committed a shutdown submission.
+    ShutdownCommitted,
+    /// Another call already committed shutdown for the thread.
+    ShutdownPending,
+}
+
 /// Explains why `CodexThread::try_start_turn_if_idle` rejected an automatic
 /// idle turn.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -130,6 +158,20 @@ impl ThreadConfigSnapshot {
         &self.environments.environments
     }
 
+    /// Checks the default-mode fields that must match before reusing an idle thread.
+    pub fn matches_default_resume_mode_and_environments(
+        &self,
+        desired_environments: &[TurnEnvironmentSelection],
+    ) -> bool {
+        self.collaboration_mode.mode == ModeKind::Default
+            && self
+                .collaboration_mode
+                .settings
+                .developer_instructions
+                .is_none()
+            && self.environments.environments == desired_environments
+    }
+
     pub fn sandbox_policy(&self) -> SandboxPolicy {
         codex_sandboxing::compatibility_sandbox_policy_for_permission_profile(
             &self.permission_profile,
@@ -215,6 +257,22 @@ impl CodexThread {
         self.codex.try_shutdown_if_idle().await
     }
 
+    /// Atomically decides whether an idle thread should be reused or shut down.
+    ///
+    /// `decide` runs synchronously while submission delivery is blocked. It must be fast and must
+    /// not re-enter this thread.
+    pub async fn resolve_idle_resume<F>(&self, decide: F) -> CodexResult<IdleResumeOutcome>
+    where
+        F: FnOnce(&IdleResumeSnapshot) -> IdleResumeDecision + Send,
+    {
+        self.codex.resolve_idle_resume(decide).await
+    }
+
+    /// Returns the immutable source assigned when the thread was created.
+    pub fn session_source(&self) -> &SessionSource {
+        &self.session_source
+    }
+
     /// Wait until the underlying session loop has terminated.
     pub async fn wait_until_terminated(&self) {
         self.codex.session_loop_termination.clone().await;

codex-rs/core/src/lib.rs

@@ -26,6 +26,9 @@ mod config_lock;
 pub use codex_thread::BackgroundTerminalInfo;
 pub use codex_thread::CodexThread;
 pub use codex_thread::CodexThreadSettingsOverrides;
+pub use codex_thread::IdleResumeDecision;
+pub use codex_thread::IdleResumeOutcome;
+pub use codex_thread::IdleResumeSnapshot;
 pub use codex_thread::ThreadConfigSnapshot;
 pub use codex_thread::TryStartTurnIfIdleError;
 pub use codex_thread::TryStartTurnIfIdleRejectionReason;

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

@@ -175,6 +175,9 @@ use tracing::warn;
 use uuid::Uuid;
 
 use crate::client::ModelClient;
+use crate::codex_thread::IdleResumeDecision;
+use crate::codex_thread::IdleResumeOutcome;
+use crate::codex_thread::IdleResumeSnapshot;
 use crate::codex_thread::ThreadConfigSnapshot;
 #[cfg(test)]
 use crate::compact::collect_user_messages;
@@ -814,38 +817,69 @@ impl Codex {
     /// Atomically submits shutdown only when no work is active, queued, or
     /// being dispatched. Once claimed, new submissions and automatic idle work
     /// are rejected so shutdown cannot race a newly starting turn.
+    pub async fn try_shutdown_if_idle(&self) -> CodexResult<bool> {
+        match self
+            .resolve_idle_resume(|_| IdleResumeDecision::Shutdown)
+            .await?
+        {
+            IdleResumeOutcome::Active => Ok(false),
+            IdleResumeOutcome::ShutdownCommitted | IdleResumeOutcome::ShutdownPending => Ok(true),
+            IdleResumeOutcome::Reused(_) => unreachable!("shutdown decision cannot reuse thread"),
+        }
+    }
+
     #[expect(
         clippy::await_holding_invalid_type,
-        reason = "the send lock serializes the idle claim with submission delivery"
+        reason = "the send lock and idle reservation make the reuse decision atomic with submission delivery"
     )]
-    pub async fn try_shutdown_if_idle(&self) -> CodexResult<bool> {
+    pub(crate) async fn resolve_idle_resume<F>(&self, decide: F) -> CodexResult<IdleResumeOutcome>
+    where
+        F: FnOnce(&IdleResumeSnapshot) -> IdleResumeDecision + Send,
+    {
         if self.submission_transport == SubmissionTransport::ForwardingProxy {
             return Err(CodexErr::InvalidRequest(
                 "idle shutdown is unavailable through a forwarding proxy".to_string(),
             ));
         }
         let _send_guard = self.session.submission_send_lock.lock().await;
         if self.session.submission_lifecycle.is_closing() {
-            return Ok(true);
+            return Ok(IdleResumeOutcome::ShutdownPending);
         }
         let Some(mut reservation) = self.session.try_claim_idle_shutdown().await else {
-            return Ok(false);
+            return Ok(IdleResumeOutcome::Active);
         };
-        if !reservation.prepare_idle_shutdown_delivery() {
-            return Ok(false);
-        }
-        let sub = Submission {
-            id: new_submission_id(),
-            op: Op::Shutdown,
-            client_user_message_id: None,
-            trace: current_span_w3c_trace_context(),
+        let snapshot = {
+            let state = self.session.state.lock().await;
+            IdleResumeSnapshot {
+                thread_config_snapshot: state.session_configuration.thread_config_snapshot(),
+                original_config: Arc::clone(
+                    &state.session_configuration.original_config_do_not_use,
+                ),
+            }
         };
-        if self.tx_sub.send(sub).await.is_err() {
-            reservation.release_after_failed_delivery();
-            return Err(CodexErr::InternalAgentDied);
+        match decide(&snapshot) {
+            IdleResumeDecision::Reuse => {
+                drop(reservation);
+                Ok(IdleResumeOutcome::Reused(Box::new(snapshot)))
+            }
+            IdleResumeDecision::Shutdown => {
+                if !reservation.prepare_idle_shutdown_delivery() {
+                    return Ok(IdleResumeOutcome::Active);
+                }
+                let sub = Submission {
+                    id: new_submission_id(),
+                    op: Op::Shutdown,
+                    client_user_message_id: None,
+                    trace: current_span_w3c_trace_context(),
+                };
+                if self.tx_sub.send(sub).await.is_err() {
+                    reservation.release_after_failed_delivery();
+                    return Err(CodexErr::InternalAgentDied);
+                }
+                reservation.commit();
+                Ok(IdleResumeOutcome::ShutdownCommitted)
+            }
         }
-        reservation.commit();
-        Ok(true)
     }
 
     /// Persist a thread-level memory mode update for the active session.
@@ -1668,12 +1702,20 @@ impl Session {
         state.session_configuration.provider.clone()
     }
 
+    #[expect(
+        clippy::await_holding_invalid_type,
+        reason = "the send lock serializes the config swap and the activity reservation covers the full refresh"
+    )]
     pub(crate) async fn refresh_runtime_config(&self, next_config: Config) {
         // Refresh only the user layer from the incoming snapshot. Preserve thread-local
         // layers such as request/session overrides that were present when this session
         // was created.
         let notify_config_contributors = !self.services.extensions.config_contributors().is_empty();
-        let (previous_config, new_config, config) = {
+        let (reservation, previous_config, new_config, config) = {
+            let _send_guard = self.submission_send_lock.lock().await;
+            let Some(reservation) = self.try_reserve_activity() else {
+                return;
+            };
             let mut state = self.state.lock().await;
             let previous_config = notify_config_contributors
                 .then(|| Self::build_effective_session_config(&state.session_configuration));
@@ -1687,7 +1729,7 @@ impl Session {
             state.session_configuration.original_config_do_not_use = Arc::clone(&config);
             let new_config = notify_config_contributors
                 .then(|| Self::build_effective_session_config(&state.session_configuration));
-            (previous_config, new_config, config)
+            (reservation, previous_config, new_config, config)
         };
         self.emit_config_changed_contributors(previous_config.as_ref(), new_config.as_ref());
         self.services.skills_service.clear_cache();
@@ -1709,6 +1751,8 @@ impl Session {
         ) {
             self.services.hooks.store(Arc::new(hooks));
         }
+        drop(state);
+        drop(reservation);
     }
 
     fn emit_config_changed_contributors(