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

@@ -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)

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)

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)

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<ResponseItem>,
         reference_context_item: Option<TurnContextItem>,
         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());

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_")));

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)]