diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 93eeda3c15..fcd274841a 100644 --- a/codex-rs/core/src/codex.rs +++ b/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, reference_context_item: Option, 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]) { diff --git a/codex-rs/core/src/codex_tests.rs b/codex-rs/core/src/codex_tests.rs index 34a1bf2328..1e0774f4ca 100644 --- a/codex-rs/core/src/codex_tests.rs +++ b/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; diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 2fb6fc01d3..1eb0f8e615 100644 --- a/codex-rs/core/src/compact.rs +++ b/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, 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 diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index ce1e368846..caf96d6273 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/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; diff --git a/codex-rs/core/src/compact_tests.rs b/codex-rs/core/src/compact_tests.rs index 8fd8dd689d..cd0182354b 100644 --- a/codex-rs/core/src/compact_tests.rs +++ b/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 {