Add Compact and Turn Context to the rollout items (#3444)

Adding compact and turn context to the rollout items

based on #3440

codex-rs/core/src/codex.rs

@@ -19,11 +19,13 @@ use codex_apply_patch::ApplyPatchAction;
 use codex_apply_patch::MaybeApplyPatchVerified;
 use codex_apply_patch::maybe_parse_apply_patch_verified;
 use codex_protocol::mcp_protocol::ConversationId;
+use codex_protocol::protocol::CompactedItem;
 use codex_protocol::protocol::ConversationPathResponseEvent;
 use codex_protocol::protocol::RolloutItem;
 use codex_protocol::protocol::TaskStartedEvent;
 use codex_protocol::protocol::TurnAbortReason;
 use codex_protocol::protocol::TurnAbortedEvent;
+use codex_protocol::protocol::TurnContextItem;
 use futures::prelude::*;
 use mcp_types::CallToolResult;
 use serde::Deserialize;
@@ -1786,6 +1788,15 @@ async fn try_run_turn(
         })
     };
 
+    let rollout_item = RolloutItem::TurnContext(TurnContextItem {
+        cwd: turn_context.cwd.clone(),
+        approval_policy: turn_context.approval_policy,
+        sandbox_policy: turn_context.sandbox_policy.clone(),
+        model: turn_context.client.get_model().clone(),
+        effort: turn_context.client.get_reasoning_effort(),
+        summary: turn_context.client.get_reasoning_summary(),
+    });
+    sess.persist_rollout_items(&[rollout_item]).await;
     let mut stream = turn_context.client.clone().stream(&prompt).await?;
 
     let mut output = Vec::new();
@@ -1968,10 +1979,14 @@ async fn run_compact_task(
 
     sess.remove_task(&sub_id);
 
-    {
+    let rollout_item = {
         let mut state = sess.state.lock_unchecked();
         state.history.keep_last_messages(1);
-    }
+        RolloutItem::Compacted(CompactedItem {
+            message: state.history.last_agent_message(),
+        })
+    };
+    sess.persist_rollout_items(&[rollout_item]).await;
 
     let event = Event {
         id: sub_id.clone(),
@@ -2997,6 +3012,15 @@ async fn drain_to_completed(
     sub_id: &str,
     prompt: &Prompt,
 ) -> CodexResult<()> {
+    let rollout_item = RolloutItem::TurnContext(TurnContextItem {
+        cwd: turn_context.cwd.clone(),
+        approval_policy: turn_context.approval_policy,
+        sandbox_policy: turn_context.sandbox_policy.clone(),
+        model: turn_context.client.get_model(),
+        effort: turn_context.client.get_reasoning_effort(),
+        summary: turn_context.client.get_reasoning_summary(),
+    });
+    sess.persist_rollout_items(&[rollout_item]).await;
     let mut stream = turn_context.client.clone().stream(prompt).await?;
     loop {
         let maybe_event = stream.next().await;

codex-rs/core/src/conversation_history.rs

@@ -1,3 +1,4 @@
+use codex_protocol::models::ContentItem;
 use codex_protocol::models::ResponseItem;
 
 /// Transcript of conversation history
@@ -59,6 +60,26 @@ impl ConversationHistory {
         kept.reverse();
         self.items = kept;
     }
+
+    pub(crate) fn last_agent_message(&self) -> String {
+        for item in self.items.iter().rev() {
+            if let ResponseItem::Message { role, content, .. } = item
+                && role == "assistant"
+            {
+                return content
+                    .iter()
+                    .find_map(|ci| {
+                        if let ContentItem::OutputText { text } = ci {
+                            Some(text.clone())
+                        } else {
+                            None
+                        }
+                    })
+                    .unwrap_or_default();
+            }
+        }
+        String::new()
+    }
 }
 
 /// Anything that is not a system message or "reasoning" message is considered

codex-rs/core/src/rollout/list.rs

@@ -318,6 +318,12 @@ async fn read_head_and_flags(
                     head.push(val);
                 }
             }
+            RolloutItem::TurnContext(_) => {
+                // Not included in `head`; skip.
+            }
+            RolloutItem::Compacted(_) => {
+                // Not included in `head`; skip.
+            }
             RolloutItem::EventMsg(ev) => {
                 if matches!(ev, EventMsg::UserMessage(_)) {
                     saw_user_event = true;

codex-rs/core/src/rollout/policy.rs

@@ -8,8 +8,10 @@ pub(crate) fn is_persisted_response_item(item: &RolloutItem) -> bool {
     match item {
         RolloutItem::ResponseItem(item) => should_persist_response_item(item),
         RolloutItem::EventMsg(ev) => should_persist_event_msg(ev),
-        // Always persist session meta
-        RolloutItem::SessionMeta(_) => true,
+        // Persist Codex executive markers so we can analyze flows (e.g., compaction, API turns).
+        RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => {
+            true
+        }
     }
 }
 

codex-rs/core/src/rollout/recorder.rs

@@ -238,6 +238,12 @@ impl RolloutRecorder {
                     RolloutItem::ResponseItem(item) => {
                         items.push(RolloutItem::ResponseItem(item));
                     }
+                    RolloutItem::Compacted(item) => {
+                        items.push(RolloutItem::Compacted(item));
+                    }
+                    RolloutItem::TurnContext(item) => {
+                        items.push(RolloutItem::TurnContext(item));
+                    }
                     RolloutItem::EventMsg(_ev) => {
                         items.push(RolloutItem::EventMsg(_ev));
                     }