codex: simplify compaction ghost snapshot merge

Narrow the compaction race fix to the concrete ghost-snapshot case.
Remove the conflict classification and warning path, and make
replace_compacted_history opportunistically preserve only an
append-only GhostSnapshot tail under the state lock.

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

codex-rs/core/src/codex.rs

@@ -3358,34 +3358,29 @@ impl Session {
         state.replace_history(items, reference_context_item);
     }
 
-    /// Replaces session history with a compacted transcript and preserves any
-    /// append-only concurrent `GhostSnapshot` tail seen under the same state
-    /// lock.
-    ///
-    /// When live history diverges in any other way, replacement still happens,
-    /// concurrent writes are dropped, and a warning is emitted.
+    /// Replaces session history with a compacted transcript and, under the same
+    /// state lock, preserves a concurrent append-only `GhostSnapshot` tail when
+    /// the live history still extends the compacted prefix.
     pub(crate) async fn replace_compacted_history(
         &self,
         mut items: Vec<ResponseItem>,
         reference_context_item: Option<TurnContextItem>,
         mut compacted_item: CompactedItem,
         base_history: &[ResponseItem],
-        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
         // the gap before replacement.
-        let merged_ghost_snapshot_tail = {
+        {
             let mut state = self.state.lock().await;
-            let merged = compact::append_concurrent_ghost_snapshot_tail_if_append_only(
+            compact::append_concurrent_ghost_snapshot_tail(
                 &mut items,
                 base_history,
                 state.history.raw_items(),
             );
             state.replace_history(items.clone(), reference_context_item.clone());
-            merged
-        };
+        }
         compacted_item.replacement_history = Some(items);
 
         self.persist_rollout_items(&[RolloutItem::Compacted(compacted_item)])
@@ -3394,12 +3389,6 @@ impl Session {
             self.persist_rollout_items(&[RolloutItem::TurnContext(turn_context_item)])
                 .await;
         }
-        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

@@ -807,7 +807,6 @@ async fn replace_compacted_history_persists_merged_ghost_snapshot_tail() {
             replacement_history: Some(vec![summary_item.clone()]),
         },
         &base_history,
-        &tc.sub_id,
     )
     .await;
 

codex-rs/core/src/compact.rs

@@ -223,7 +223,6 @@ async fn run_compact_task_inner(
         reference_context_item,
         compacted_item,
         history_items,
-        &turn_context.sub_id,
     )
     .await;
     sess.recompute_token_usage(&turn_context).await;
@@ -276,41 +275,28 @@ pub(crate) fn is_summary_message(message: &str) -> bool {
     message.starts_with(format!("{SUMMARY_PREFIX}\n").as_str())
 }
 
-/// Appends ghost snapshots added after `base_history` only when
-/// `latest_history` still preserves `base_history` as an exact prefix.
+/// Appends ghost snapshots added after `base_history` when `latest_history`
+/// still preserves `base_history` as an exact prefix and only appends
+/// `GhostSnapshot` items.
 ///
 /// Ghost snapshots are appended by detached background tasks for `/undo`, but
 /// `ContextManager::for_prompt()` strips them before any model request. That
-/// makes it safe to preserve a concurrent append-only ghost-snapshot tail while
+/// makes it safe to preserve the concurrent ghost-snapshot-only case while
 /// avoiding model-visible items that the compaction request never saw.
-///
-/// Returns `true` when no concurrent history change occurred or when the newer
-/// history differs only by append-only ghost snapshots, in which case that tail
-/// is appended to `new_history`. Returns `false` when concurrent history
-/// mutation rewrote or removed earlier items, or appended any non-ghost item,
-/// since this helper cannot safely merge those shapes.
-pub(crate) fn append_concurrent_ghost_snapshot_tail_if_append_only(
+pub(crate) fn append_concurrent_ghost_snapshot_tail(
     new_history: &mut Vec<ResponseItem>,
     base_history: &[ResponseItem],
     latest_history: &[ResponseItem],
-) -> bool {
-    if latest_history == base_history {
-        return true;
-    }
-    if !latest_history.starts_with(base_history) {
-        return false;
-    }
-
-    let appended_items = &latest_history[base_history.len()..];
-    if !appended_items
+) {
+    let Some(appended_items) = latest_history.strip_prefix(base_history) else {
+        return;
+    };
+    if appended_items
         .iter()
         .all(|item| matches!(item, ResponseItem::GhostSnapshot { .. }))
     {
-        return false;
+        new_history.extend_from_slice(appended_items);
     }
-
-    new_history.extend_from_slice(appended_items);
-    true
 }
 
 /// Inserts canonical initial context into compacted replacement history at the

codex-rs/core/src/compact_remote.rs

@@ -162,7 +162,6 @@ async fn run_remote_compact_task_inner_impl(
         reference_context_item,
         compacted_item,
         history_snapshot.raw_items(),
-        &turn_context.sub_id,
     )
     .await;
     sess.recompute_token_usage(turn_context).await;

codex-rs/core/src/compact_tests.rs

@@ -186,7 +186,7 @@ fn build_token_limited_compacted_history_appends_summary_message() {
 }
 
 #[test]
-fn append_concurrent_ghost_snapshot_tail_if_append_only_appends_concurrent_tail() {
+fn append_concurrent_ghost_snapshot_tail_appends_concurrent_tail() {
     let base_history = vec![ResponseItem::Message {
         id: None,
         role: "user".to_string(),
@@ -215,13 +215,7 @@ fn append_concurrent_ghost_snapshot_tail_if_append_only_appends_concurrent_tail(
         phase: None,
     }];
 
-    let merged = append_concurrent_ghost_snapshot_tail_if_append_only(
-        &mut compacted_history,
-        &base_history,
-        &latest_history,
-    );
-
-    assert!(merged);
+    append_concurrent_ghost_snapshot_tail(&mut compacted_history, &base_history, &latest_history);
     assert_eq!(
         compacted_history,
         vec![
@@ -240,7 +234,7 @@ fn append_concurrent_ghost_snapshot_tail_if_append_only_appends_concurrent_tail(
 }
 
 #[test]
-fn append_concurrent_ghost_snapshot_tail_if_append_only_rejects_model_visible_tail() {
+fn append_concurrent_ghost_snapshot_tail_ignores_model_visible_tail() {
     let base_history = vec![ResponseItem::Message {
         id: None,
         role: "user".to_string(),
@@ -272,13 +266,7 @@ fn append_concurrent_ghost_snapshot_tail_if_append_only_rejects_model_visible_ta
         phase: None,
     }];
 
-    let merged = append_concurrent_ghost_snapshot_tail_if_append_only(
-        &mut compacted_history,
-        &base_history,
-        &latest_history,
-    );
-
-    assert!(!merged);
+    append_concurrent_ghost_snapshot_tail(&mut compacted_history, &base_history, &latest_history);
     assert_eq!(
         compacted_history,
         vec![ResponseItem::Message {
@@ -294,7 +282,7 @@ fn append_concurrent_ghost_snapshot_tail_if_append_only_rejects_model_visible_ta
 }
 
 #[test]
-fn append_concurrent_ghost_snapshot_tail_if_append_only_rejects_non_append_only_changes() {
+fn append_concurrent_ghost_snapshot_tail_ignores_non_append_only_changes() {
     let base_history = vec![ResponseItem::Message {
         id: None,
         role: "user".to_string(),
@@ -323,13 +311,7 @@ fn append_concurrent_ghost_snapshot_tail_if_append_only_rejects_non_append_only_
         phase: None,
     }];
 
-    let merged = append_concurrent_ghost_snapshot_tail_if_append_only(
-        &mut compacted_history,
-        &base_history,
-        &latest_history,
-    );
-
-    assert!(!merged);
+    append_concurrent_ghost_snapshot_tail(&mut compacted_history, &base_history, &latest_history);
     assert_eq!(
         compacted_history,
         vec![ResponseItem::Message {