From 70a6aa2634adedaabf66a66f11c16e869c1c7d1e Mon Sep 17 00:00:00 2001 From: pakrym-oai Date: Thu, 18 Jun 2026 18:49:57 -0700 Subject: [PATCH] core: assign item IDs to compacted replacement history (#29012) ## Why Remote v2 compaction can return replacement-history items without IDs. Because replacement history is installed directly, those items bypass normal history preparation and remain ID-less in later Responses requests even when the `item_ids` feature is enabled. ## What changed - Pass the active `TurnContext` into `replace_compacted_history`. - When `item_ids` is enabled, assign missing IDs before installing and persisting replacement history. - Rebuild `CompactedItem` from the prepared history so live and persisted replacement histories match. - Add integration coverage requiring IDs on every ID-capable input item in the initial, remote v2 compaction, and post-compaction requests. ## Test plan - `just test -p codex-core response_item_ids` - `just test -p codex-core websocket_v2_test_codex_shell_chain` - `just test -p codex-core remote_compaction_parity_pre_turn_auto` - `just test -p codex-app-server thread_inject_items_adds_raw_response_items_to_thread_history` --- codex-rs/core/src/compact.rs | 9 +++- codex-rs/core/src/compact_remote.rs | 9 +++- codex-rs/core/src/compact_remote_v2.rs | 9 +++- codex-rs/core/src/session/mod.rs | 21 ++++++++-- codex-rs/core/src/session/tests.rs | 4 +- codex-rs/core/tests/suite/client.rs | 57 ++++++++++++++++++++++++++ 6 files changed, 97 insertions(+), 12 deletions(-) diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index d7d2f57bf..7b93195de 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -322,8 +322,13 @@ async fn run_compact_task_inner_impl( window_number: Some(window_number), window_id: Some(window_id), }; - sess.replace_compacted_history(new_history, reference_context_item, compacted_item) - .await; + sess.replace_compacted_history( + turn_context.as_ref(), + new_history, + reference_context_item, + compacted_item, + ) + .await; sess.recompute_token_usage(&turn_context).await; sess.emit_turn_item_completed(&turn_context, compaction_item) diff --git a/codex-rs/core/src/compact_remote.rs b/codex-rs/core/src/compact_remote.rs index 1eebc2317..26469c201 100644 --- a/codex-rs/core/src/compact_remote.rs +++ b/codex-rs/core/src/compact_remote.rs @@ -284,8 +284,13 @@ async fn run_remote_compact_task_inner_impl( input_history: &trace_input_history, replacement_history: &new_history, }); - sess.replace_compacted_history(new_history, reference_context_item, compacted_item) - .await; + sess.replace_compacted_history( + turn_context.as_ref(), + new_history, + reference_context_item, + compacted_item, + ) + .await; sess.recompute_token_usage(turn_context).await; sess.emit_turn_item_completed(turn_context, compaction_item) diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 8a6026bb8..d941388b0 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -312,8 +312,13 @@ async fn run_remote_compact_task_inner_impl( input_history: &trace_input_history, replacement_history: &new_history, }); - sess.replace_compacted_history(new_history, reference_context_item, compacted_item) - .await; + sess.replace_compacted_history( + turn_context.as_ref(), + new_history, + reference_context_item, + compacted_item, + ) + .await; sess.recompute_token_usage(turn_context).await; sess.emit_turn_item_completed(turn_context, compaction_item) diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 7fd3b9540..fbd5a61bc 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -2673,15 +2673,17 @@ impl Session { prepare_response_items(items.to_mut()); } if turn_context.config.features.enabled(Feature::ItemIds) { - Self::assign_missing_response_item_ids(&mut items); + Self::assign_missing_response_item_ids(items) + } else { + items } - items } - fn assign_missing_response_item_ids(items: &mut Cow<'_, [ResponseItem]>) { + fn assign_missing_response_item_ids(items: Cow<'_, [ResponseItem]>) -> Cow<'_, [ResponseItem]> { if items.iter().all(|item| item.id().is_some()) { - return; + return items; } + let mut items = items; for item in items.to_mut() { if item.id().is_some() { continue; @@ -2705,6 +2707,7 @@ impl Session { }; item.set_id(Some(format!("{prefix}_{}", Uuid::now_v7()))); } + items } pub(crate) fn response_item_from_user_input( @@ -2840,10 +2843,20 @@ impl Session { pub(crate) async fn replace_compacted_history( &self, + turn_context: &TurnContext, items: Vec, reference_context_item: Option, compacted_item: CompactedItem, ) { + let items = if turn_context.config.features.enabled(Feature::ItemIds) { + Self::assign_missing_response_item_ids(Cow::Owned(items)).into_owned() + } else { + items + }; + let compacted_item = CompactedItem { + replacement_history: Some(items.clone()), + ..compacted_item + }; { let mut state = self.state.lock().await; state.replace_history(items, reference_context_item.clone()); diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 8fec6f97d..436b85907 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -201,7 +201,7 @@ fn user_message(text: &str) -> ResponseItem { #[test] fn assign_missing_response_item_ids_skips_agent_messages() { - let mut items = Cow::Owned(vec![ + let items = Cow::Owned(vec![ ResponseItem::AgentMessage { id: None, author: "worker".to_string(), @@ -214,7 +214,7 @@ fn assign_missing_response_item_ids_skips_agent_messages() { user_message("hello"), ]); - Session::assign_missing_response_item_ids(&mut items); + let items = Session::assign_missing_response_item_ids(items); assert_eq!(items[0].id(), None); assert!(items[1].id().is_some_and(|id| id.starts_with("msg_"))); diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index dff440a30..b88960f95 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -300,6 +300,63 @@ async fn response_item_ids_persist_across_resume_and_preserve_server_ids() -> an Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn response_item_ids_are_sent_for_all_remote_v2_compaction_requests() -> anyhow::Result<()> { + let server = MockServer::start().await; + let response_mock = mount_sse_sequence( + &server, + vec![ + sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), + sse(vec![ + json!({ + "type": "response.output_item.done", + "item": { + "type": "compaction", + "encrypted_content": "ENCRYPTED_CONTEXT_COMPACTION_SUMMARY", + } + }), + ev_completed("resp-compact"), + ]), + sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]), + ], + ) + .await; + let test = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config(|config| { + let _ = config.features.enable(Feature::ItemIds); + let _ = config.features.enable(Feature::RemoteCompactionV2); + }) + .build(&server) + .await?; + + test.submit_turn("before compaction").await?; + test.codex.submit(Op::Compact).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + test.submit_turn("after compaction").await?; + + let requests = response_mock.requests(); + assert_eq!(requests.len(), 3); + for (request_index, request) in requests.iter().enumerate() { + let input = request.input(); + assert!(!input.is_empty(), "request {request_index} input is empty"); + for item in input { + if item.get("type").and_then(serde_json::Value::as_str) == Some("compaction_trigger") { + continue; + } + assert!( + item.get("id").and_then(serde_json::Value::as_str).is_some(), + "request {request_index} item should have an ID: {item:#?}" + ); + } + } + + Ok(()) +} + /// Writes an `auth.json` into the provided `codex_home` with the specified parameters. /// Returns the fake JWT string written to `tokens.id_token`. #[expect(clippy::unwrap_used)]