codex: simplify compaction conflict handling (#14563)

Co-authored-by: Codex <noreply@openai.com>

codex-rs/core/src/codex.rs

@@ -3362,18 +3362,17 @@ impl Session {
     /// append-only concurrent `GhostSnapshot` tail seen under the same state
     /// lock.
     ///
-    /// Returns `true` when live history was unchanged or when the only
-    /// concurrent writes were appended `GhostSnapshot` items that were merged
-    /// into `items` before replacement. Returns `false` when live history
-    /// diverged in any other way, in which case replacement still happens but
-    /// no concurrent tail is merged.
+    /// When live history diverges in any other way, replacement still happens,
+    /// concurrent writes are dropped, and a warning is emitted instead of
+    /// propagating that detail through the return value.
     pub(crate) async fn replace_compacted_history(
         &self,
         mut items: Vec<ResponseItem>,
         reference_context_item: Option<TurnContextItem>,
         mut compacted_item: CompactedItem,
         base_history: &[ResponseItem],
-    ) -> bool {
+        turn_id: &str,
+    ) {
         // Compaction snapshots history and waits on a model/API call. Preserve
         // any append-only ghost snapshot tail while holding the same lock that
         // replaces history so detached `/undo` metadata writes are not lost in
@@ -3396,7 +3395,12 @@ impl Session {
             self.persist_rollout_items(&[RolloutItem::TurnContext(turn_context_item)])
                 .await;
         }
-        merged_ghost_snapshot_tail
+        if !merged_ghost_snapshot_tail {
+            warn!(
+                turn_id,
+                "session history changed beyond append-only ghost snapshots during compaction; dropping concurrent writes"
+            );
+        }
     }
 
     async fn persist_rollout_response_items(&self, items: &[ResponseItem]) {

codex-rs/core/src/codex_tests.rs

@@ -799,19 +799,18 @@ async fn replace_compacted_history_persists_merged_ghost_snapshot_tail() {
     sess.record_conversation_items(tc.as_ref(), std::slice::from_ref(&ghost_snapshot))
         .await;
 
-    let merged = sess
-        .replace_compacted_history(
-            vec![summary_item.clone()],
-            Some(tc.to_turn_context_item()),
-            CompactedItem {
-                message: String::new(),
-                replacement_history: Some(vec![summary_item.clone()]),
-            },
-            &base_history,
-        )
-        .await;
+    sess.replace_compacted_history(
+        vec![summary_item.clone()],
+        Some(tc.to_turn_context_item()),
+        CompactedItem {
+            message: String::new(),
+            replacement_history: Some(vec![summary_item.clone()]),
+        },
+        &base_history,
+        &tc.sub_id,
+    )
+    .await;
 
-    assert!(merged);
     let expected_history = vec![summary_item.clone(), ghost_snapshot.clone()];
     assert_eq!(sess.clone_history().await.raw_items(), expected_history);
 

codex-rs/core/src/compact.rs

@@ -27,7 +27,6 @@ use codex_protocol::models::ResponseItem;
 use codex_protocol::user_input::UserInput;
 use futures::prelude::*;
 use tracing::error;
-use tracing::warn;
 
 pub const SUMMARIZATION_PROMPT: &str = include_str!("../templates/compact/prompt.md");
 pub const SUMMARY_PREFIX: &str = include_str!("../templates/compact/summary_prefix.md");
@@ -219,20 +218,14 @@ async fn run_compact_task_inner(
         message: summary_text.clone(),
         replacement_history: Some(new_history.clone()),
     };
-    if !sess
-        .replace_compacted_history(
-            new_history,
-            reference_context_item,
-            compacted_item,
-            history_items,
-        )
-        .await
-    {
-        warn!(
-            turn_id = %turn_context.sub_id,
-            "session history changed beyond append-only ghost snapshots during compaction; skipping concurrent ghost snapshot merge"
-        );
-    }
+    sess.replace_compacted_history(
+        new_history,
+        reference_context_item,
+        compacted_item,
+        history_items,
+        &turn_context.sub_id,
+    )
+    .await;
     sess.recompute_token_usage(&turn_context).await;
 
     sess.emit_turn_item_completed(&turn_context, compaction_item)

codex-rs/core/src/compact_remote.rs

@@ -24,7 +24,6 @@ use futures::TryFutureExt;
 use tokio_util::sync::CancellationToken;
 use tracing::error;
 use tracing::info;
-use tracing::warn;
 
 pub(crate) async fn run_inline_remote_auto_compact_task(
     sess: Arc<Session>,
@@ -158,20 +157,14 @@ async fn run_remote_compact_task_inner_impl(
         message: String::new(),
         replacement_history: Some(new_history.clone()),
     };
-    if !sess
-        .replace_compacted_history(
-            new_history,
-            reference_context_item,
-            compacted_item,
-            history_snapshot.raw_items(),
-        )
-        .await
-    {
-        warn!(
-            turn_id = %turn_context.sub_id,
-            "session history changed beyond append-only ghost snapshots during remote compaction; skipping concurrent ghost snapshot merge"
-        );
-    }
+    sess.replace_compacted_history(
+        new_history,
+        reference_context_item,
+        compacted_item,
+        history_snapshot.raw_items(),
+        &turn_context.sub_id,
+    )
+    .await;
     sess.recompute_token_usage(turn_context).await;
 
     sess.emit_turn_item_completed(turn_context, compaction_item)