Persist agent messages as response items (#29829)

## Why

Inter-agent messages are recorded in live history as
`ResponseItem::AgentMessage`, but rollouts stored
`InterAgentCommunication` and rebuilt the response item during resume.
This made the rollout differ from the actual Responses history.

## What changed

- store the prepared `agent_message` response item directly
- keep `trigger_turn` in a small local metadata record for fork
truncation
- keep reading older `inter_agent_communication` rollout items

codex-rs/core/src/agent/control/spawn.rs

@@ -58,7 +58,8 @@ fn keep_forked_rollout_item(item: &RolloutItem, preserve_reference_context_item:
             | ResponseItem::ContextCompaction { .. }
             | ResponseItem::Other,
         ) => false,
-        RolloutItem::InterAgentCommunication(_) => false,
+        RolloutItem::InterAgentCommunication(_)
+        | RolloutItem::InterAgentCommunicationMetadata { .. } => false,
         // Full-history forks preserve the cached prompt prefix and can keep diffing
         // from the parent's durable baseline. Truncated forks drop part of that prompt,
         // so they must rebuild context on their first child turn.

codex-rs/core/src/agent/control_tests.rs

@@ -163,6 +163,7 @@ async fn persisted_originator(thread: &CodexThread) -> String {
             RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.originator.clone()),
             RolloutItem::ResponseItem(_)
             | RolloutItem::InterAgentCommunication(_)
+            | RolloutItem::InterAgentCommunicationMetadata { .. }
             | RolloutItem::EventMsg(_)
             | RolloutItem::Compacted(_)
             | RolloutItem::TurnContext(_) => None,

codex-rs/core/src/session/mod.rs

@@ -2841,7 +2841,7 @@ impl Session {
             std::slice::from_ref(&response_item),
         );
         let items = items.as_ref();
-        communication.id = items.first().and_then(ResponseItem::id).map(str::to_string);
+        let response_item = items[0].clone();
         {
             let mut state = self.state.lock().await;
             state.record_items(
@@ -2849,8 +2849,13 @@ impl Session {
                 turn_context.model_info.truncation_policy.into(),
             );
         }
-        self.persist_rollout_items(&[RolloutItem::InterAgentCommunication(communication)])
-            .await;
+        self.persist_rollout_items(&[
+            RolloutItem::InterAgentCommunicationMetadata {
+                trigger_turn: communication.trigger_turn,
+            },
+            RolloutItem::ResponseItem(response_item),
+        ])
+        .await;
         self.send_raw_response_items(turn_context, items).await;
     }
 

codex-rs/core/src/session/rollout_reconstruction.rs

@@ -264,6 +264,7 @@ impl Session {
                         active_segment.get_or_insert_with(ActiveReplaySegment::default);
                     active_segment.counts_as_user_turn = true;
                 }
+                RolloutItem::InterAgentCommunicationMetadata { .. } => {}
                 RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) => {}
             }
 
@@ -320,6 +321,7 @@ impl Session {
                         turn_context.model_info.truncation_policy.into(),
                     );
                 }
