progress

codex-rs/core/src/codex.rs

@@ -592,22 +592,12 @@ impl Session {
         let mut msgs = Vec::new();
         for item in items.clone() {
             match item {
-<<<<<<< HEAD
-                RolloutItem::ResponseItem(ref v) => {
-                    responses.extend(v.clone());
-                    let new_msgs: Vec<EventMsg> = v
-                        .iter()
-                        .flat_map(|ri| {
-                            map_response_item_to_event_messages(ri, self.show_raw_agent_reasoning)
-                        })
-                        .collect();
-=======
-                RolloutItem::ResponseItem(response) => {
+                RolloutItem::ResponseItem(ref response) => {
+                    responses.push(response.clone());
                     let new_msgs: Vec<EventMsg> = map_response_item_to_event_messages(
-                        &response,
+                        response,
                         self.show_raw_agent_reasoning,
                     );
->>>>>>> 4245659e (progress)
                     if before_resume_session {
                         msgs.extend(new_msgs);
                     } else {
@@ -618,15 +608,8 @@ impl Session {
                         );
                     }
                 }
-<<<<<<< HEAD
-                RolloutItem::Event(events) => msgs.extend(events.iter().map(|e| e.msg.clone())),
-                RolloutItem::SessionMeta(..) => {}
-=======
                 RolloutItem::Event(event) => msgs.push(event.msg.clone()),
-                RolloutItem::SessionMeta(..) => {
-                    // Session meta does not emit events
-                }
->>>>>>> 4245659e (progress)
+                RolloutItem::SessionMeta(..) => {}
             }
         }
 
@@ -742,7 +725,6 @@ impl Session {
 
     /// Records items to both the rollout and the chat completions/ZDR
     /// transcript, if enabled.
-<<<<<<< HEAD
     async fn record_conversation_items(&self, items: &[ResponseItem]) {
         self.record_conversation_items_internal(items, true).await;
     }
@@ -752,20 +734,6 @@ impl Session {
         if persist {
             self.record_state_snapshot(RolloutItem::ResponseItem(items.to_vec()))
                 .await;
-=======
-    async fn record_conversation_items<T>(&self, item: T)
-    where
-        T: Into<RolloutItem>,
-    {
-        let item: RolloutItem = item.into();
-        debug!("Recording items for conversation: {item:?}");
-        self.record_state_snapshot(item.clone()).await;
-        if let RolloutItem::ResponseItem(response_item) = &item {
-            self.state
-                .lock_unchecked()
-                .history
-                .record_items(std::slice::from_ref(response_item));
->>>>>>> 4245659e (progress)
         }
 
         self.state.lock_unchecked().history.record_items(items);

codex-rs/core/src/conversation_manager.rs

@@ -263,6 +263,8 @@ mod tests {
             assistant_msg("a4"),
         ];
 
+        let items: Vec<RolloutItem> = items.into_iter().map(RolloutItem::from).collect();
+
         let truncated = truncate_after_dropping_last_messages(items.clone(), 1);
         if let InitialHistory::Resumed(resumed) = truncated {
             let get_text = |ri: &ResponseItem| -> Option<String> {
@@ -279,7 +281,7 @@ mod tests {
             let texts: Vec<String> = resumed
                 .iter()
                 .filter_map(|it| match it {
-                    RolloutItem::ResponseItem(v) if !v.is_empty() => get_text(&v[0]),
+                    RolloutItem::ResponseItem(ri) => get_text(ri),
                     _ => None,
                 })
                 .collect();

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

@@ -10,6 +10,7 @@ use time::macros::format_description;
 use uuid::Uuid;
 
 use super::SESSIONS_SUBDIR;
+use super::recorder::SessionMetaWithGit;
 
 /// Returned page of conversation summaries.
 #[derive(Debug, Default, PartialEq)]
@@ -170,7 +171,9 @@ async fn traverse_directories_for_paths(
                     let head = read_first_jsonl_records(&path, HEAD_RECORD_LIMIT)
                         .await
                         .unwrap_or_default();
-                    items.push(ConversationItem { path, head });
+                    if should_include_session(&head) {
+                        items.push(ConversationItem { path, head });
+                    }
                 }
             }
         }
@@ -296,3 +299,37 @@ async fn read_first_jsonl_records(
     }
     Ok(head)
 }
+
+/// Return true if this conversation should be included in the listing.
+///
+/// Current rule: include only when the first JSON object is a session meta record
+/// (i.e., has `{"record_type": "session_meta", ...}`), which is how rollout
+/// files are written. Empty or malformed heads are excluded.
+fn should_include_session(head: &[serde_json::Value]) -> bool {
+    let Some(first) = head.first() else {
+        return false;
+    };
+    passes_session_meta_filter(first)
+}
+
+/// Validate that the first record is a fully‑formed session meta line.
+///
+/// Requirements:
+/// - `record_type == "session_meta"`
+/// - Remaining fields (after removing `record_type`) deserialize into
+///   `SessionMetaWithGit`.
+fn passes_session_meta_filter(first: &serde_json::Value) -> bool {
+    let Some(obj) = first.as_object() else {
+        return false;
+    };
+    let record_type = obj.get("record_type").and_then(|v| v.as_str());
+    if record_type != Some("session_meta") {
+        return false;
+    }
+
+    // Remove the marker field and validate the remainder matches SessionMetaWithGit
+    let mut cleaned = obj.clone();
+    cleaned.remove("record_type");
+    let val = serde_json::Value::Object(cleaned);
+    serde_json::from_value::<SessionMetaWithGit>(val).is_ok()
+}

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

@@ -7,6 +7,7 @@ use codex_core::protocol::ConversationPathResponseEvent;
 use codex_core::protocol::EventMsg;
 use codex_core::protocol::InputItem;
 use codex_core::protocol::Op;
+use codex_protocol::models::ResponseItem;
 use core_test_support::load_default_config_for_test;
 use core_test_support::wait_for_event;
 use tempfile::TempDir;
@@ -76,13 +77,30 @@ async fn fork_conversation_twice_drops_to_first_message() {
     let base_history =
         wait_for_event(&codex, |ev| matches!(ev, EventMsg::ConversationHistory(_))).await;
 
-    // Capture entries from the base history and compute expected prefixes after each fork.
-    let entries_after_three = match &base_history {
-        EventMsg::ConversationHistory(ConversationPathResponseEvent { entries, .. }) => {
-            entries.clone()
-        }
+    // Capture path/id from the base history and compute expected prefixes after each fork.
+    let (base_conv_id, base_path) = match &base_history {
+        EventMsg::ConversationHistory(ConversationPathResponseEvent {
+            conversation_id,
+            path,
+        }) => (*conversation_id, path.clone()),
         _ => panic!("expected ConversationHistory event"),
     };
+
+    // Read entries from rollout file.
+    async fn read_response_entries(path: &std::path::Path) -> Vec<ResponseItem> {
+        let text = tokio::fs::read_to_string(path).await.unwrap_or_default();
+        let mut out = Vec::new();
+        for line in text.lines() {
+            if line.trim().is_empty() {
+                continue;
+            }
+            if let Ok(item) = serde_json::from_str::<ResponseItem>(line) {
+                out.push(item);
+            }
+        }
+        out
+    }
+    let entries_after_three: Vec<ResponseItem> = read_response_entries(&base_path).await;
     // History layout for this test:
     // [0] user instructions,
     // [1] environment context,
@@ -113,7 +131,7 @@ async fn fork_conversation_twice_drops_to_first_message() {
         conversation: codex_fork1,
         ..
     } = conversation_manager
-        .fork_conversation(entries_after_three.clone(), 1, config_for_fork.clone())
+        .fork_conversation(base_path.clone(), base_conv_id, 1, config_for_fork.clone())
         .await
         .expect("fork 1");
 
@@ -122,23 +140,22 @@ async fn fork_conversation_twice_drops_to_first_message() {
         matches!(ev, EventMsg::ConversationHistory(_))
     })
     .await;
-    let entries_after_first_fork = match &fork1_history {
-        EventMsg::ConversationHistory(ConversationPathResponseEvent { entries, .. }) => {
-            assert!(matches!(
-                fork1_history,
-                EventMsg::ConversationHistory(ConversationPathResponseEvent { ref entries, .. }) if *entries == expected_after_first
-            ));
-            entries.clone()
-        }
+    let (fork1_id, fork1_path) = match &fork1_history {
+        EventMsg::ConversationHistory(ConversationPathResponseEvent {
+            conversation_id,
+            path,
+        }) => (*conversation_id, path.clone()),
         _ => panic!("expected ConversationHistory event after first fork"),
     };
+    let entries_after_first_fork: Vec<ResponseItem> = read_response_entries(&fork1_path).await;
+    assert_eq!(entries_after_first_fork, expected_after_first);
 
     // Fork again with n=1 → drops the (new) last user message, leaving only the first.
     let NewConversation {
         conversation: codex_fork2,
         ..
     } = conversation_manager
-        .fork_conversation(entries_after_first_fork.clone(), 1, config_for_fork.clone())
+        .fork_conversation(fork1_path.clone(), fork1_id, 1, config_for_fork.clone())
         .await
         .expect("fork 2");
 
@@ -147,8 +164,13 @@ async fn fork_conversation_twice_drops_to_first_message() {
         matches!(ev, EventMsg::ConversationHistory(_))
     })
     .await;
-    assert!(matches!(
-        fork2_history,
-        EventMsg::ConversationHistory(ConversationPathResponseEvent { ref entries, .. }) if *entries == expected_after_second
-    ));
+    let (_fork2_id, fork2_path) = match &fork2_history {
+        EventMsg::ConversationHistory(ConversationPathResponseEvent {
+            conversation_id,
+            path,
+        }) => (*conversation_id, path.clone()),
+        _ => panic!("expected ConversationHistory event after second fork"),
+    };
+    let entries_after_second_fork: Vec<ResponseItem> = read_response_entries(&fork2_path).await;
+    assert_eq!(entries_after_second_fork, expected_after_second);
 }