From 4a7b842b537aece10a5d9642db4a9d4d1e74b845 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Fri, 5 Sep 2025 13:18:49 -0700 Subject: [PATCH] progress --- codex-rs/core/src/codex.rs | 40 ++----------- codex-rs/core/src/conversation_manager.rs | 4 +- codex-rs/core/src/rollout/list.rs | 39 +++++++++++- .../core/tests/suite/fork_conversation.rs | 60 +++++++++++++------ 4 files changed, 86 insertions(+), 57 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index a9a135f3bb..f59eb3f695 100644 --- a/codex-rs/core/src/codex.rs +++ b/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 = 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 = 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(&self, item: T) - where - T: Into, - { - 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); diff --git a/codex-rs/core/src/conversation_manager.rs b/codex-rs/core/src/conversation_manager.rs index 3b9a3d8ce8..2f387a3486 100644 --- a/codex-rs/core/src/conversation_manager.rs +++ b/codex-rs/core/src/conversation_manager.rs @@ -263,6 +263,8 @@ mod tests { assistant_msg("a4"), ]; + let items: Vec = 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 { @@ -279,7 +281,7 @@ mod tests { let texts: Vec = 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(); diff --git a/codex-rs/core/src/rollout/list.rs b/codex-rs/core/src/rollout/list.rs index 44f08e67a4..5d1030ad65 100644 --- a/codex-rs/core/src/rollout/list.rs +++ b/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::(val).is_ok() +} diff --git a/codex-rs/core/tests/suite/fork_conversation.rs b/codex-rs/core/tests/suite/fork_conversation.rs index b809d62536..5b9c26a7ae 100644 --- a/codex-rs/core/tests/suite/fork_conversation.rs +++ b/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 { + 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::(line) { + out.push(item); + } + } + out + } + let entries_after_three: Vec = 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 = 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 = read_response_entries(&fork2_path).await; + assert_eq!(entries_after_second_fork, expected_after_second); }