+                RolloutItem::InterAgentCommunicationMetadata { .. } => {}
                 RolloutItem::Compacted(compacted) => {
                     if let Some(replacement_history) = &compacted.replacement_history {
                         // This should actually never happen, because the reverse loop above (to build rollout_suffix)

codex-rs/core/src/session/tests.rs

@@ -1766,6 +1766,29 @@ async fn record_inter_agent_communication_sets_turn_id_in_rollout_and_resume() {
     else {
         panic!("expected resumed rollout history");
     };
+    let persisted_items = resumed
+        .history
+        .iter()
+        .filter(|item| {
+            matches!(
+                item,
+                RolloutItem::ResponseItem(_)
+                    | RolloutItem::InterAgentCommunication(_)
+                    | RolloutItem::InterAgentCommunicationMetadata { .. }
+            )
+        })
+        .cloned()
+        .collect::<Vec<_>>();
+    let expected_persisted_items = vec![
+        RolloutItem::InterAgentCommunicationMetadata {
+            trigger_turn: false,
+        },
+        RolloutItem::ResponseItem(expected_item.clone()),
+    ];
+    assert_eq!(
+        serde_json::to_value(persisted_items).unwrap(),
+        serde_json::to_value(expected_persisted_items).unwrap()
+    );
 
     let (resumed_session, _resumed_turn_context) = make_session_and_context().await;
     resumed_session
@@ -1818,14 +1841,11 @@ async fn record_inter_agent_communication_preserves_item_id_in_rollout_and_resum
     else {
         panic!("expected resumed rollout history");
     };
-    let persisted_communication = resumed.history.iter().find_map(|item| match item {
-        RolloutItem::InterAgentCommunication(communication) => Some(communication),
+    let persisted_item_id = resumed.history.iter().find_map(|item| match item {
+        RolloutItem::ResponseItem(item @ ResponseItem::AgentMessage { .. }) => item.id(),
         _ => None,
     });
-    assert_eq!(
-        persisted_communication.and_then(|communication| communication.id.as_deref()),
-        Some(live_item_id.as_str())
-    );
+    assert_eq!(persisted_item_id, Some(live_item_id.as_str()));
 
     let (resumed_session, _resumed_turn_context, _rx) =
         make_session_and_context_with_auth_and_config_and_rx(
@@ -2727,6 +2747,7 @@ async fn start_new_context_window_assigns_and_persists_item_ids() {
         RolloutItem::SessionMeta(_)
         | RolloutItem::ResponseItem(_)
         | RolloutItem::InterAgentCommunication(_)
+        | RolloutItem::InterAgentCommunicationMetadata { .. }
         | RolloutItem::TurnContext(_)
         | RolloutItem::EventMsg(_) => None,
     });
@@ -2783,6 +2804,7 @@ async fn record_initial_history_assigns_and_persists_id_for_forked_response_item
         RolloutItem::ResponseItem(response_item) => response_item.id(),
         RolloutItem::SessionMeta(_)
         | RolloutItem::InterAgentCommunication(_)
+        | RolloutItem::InterAgentCommunicationMetadata { .. }
         | RolloutItem::Compacted(_)
         | RolloutItem::TurnContext(_)
         | RolloutItem::EventMsg(_) => None,

codex-rs/core/src/thread_rollout_truncation.rs

@@ -72,7 +72,14 @@ pub(crate) fn fork_turn_positions_in_rollout(items: &[RolloutItem]) -> Vec<usize
     for (idx, item) in items.iter().enumerate() {
         match item {
             RolloutItem::ResponseItem(item) => {
-                if is_user_turn_boundary(item) {
+                let has_delivery_metadata = matches!(item, ResponseItem::AgentMessage { .. })
+                    && idx.checked_sub(1).is_some_and(|previous_idx| {
+                        matches!(
+                            items.get(previous_idx),
+                            Some(RolloutItem::InterAgentCommunicationMetadata { .. })
+                        )
+                    });
+                if is_user_turn_boundary(item) && !has_delivery_metadata {
                     rollback_turn_positions.push(idx);
                 }
                 if is_real_user_message_boundary(item) || is_trigger_turn_boundary(item) {
@@ -85,6 +92,12 @@ pub(crate) fn fork_turn_positions_in_rollout(items: &[RolloutItem]) -> Vec<usize
                     fork_turn_positions.push(idx);
                 }
             }
+            RolloutItem::InterAgentCommunicationMetadata { trigger_turn } => {
+                rollback_turn_positions.push(idx);
+                if *trigger_turn {
+                    fork_turn_positions.push(idx);
+                }
+            }
             RolloutItem::EventMsg(EventMsg::ThreadRolledBack(rollback)) => {
                 let num_turns = usize::try_from(rollback.num_turns).unwrap_or(usize::MAX);
                 if num_turns == 0 {

codex-rs/core/src/thread_rollout_truncation_tests.rs

@@ -241,6 +241,46 @@ fn fork_turn_positions_use_inter_agent_delivery_metadata() {
     assert_eq!(fork_turn_positions_in_rollout(&rollout), vec![0, 3, 5]);
 }
 
+#[test]
+fn fork_turn_positions_use_canonical_agent_messages_and_delivery_metadata() {
+    let queued = InterAgentCommunication::new(
+        AgentPath::root(),
+        AgentPath::try_from("/root/worker").expect("agent path"),
+        Vec::new(),
+        "queued during user turn".to_string(),
+        /*trigger_turn*/ false,
+    );
+    let triggered = InterAgentCommunication::new(
+        AgentPath::root(),
+        AgentPath::try_from("/root/worker").expect("agent path"),
+        Vec::new(),
+        "follow-up task".to_string(),
+        /*trigger_turn*/ true,
+    );
+    let mut rollout = vec![
+        RolloutItem::ResponseItem(user_msg("user task")),
+        RolloutItem::InterAgentCommunicationMetadata {
+            trigger_turn: false,
+        },
+        RolloutItem::ResponseItem(queued.to_model_input_item()),
+        RolloutItem::ResponseItem(assistant_msg("first answer")),
+        RolloutItem::InterAgentCommunicationMetadata { trigger_turn: true },
+        RolloutItem::ResponseItem(triggered.to_model_input_item()),
+        RolloutItem::ResponseItem(assistant_msg("second answer")),
+        RolloutItem::ResponseItem(user_msg("next user task")),
+    ];
+
+    assert_eq!(fork_turn_positions_in_rollout(&rollout), vec![0, 4, 7]);
+
+    rollout.insert(
+        7,
+        RolloutItem::EventMsg(EventMsg::ThreadRolledBack(ThreadRolledBackEvent {
+            num_turns: 1,
+        })),
+    );
+    assert_eq!(fork_turn_positions_in_rollout(&rollout), vec![0, 8]);
+}
+
 #[test]
 fn truncates_rollout_to_last_n_fork_turns_drops_startup_prefix_even_when_under_limit() {
     let rollout = vec![