Normalize compacted user message annotations (#40273)

## Why

Local compaction rebuilds preserved user input as a single text item. Carrying
over the original per-item content kinds can leave its annotation metadata out
of sync with the rebuilt content.

## What changed

- Set the rebuilt item's content kind to `user.text` when the original message
  had content-kind metadata.
- Preserve the remaining message and harness metadata.

## Testing

- Cover metadata normalization in compacted history.
- Verify persisted replacement history contains the merged text and a matching
  `user.text` content kind.

GitOrigin-RevId: cd10b8ec68ef8ec2a272a90182e69875b60dc35c

codex-rs/core/src/compact.rs

@@ -28,6 +28,8 @@ use codex_analytics::CompactionStatus;
 use codex_analytics::CompactionStrategy;
 use codex_analytics::CompactionTrigger;
 use codex_analytics::now_unix_seconds;
+use codex_context_fragments::AnnotatedContent;
+use codex_context_fragments::set_annotated_content;
 use codex_history::CodexHarnessMetadata;
 use codex_history::ResponseItemEnvelope;
 use codex_protocol::error::CodexErr;
@@ -37,6 +39,7 @@ use codex_protocol::items::ContextCompactionItem;
 use codex_protocol::items::TurnItem;
 use codex_protocol::models::AgentMessageInputContent;
 use codex_protocol::models::ContentItem;
+use codex_protocol::models::ContentItemKind;
 use codex_protocol::models::InternalChatMessageMetadataPassthrough;
 use codex_protocol::models::ResponseInputItem;
 use codex_protocol::models::ResponseItem;
@@ -683,18 +686,33 @@ fn build_compacted_history_with_limit(
     }
 
     for message in &selected_messages {
+        let mut item = ResponseItem::Message {
+            id: None,
+            role: "user".to_string(),
+            content: vec![ContentItem::InputText {
+                text: message.message.clone(),
+            }],
+            phase: None,
+            internal_chat_message_metadata_passthrough: message
+                .internal_chat_message_metadata_passthrough
+                .clone(),
+        };
+        if message
+            .internal_chat_message_metadata_passthrough
+            .as_ref()
+            .and_then(|metadata| metadata.content_item_kinds.as_ref())
+            .is_some()
+        {
+            let _ = set_annotated_content(
+                &mut item,
+                vec![AnnotatedContent::input_text(
+                    &message.message,
+                    ContentItemKind("user.text".to_string()),
+                )],
+            );
+        }
         history.push(ResponseItemEnvelope {
-            item: ResponseItem::Message {
-                id: None,
-                role: "user".to_string(),
-                content: vec![ContentItem::InputText {
-                    text: message.message.clone(),
-                }],
-                phase: None,
-                internal_chat_message_metadata_passthrough: message
-                    .internal_chat_message_metadata_passthrough
-                    .clone(),
-            },
+            item,
             metadata: message.harness_metadata.clone(),
         });
     }

codex-rs/core/src/compact_tests.rs

@@ -2,6 +2,7 @@ use super::*;
 use codex_history::CodexHarnessMetadata;
 use codex_history::ResponseItemEnvelope;
 use codex_protocol::ResponseItemId;
+use codex_protocol::models::ContentItemKind;
 use codex_protocol::models::DEFAULT_IMAGE_DETAIL;
 use codex_protocol::models::InternalChatMessageMetadataPassthrough;
 use pretty_assertions::assert_eq;
@@ -295,6 +296,11 @@ fn build_compacted_history_preserves_user_message_passthrough_metadata() {
             internal_chat_message_metadata_passthrough: Some(
                 InternalChatMessageMetadataPassthrough {
                     turn_id: Some("turn-1".to_string()),
+                    content_item_kinds: Some(vec![
+                        ContentItemKind("user.image".to_string()),
+                        ContentItemKind("user.text".to_string()),
+                        ContentItemKind("user.audio".to_string()),
+                    ]),
                     ..Default::default()
                 },
             ),
@@ -303,10 +309,32 @@ fn build_compacted_history_preserves_user_message_passthrough_metadata() {
         "summary text",
     );
 
-    assert_eq!(history[0].turn_id(), Some("turn-1"));
-    assert_eq!(history[1].turn_id(), None);
-    assert_eq!(history[0].metadata, Some(CodexHarnessMetadata::default()));
-    assert_eq!(history[1].metadata, None);
+    assert_eq!(
+        history,
+        vec![
+            ResponseItemEnvelope {
+                item: ResponseItem::Message {
+                    id: None,
+                    role: "user".to_string(),
+                    content: vec![ContentItem::InputText {
+                        text: "first user message".to_string(),
+                    }],
+                    phase: None,
+                    internal_chat_message_metadata_passthrough: Some(
+                        InternalChatMessageMetadataPassthrough {
+                            turn_id: Some("turn-1".to_string()),
+                            content_item_kinds: Some(vec![ContentItemKind(
+                                "user.text".to_string()
+                            )]),
+                            ..Default::default()
+                        },
+                    ),
+                },
+                metadata: Some(CodexHarnessMetadata::default()),
+            },
+            ResponseItemEnvelope::new(user_message("summary text")),
+        ]
+    );
 }
 
 #[tokio::test]

codex-rs/core/tests/suite/compact.rs

@@ -542,10 +542,16 @@ async fn summarize_context_three_requests_and_instructions() {
 
     // 1) Normal user input – should hit server once.
     codex
-        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
-            text: "hello world".into(),
-            text_elements: Vec::new(),
-        }]))
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![
+            UserInput::Text {
+                text: "hello world".into(),
+                text_elements: Vec::new(),
+            },
+            UserInput::Text {
+                text: " second fragment".into(),
+                text_elements: Vec::new(),
+            },
+        ]))
         .await
         .unwrap();
     wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await;
@@ -644,7 +650,7 @@ async fn summarize_context_three_requests_and_instructions() {
     assert!(
         messages
             .iter()
-            .any(|(r, t)| r == "user" && t == "hello world"),
+            .any(|(r, t)| r == "user" && t == "hello world second fragment"),
         "third request should include the original user message"
     );
     assert!(
@@ -664,6 +670,28 @@ async fn summarize_context_three_requests_and_instructions() {
     codex.submit(Op::Shutdown).await.unwrap();
     wait_for_event(&codex, |ev| matches!(ev, EventMsg::ShutdownComplete)).await;
 
+    let replacement_history = replacement_history_from_rollout(&rollout_path)
+        .expect("local compaction should persist replacement history");
+    let compacted_user_message = replacement_history
+        .iter()
+        .find(|item| item["content"][0]["text"] == "hello world second fragment")
+        .expect("persisted replacement history should contain the compacted user message");
+    assert_eq!(
+        json!({
+            "type": compacted_user_message["type"],
+            "role": compacted_user_message["role"],
+            "content": compacted_user_message["content"],
+            "content_item_kinds": compacted_user_message
+                ["internal_chat_message_metadata_passthrough"]["content_item_kinds"],
+        }),
+        json!({
+            "type": "message",
+            "role": "user",
+            "content": [{"type": "input_text", "text": "hello world second fragment"}],
+            "content_item_kinds": ["user.text"],
+        }),
+    );
+
     // Verify rollout contains user-turn TurnContext entries and a Compacted entry.
     println!("rollout path: {}", rollout_path.display());
     let text = std::fs::read_to_string(&rollout_path).expect("failed to read rollout file");