Restore thread cwd from owned settings snapshots (#41567)

## Why

Resuming a thread without an explicit `cwd` should restore that thread's latest
retained setting. Forked history can contain settings copied from another
thread, while compaction can move the latest setting outside the replay window.

## What changed

- Record the owning thread ID on new settings snapshots and only use snapshots
  owned by the resumed thread when restoring `cwd`. Older snapshots without an
  owner remain readable but do not override the startup `cwd`.
- Checkpoint the current settings after compaction and serialize checkpoints
  with settings updates so the retained history contains an accepted, current
  snapshot.
- Cover resume behavior across compaction, forks, reverts, legacy histories,
  and concurrent settings updates.

GitOrigin-RevId: 1efc9cf55472d6b88c465c4efe44672a232dbf64

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

@@ -3634,6 +3634,9 @@ impl Session {
                 .map(|id| id.to_string()),
             window_id: Some(metadata.window_ids.window_id.to_string()),
         };
+        // Wait for accepted updates to finish persisting, then keep later updates from
+        // overtaking the current settings snapshot while its checkpoint is written.
+        let _settings_guard = thread_settings::acquire_persistence_lock(self).await;
         // Compaction starts a new history window, so its WorldState baseline must be full.
         let mut world_state_item = None;
         {
@@ -3646,17 +3649,19 @@ impl Session {
             }
         }
 
-        self.persist_rollout_items(&[RolloutItem::Compacted(compacted_item)])
-            .await;
+        let mut rollout_items = vec![RolloutItem::Compacted(compacted_item)];
         // Persist the baseline after the replacement history that established it.
         if let Some(world_state_item) = world_state_item {
-            self.persist_rollout_items(&[RolloutItem::WorldState(world_state_item)])
-                .await;
+            rollout_items.push(RolloutItem::WorldState(world_state_item));
         }
         if let Some(turn_context_item) = reference_context_item {
-            self.persist_rollout_items(&[RolloutItem::TurnContext(turn_context_item)])
-                .await;
+            rollout_items.push(RolloutItem::TurnContext(turn_context_item));
         }
+        // The frozen turn context must not override current settings in persisted metadata.
+        rollout_items.push(RolloutItem::EventMsg(
+            thread_settings::applied_event(self).await,
+        ));
+        self.persist_rollout_items(&rollout_items).await;
         {
             let mut state = self.state.lock().await;
             state.queue_pending_session_start_source(codex_hooks::SessionStartSource::Compact);

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

@@ -44,6 +44,9 @@ pub(crate) struct Session {
     pub(super) tx_event: Sender<Event>,
     pub(super) agent_status: watch::Sender<AgentStatus>,
     pub(super) state: Mutex<SessionState>,
+    /// Orders accepted settings commits and their persisted events with compaction checkpoints.
+    /// Keep this separate from `state` so storage I/O does not block runtime state access.
+    pub(super) thread_settings_persistence: Semaphore,
     /// Serializes rebuild/apply cycles for the running proxy; each cycle
     /// rebuilds from the current SessionState while holding this lock.
     pub(super) managed_network_proxy_refresh_lock: Semaphore,
@@ -1469,6 +1472,7 @@ impl Session {
                 tx_event: tx_event.clone(),
                 agent_status,
                 state: Mutex::new(state),
+                thread_settings_persistence: Semaphore::new(/*permits*/ 1),
                 managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1),
                 features: config.features.clone(),
                 windows_sandbox_proxy_settings_mode,

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

@@ -5817,6 +5817,101 @@ async fn session_configuration_apply_preserves_absolute_cwd_write_root_on_cwd_up
     );
 }
 
+#[tokio::test]
+async fn compaction_checkpoint_waits_for_accepted_settings_persistence() {
+    let (mut session, _turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx(
+        CodexAuth::from_api_key("Test API Key"),
+        Vec::new(),
+        |config| {
+            config
+                .permissions
+                .set_permission_profile(PermissionProfile::workspace_write())
+                .expect("set initial permission profile");
+        },
+    )
+    .await;
+    let rollout_path =
+        attach_thread_persistence(Arc::get_mut(&mut session).expect("unique session")).await;
+    let refresh_guard = session
+        .managed_network_proxy_refresh_lock
+        .acquire()
+        .await
+        .expect("network refresh lock");
+    let mut update = Box::pin(tokio::task::unconstrained(thread_settings::apply_update(
+        &session,
+        "settings".to_string(),
+        SessionSettingsUpdate {
+            step_settings: StepSettingsUpdate {
+                service_tier: Some(Some(ServiceTier::Fast.request_value().to_string())),
+                ..Default::default()
+            },
+            permission_profile: Some(PermissionProfile::read_only()),
+            ..Default::default()
+        },
+    )));
+    // Pause after committing settings but before their accepted snapshot is persisted.
+    assert!(futures::poll!(update.as_mut()).is_pending());
+    let committed = session.thread_settings_snapshot().await;
+    let history_before = session.clone_history().await;
+    let (window_number, window_ids) = session.advance_auto_compact_window().await;
+    let mut checkpoint = Box::pin(tokio::task::unconstrained(
+        session.replace_compacted_history(
+            vec![ResponseItemEnvelope::new(user_message("compacted history"))],
+            /*reference_context_item*/ None,
+            /*world_state_baseline*/ None,
+            CompactedHistoryMetadata {
+                message: "summary".to_string(),
+                window_number,
+                window_ids,
+            },
+        ),
+    ));
+    assert!(futures::poll!(checkpoint.as_mut()).is_pending());
+    assert_eq!(
+        session.clone_history().await.annotated_items(),
+        history_before.annotated_items()
+    );
+
+    // Direct runtime restoration may overlap postcommit work. The checkpoint must wait
+    // for the accepted event, then capture current settings rather than its older commit.
+    let restored = session
+        .update_settings(SessionSettingsUpdate {
+            step_settings: StepSettingsUpdate {
+                service_tier: Some(None),
+                ..Default::default()
+            },
+            ..Default::default()
+        })
+        .await
+        .expect("restore current settings")
+        .snapshot;
+    assert_ne!(committed, restored);
+    drop(refresh_guard);
+    update.await.expect("accepted settings update");
+    checkpoint.await;
+
+    session.flush_rollout().await.expect("flush checkpoint");
+    let (items, _, _) = RolloutRecorder::load_rollout_items(&rollout_path)
+        .await
+        .expect("read persisted settings");
+    let snapshots = items
+        .into_iter()
+        .filter_map(|item| match item {
+            RolloutItem::EventMsg(EventMsg::ThreadSettingsApplied(event)) => {
+                Some((event.thread_id, event.thread_settings))
+            }
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+    assert_eq!(
+        snapshots,
+        vec![
+            (Some(session.thread_id), committed),
+            (Some(session.thread_id), restored),
+        ]
+    );
+}
+
 #[tokio::test]
 async fn session_settings_commit_keeps_snapshot_across_postcommit_wait() {
     let (session, _turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx(
@@ -6430,6 +6525,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
         tx_event,
         agent_status: agent_status_tx,
         state: Mutex::new(state),
+        thread_settings_persistence: Semaphore::new(/*permits*/ 1),
         managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1),
         features: config.features.clone(),
         windows_sandbox_proxy_settings_mode:
@@ -8709,6 +8805,7 @@ where
         tx_event,
         agent_status: agent_status_tx,
         state: Mutex::new(state),
+        thread_settings_persistence: Semaphore::new(/*permits*/ 1),
         managed_network_proxy_refresh_lock: Semaphore::new(/*permits*/ 1),
         features: config.features.clone(),
         windows_sandbox_proxy_settings_mode:

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

@@ -1,5 +1,5 @@
-//! Handles persistent thread-settings updates shared by standalone settings
-//! submissions and turn-input submission.
+//! Handles persistent thread-settings updates and serializes their persistence
+//! with compaction checkpoints.
 
 use super::session::Session;
 use super::session::SessionSettingsUpdate;
@@ -13,6 +13,7 @@ use codex_protocol::protocol::ThreadSettingsAppliedEvent;
 use codex_protocol::protocol::ThreadSettingsOverrides;
 use codex_protocol::protocol::ThreadSettingsSnapshot;
 use std::sync::Arc;
+use tokio::sync::SemaphorePermit;
 
 /// Applies standalone thread settings and reports invalid overrides through the
 /// normal event stream.
@@ -75,12 +76,22 @@ pub(super) fn prepare_update(overrides: ThreadSettingsOverrides) -> SessionSetti
     }
 }
 
+/// Acquires the shared permit before capturing or changing persistent settings.
+pub(super) async fn acquire_persistence_lock(session: &Session) -> SemaphorePermit<'_> {
+    session
+        .thread_settings_persistence
+        .acquire()
+        .await
+        .unwrap_or_else(|_| unreachable!("thread settings persistence semaphore is never closed"))
+}
+
 /// Applies persistent settings and emits the resulting thread-owned snapshot.
 pub(super) async fn apply_update(
     session: &Session,
     submission_id: String,
     updates: SessionSettingsUpdate,
 ) -> ConstraintResult<()> {
+    let _settings_guard = acquire_persistence_lock(session).await;
     let commit = session.update_settings(updates).await?;
     emit_applied(session, submission_id, commit.snapshot).await;
     Ok(())
@@ -93,6 +104,7 @@ pub(super) async fn emit_applied(
     snapshot: ThreadSettingsSnapshot,
 ) {
     let msg = EventMsg::ThreadSettingsApplied(ThreadSettingsAppliedEvent {
+        thread_id: Some(session.thread_id()),
         thread_settings: snapshot,
     });
     session
@@ -103,9 +115,10 @@ pub(super) async fn emit_applied(
         .await;
 }
 
-/// Builds the current thread-settings event for synthesized fork history.
+/// Builds a current thread-owned snapshot for fork and compaction persistence.
 pub(super) async fn applied_event(session: &Session) -> EventMsg {
     EventMsg::ThreadSettingsApplied(ThreadSettingsAppliedEvent {
+        thread_id: Some(session.thread_id()),
         thread_settings: session.thread_settings_snapshot().await,
     })
 }

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

@@ -129,6 +129,11 @@ impl PreparedTurnInputSettings {
             cyber_access_program,
         } = self.start_options;
         let emit_thread_settings_applied = self.thread_settings_update.is_some();
+        let _settings_guard = if emit_thread_settings_applied {
+            Some(thread_settings::acquire_persistence_lock(session).await)
+        } else {
+            None
+        };
         let mut updates = self.thread_settings_update.unwrap_or_default();
         updates.service_tier_for_turn = service_tier;