diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index 682361ea9d..887613e373 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/codex-rs/rollout/src/compression_tests.rs @@ -457,6 +457,7 @@ fn write_rollout(path: &std::path::Path, thread_id: ThreadId, message: &str) -> let session_meta_line = SessionMetaLine { meta: SessionMeta { id: thread_id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp: "2025-01-03T12:00:00Z".to_string(), diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 23be07996c..5c350407e2 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -23,6 +23,7 @@ pub(crate) use codex_protocol::protocol; pub const SESSIONS_SUBDIR: &str = "sessions"; pub const ARCHIVED_SESSIONS_SUBDIR: &str = "archived_sessions"; +pub const ROTATED_ROLLOUT_SEGMENTS_SUBDIR: &str = "rotated_rollout_segments"; pub static INTERACTIVE_SESSION_SOURCES: LazyLock> = LazyLock::new(|| { vec![ SessionSource::Cli, @@ -41,6 +42,7 @@ pub use compression::spawn_rollout_compression_worker; pub use config::Config; pub use config::RolloutConfig; pub use config::RolloutConfigView; +pub use list::ArchivedThreadRolloutDisposition; pub use list::Cursor; pub use list::SortDirection; pub use list::ThreadItem; @@ -48,7 +50,9 @@ pub use list::ThreadListConfig; pub use list::ThreadListLayout; pub use list::ThreadSortKey; pub use list::ThreadsPage; +pub use list::classify_archived_thread_rollout; pub use list::find_archived_thread_path_by_id_str; +pub use list::find_rollout_path_by_segment_id; pub use list::find_thread_path_by_id_str; #[deprecated(note = "use find_thread_path_by_id_str")] pub use list::find_thread_path_by_id_str as find_conversation_path_by_id_str; @@ -58,6 +62,7 @@ pub use list::parse_cursor; pub use list::read_head_for_summary; pub use list::read_session_meta_line; pub use list::read_thread_item_from_rollout; +pub use list::resolve_rollout_reference_rollout_path; pub use list::rollout_date_parts; pub use metadata::builder_from_items; pub use policy::is_persisted_rollout_item; diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index 845145aaef..6f0792a19b 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -16,12 +16,15 @@ use time::macros::format_description; use uuid::Uuid; use super::ARCHIVED_SESSIONS_SUBDIR; +use super::ROTATED_ROLLOUT_SEGMENTS_SUBDIR; use super::SESSIONS_SUBDIR; use super::compression; use crate::protocol::EventMsg; use crate::state_db; use codex_file_search as file_search; +use codex_protocol::SegmentId; use codex_protocol::ThreadId; +use codex_protocol::models::ResponseItem; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::SessionMetaLine; @@ -89,6 +92,7 @@ pub type ConversationsPage = ThreadsPage; #[derive(Default)] struct HeadTailSummary { saw_session_meta: bool, + saw_compacted_user_message: bool, thread_id: Option, first_user_message: Option, preview: Option, @@ -137,6 +141,12 @@ pub struct ThreadListConfig<'a> { pub layout: ThreadListLayout, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ArchivedThreadRolloutDisposition { + CanonicalArchivedThread, + LegacyRotatedSegment { live_rollout_path: Option }, +} + /// Pagination cursor identifying the timestamp of the last item in a page. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Cursor { @@ -767,8 +777,10 @@ async fn build_thread_item( { return None; } - // Apply filters: must have session meta and a discoverable preview. - if summary.saw_session_meta && summary.preview.is_some() { + // Apply filters: must have session meta and either a discoverable preview or compacted user + // history. + if summary.saw_session_meta && (summary.preview.is_some() || summary.saw_compacted_user_message) + { let HeadTailSummary { thread_id, first_user_message, @@ -942,6 +954,19 @@ pub(crate) fn parse_timestamp_uuid_from_filename(name: &str) -> Option<(OffsetDa Some((ts, uuid)) } +fn parse_timestamp_string_uuid_from_filename(name: &str) -> Option<(&str, Uuid)> { + // Expected: rollout-YYYY-MM-DDThh-mm-ss-.jsonl + let core = name.strip_prefix("rollout-")?.strip_suffix(".jsonl")?; + core.match_indices('-') + .rev() + .find_map(|(index, _)| { + Uuid::parse_str(&core[index + 1..]) + .ok() + .map(|uuid| (index, uuid)) + }) + .map(|(index, uuid)| (&core[..index], uuid)) +} + struct ThreadCandidate { path: PathBuf, id: Uuid, @@ -1070,8 +1095,93 @@ impl<'a> ProviderMatcher<'a> { } async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result { + const MAX_SUMMARY_REFERENCE_DEPTH: usize = 8; + + let (mut summary, mut reference) = read_head_summary_file(path, head_limit).await?; + let mut referenced_summaries = Vec::new(); + let mut remaining_depth = MAX_SUMMARY_REFERENCE_DEPTH; + while remaining_depth > 0 { + let Some(next_reference) = reference.take() else { + break; + }; + if next_reference.max_depth == 0 || next_reference.nth_user_message.is_some() { + break; + } + let reference_depth = next_reference.max_depth; + let reference_path = if let Some(codex_home) = codex_home_from_rollout_path(path) { + match Box::pin(resolve_rollout_reference_rollout_path( + codex_home, + &next_reference, + )) + .await + { + Ok(reference_path) => reference_path, + Err(_) => break, + } + } else if let Some(reference_path) = + compression::existing_rollout_path(next_reference.rollout_path.as_path()).await + { + reference_path + } else { + break; + }; + let (referenced_summary, next_reference) = + read_head_summary_file(reference_path.as_path(), head_limit).await?; + referenced_summaries.push(referenced_summary); + reference = next_reference; + remaining_depth = remaining_depth + .saturating_sub(1) + .min(reference_depth.saturating_sub(1)); + } + + let mut referenced_preview = None; + let mut referenced_first_user_message = None; + for referenced_summary in referenced_summaries.into_iter().rev() { + if referenced_preview.is_none() { + referenced_preview = referenced_summary.preview; + } + if referenced_first_user_message.is_none() { + referenced_first_user_message = referenced_summary.first_user_message; + } + summary.saw_compacted_user_message |= referenced_summary.saw_compacted_user_message; + } + if let Some(preview) = referenced_preview { + summary.preview = Some(preview); + } + if let Some(first_user_message) = referenced_first_user_message { + summary.first_user_message = Some(first_user_message); + } + Ok(summary) +} + +pub(crate) async fn read_preview_metadata( + path: &Path, +) -> io::Result<(Option, Option)> { + let summary = read_head_summary(path, HEAD_RECORD_LIMIT).await?; + Ok((summary.first_user_message, summary.preview)) +} + +fn codex_home_from_rollout_path(path: &Path) -> Option<&Path> { + path.ancestors().find_map(|ancestor| { + matches!( + ancestor.file_name().and_then(|name| name.to_str()), + Some(SESSIONS_SUBDIR | ARCHIVED_SESSIONS_SUBDIR) + ) + .then(|| ancestor.parent()) + .flatten() + }) +} + +async fn read_head_summary_file( + path: &Path, + head_limit: usize, +) -> io::Result<( + HeadTailSummary, + Option, +)> { let mut lines = compression::open_rollout_line_reader(path).await?; let mut summary = HeadTailSummary::default(); + let mut reference = None; let mut lines_scanned = 0usize; while lines_scanned < head_limit @@ -1117,6 +1227,11 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result { + if reference.is_none() { + reference = Some(item); + } + } RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) => { summary .created_at @@ -1125,8 +1240,14 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result { // Not included in `head`; skip. } - RolloutItem::Compacted(_) => { - // Not included in `head`; skip. + RolloutItem::Compacted(compacted) => { + if compacted + .replacement_history + .as_deref() + .is_some_and(|history| history.iter().any(ResponseItem::is_user_message)) + { + summary.saw_compacted_user_message = true; + } } RolloutItem::EventMsg(ev) => { if let Some(preview) = event_msg_preview(&ev) { @@ -1150,7 +1271,7 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result io::Result {} } @@ -1264,6 +1386,38 @@ async fn find_thread_path_by_id_str_in_subdir( subdir: &str, id_str: &str, state_db_ctx: Option<&codex_state::StateRuntime>, +) -> io::Result> { + let found = + find_thread_path_by_id_str_in_subdir_unvalidated(codex_home, subdir, id_str, state_db_ctx) + .await?; + if subdir != ARCHIVED_SESSIONS_SUBDIR { + return Ok(found); + } + let Some(found_path) = found else { + return Ok(None); + }; + match classify_archived_thread_rollout(codex_home, found_path.as_path(), state_db_ctx).await? { + ArchivedThreadRolloutDisposition::CanonicalArchivedThread => Ok(Some(found_path)), + ArchivedThreadRolloutDisposition::LegacyRotatedSegment { live_rollout_path } => { + if let Some(live_rollout_path) = live_rollout_path { + state_db::read_repair_rollout_path( + state_db_ctx, + ThreadId::from_string(id_str).ok(), + Some(false), + live_rollout_path.as_path(), + ) + .await; + } + Ok(None) + } + } +} + +async fn find_thread_path_by_id_str_in_subdir_unvalidated( + codex_home: &Path, + subdir: &str, + id_str: &str, + state_db_ctx: Option<&codex_state::StateRuntime>, ) -> io::Result> { // Validate UUID format early. if Uuid::parse_str(id_str).is_err() { @@ -1484,6 +1638,237 @@ pub async fn find_archived_thread_path_by_id_str( .await } +pub async fn find_rollout_path_by_segment_id( + codex_home: &Path, + thread_id: ThreadId, + segment_id: SegmentId, +) -> io::Result> { + if let Some(path) = find_rollout_path_by_segment_id_in_subdir( + codex_home, + SESSIONS_SUBDIR, + thread_id, + segment_id, + ) + .await? + { + return Ok(Some(path)); + } + if let Some(path) = find_rollout_path_by_segment_id_in_subdir( + codex_home, + ROTATED_ROLLOUT_SEGMENTS_SUBDIR, + thread_id, + segment_id, + ) + .await? + { + return Ok(Some(path)); + } + find_rollout_path_by_segment_id_in_subdir( + codex_home, + ARCHIVED_SESSIONS_SUBDIR, + thread_id, + segment_id, + ) + .await +} + +pub async fn classify_archived_thread_rollout( + codex_home: &Path, + rollout_path: &Path, + state_db_ctx: Option<&codex_state::StateRuntime>, +) -> io::Result { + let archived_root = codex_home.join(ARCHIVED_SESSIONS_SUBDIR); + let Ok(relative_path) = rollout_path.strip_prefix(archived_root.as_path()) else { + return Ok(ArchivedThreadRolloutDisposition::CanonicalArchivedThread); + }; + if relative_path.components().count() != 1 { + return Ok(ArchivedThreadRolloutDisposition::CanonicalArchivedThread); + } + + let live_rollout_path = + find_live_rollout_path_for_archived_candidate(codex_home, rollout_path, state_db_ctx) + .await?; + if live_rollout_path.is_some() { + return Ok(ArchivedThreadRolloutDisposition::LegacyRotatedSegment { live_rollout_path }); + } + Ok(ArchivedThreadRolloutDisposition::CanonicalArchivedThread) +} + +async fn find_live_rollout_path_for_archived_candidate( + codex_home: &Path, + rollout_path: &Path, + state_db_ctx: Option<&codex_state::StateRuntime>, +) -> io::Result> { + let meta_line = match read_session_meta_line(rollout_path).await { + Ok(meta_line) => meta_line, + Err(_) => return Ok(None), + }; + let thread_id = meta_line.meta.id.to_string(); + find_thread_path_by_id_str_in_subdir_unvalidated( + codex_home, + SESSIONS_SUBDIR, + thread_id.as_str(), + state_db_ctx, + ) + .await +} + +async fn find_rollout_path_by_segment_id_in_subdir( + codex_home: &Path, + subdir: &str, + thread_id: ThreadId, + segment_id: SegmentId, +) -> io::Result> { + let root = codex_home.join(subdir); + if !tokio::fs::try_exists(&root).await.unwrap_or(false) { + return Ok(None); + } + + let target_thread_id = thread_id.to_string(); + let mut stack = vec![root]; + let mut scanned_files = 0usize; + while let Some(dir) = stack.pop() { + let mut read_dir = match tokio::fs::read_dir(&dir).await { + Ok(read_dir) => read_dir, + Err(err) => { + tracing::warn!("failed to read rollout directory {}: {err}", dir.display()); + continue; + } + }; + while let Some(entry) = read_dir.next_entry().await? { + let path = entry.path(); + let file_type = entry.file_type().await?; + if file_type.is_dir() { + stack.push(path); + continue; + } + if !file_type.is_file() { + continue; + } + let Some(rollout_file) = compression::RolloutFile::from_path(path) else { + continue; + }; + let Some((_, uuid)) = + parse_timestamp_uuid_from_filename(rollout_file.plain_file_name()) + else { + continue; + }; + if uuid.to_string() != target_thread_id { + continue; + } + scanned_files = scanned_files.saturating_add(1); + if scanned_files > MAX_SCAN_FILES { + return Ok(None); + } + let Ok(meta_line) = read_session_meta_line(rollout_file.path()).await else { + continue; + }; + if meta_line.meta.id == thread_id && meta_line.meta.segment_id == Some(segment_id) { + return Ok(Some(rollout_file.into_path())); + } + } + } + + Ok(None) +} + +pub async fn resolve_rollout_reference_rollout_path( + codex_home: &Path, + reference: &codex_protocol::protocol::RolloutReferenceItem, +) -> io::Result { + // A compacted live thread can keep the same rollout path and timestamp while moving the + // referenced predecessor into archive storage, so segment identity must win before fallback + // lookup by mutable live-path metadata. + if let (Some(thread_id), Some(segment_id)) = (reference.thread_id, reference.segment_id) + && let Some(path) = + find_rollout_path_by_segment_id(codex_home, thread_id, segment_id).await? + { + return Ok(path); + } + + let rollout_path = reference.rollout_path.as_path(); + if tokio::fs::try_exists(rollout_path).await.unwrap_or(false) { + return Ok(rollout_path.to_path_buf()); + } + + if let (Some(thread_id), Some(rollout_timestamp)) = + (reference.thread_id, reference.rollout_timestamp.as_deref()) + { + let file_name = format!("rollout-{rollout_timestamp}-{thread_id}.jsonl"); + if let Some(active_path) = + rollout_path_for_timestamp_file(codex_home, rollout_timestamp, &file_name) + && tokio::fs::try_exists(active_path.as_path()) + .await + .unwrap_or(false) + { + return Ok(active_path); + } + let archived_path = codex_home.join(ARCHIVED_SESSIONS_SUBDIR).join(&file_name); + if tokio::fs::try_exists(archived_path.as_path()) + .await + .unwrap_or(false) + { + return Ok(archived_path); + } + } + + let Some(file_name) = rollout_path + .file_name() + .and_then(|file_name| file_name.to_str()) + else { + return Ok(rollout_path.to_path_buf()); + }; + let Some((rollout_timestamp, uuid)) = parse_timestamp_string_uuid_from_filename(file_name) + else { + return Ok(rollout_path.to_path_buf()); + }; + let archived_path = codex_home.join(ARCHIVED_SESSIONS_SUBDIR).join(file_name); + if tokio::fs::try_exists(archived_path.as_path()) + .await + .unwrap_or(false) + { + return Ok(archived_path); + } + if let Some(active_path) = + rollout_path_for_timestamp_file(codex_home, rollout_timestamp, file_name) + && tokio::fs::try_exists(active_path.as_path()) + .await + .unwrap_or(false) + { + return Ok(active_path); + } + let id = uuid.to_string(); + if let Some(path) = + find_thread_path_by_id_str(codex_home, id.as_str(), /*state_db_ctx*/ None).await? + { + return Ok(path); + } + if let Some(path) = + find_archived_thread_path_by_id_str(codex_home, id.as_str(), /*state_db_ctx*/ None).await? + { + return Ok(path); + } + Ok(rollout_path.to_path_buf()) +} + +fn rollout_path_for_timestamp_file( + codex_home: &Path, + rollout_timestamp: &str, + file_name: &str, +) -> Option { + let year = rollout_timestamp.get(0..4)?; + let month = rollout_timestamp.get(5..7)?; + let day = rollout_timestamp.get(8..10)?; + Some( + codex_home + .join(SESSIONS_SUBDIR) + .join(year) + .join(month) + .join(day) + .join(file_name), + ) +} + /// Extract the `YYYY/MM/DD` directory components from a rollout filename. pub fn rollout_date_parts(file_name: &OsStr) -> Option<(String, String, String)> { let name = file_name.to_string_lossy(); diff --git a/codex-rs/rollout/src/metadata.rs b/codex-rs/rollout/src/metadata.rs index 19e9a9320c..f85d803ca3 100644 --- a/codex-rs/rollout/src/metadata.rs +++ b/codex-rs/rollout/src/metadata.rs @@ -2,6 +2,7 @@ use crate::ARCHIVED_SESSIONS_SUBDIR; use crate::SESSIONS_SUBDIR; use crate::compression; use crate::list::parse_timestamp_uuid_from_filename; +use crate::list::read_preview_metadata; use crate::recorder::RolloutRecorder; use crate::state_db::normalize_cwd_for_state_db; use chrono::DateTime; @@ -69,6 +70,7 @@ pub fn builder_from_items( RolloutItem::SessionMeta(meta_line) => Some(meta_line), RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) + | RolloutItem::RolloutReference(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, @@ -113,6 +115,22 @@ pub async fn extract_metadata_from_rollout( for item in &items { apply_rollout_item(&mut metadata, item, default_provider); } + let has_segment_reference = items.iter().any(|item| { + matches!( + item, + RolloutItem::RolloutReference(reference) if reference.nth_user_message.is_none() + ) + }); + if has_segment_reference + && let Ok((first_user_message, preview)) = read_preview_metadata(rollout_path).await + { + if first_user_message.is_some() { + metadata.first_user_message = first_user_message; + } + if preview.is_some() { + metadata.preview = preview; + } + } if let Some(updated_at) = file_modified_time_utc(rollout_path).await { metadata.updated_at = updated_at; } @@ -122,6 +140,7 @@ pub async fn extract_metadata_from_rollout( RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) + | RolloutItem::RolloutReference(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, @@ -217,11 +236,26 @@ pub(crate) async fn backfill_sessions_with_lease( } match collect_rollout_paths(&root).await { Ok(paths) => { - rollout_paths.extend(paths.into_iter().map(|path| BackfillRolloutPath { - watermark: backfill_watermark_for_path(codex_home, &path), - path, - archived, - })); + for path in paths { + if archived + && matches!( + crate::list::classify_archived_thread_rollout( + codex_home, + path.as_path(), + Some(runtime), + ) + .await, + Ok(crate::list::ArchivedThreadRolloutDisposition::LegacyRotatedSegment { .. }) + ) + { + continue; + } + rollout_paths.push(BackfillRolloutPath { + watermark: backfill_watermark_for_path(codex_home, &path), + path, + archived, + }); + } } Err(err) => { warn!( diff --git a/codex-rs/rollout/src/metadata_tests.rs b/codex-rs/rollout/src/metadata_tests.rs index 03d251e1b3..8748d4a8ab 100644 --- a/codex-rs/rollout/src/metadata_tests.rs +++ b/codex-rs/rollout/src/metadata_tests.rs @@ -5,14 +5,18 @@ use chrono::DateTime; use chrono::NaiveDateTime; use chrono::Timelike; use chrono::Utc; +use codex_protocol::SegmentId; use codex_protocol::ThreadId; use codex_protocol::protocol::CompactedItem; +use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GitInfo; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; +use codex_protocol::protocol::RolloutReferenceItem; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::UserMessageEvent; use codex_state::BackfillStatus; use codex_state::ThreadMetadataBuilder; use pretty_assertions::assert_eq; @@ -34,6 +38,7 @@ async fn extract_metadata_from_rollout_uses_session_meta() { let session_meta = SessionMeta { id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp: "2026-01-27T12:34:56Z".to_string(), @@ -88,6 +93,7 @@ async fn extract_metadata_from_rollout_returns_latest_memory_mode() { let session_meta = SessionMeta { id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp: "2026-01-27T12:34:56Z".to_string(), @@ -143,6 +149,109 @@ async fn extract_metadata_from_rollout_returns_latest_memory_mode() { assert_eq!(outcome.memory_mode.as_deref(), Some("polluted")); } +#[tokio::test] +async fn extract_metadata_from_rollout_reads_referenced_preview() { + let dir = tempdir().expect("tempdir"); + let codex_home = dir.path(); + let uuid = Uuid::new_v4(); + let id = ThreadId::from_string(&uuid.to_string()).expect("thread id"); + let previous_segment_id = SegmentId::new(); + let current_segment_id = SegmentId::new(); + let file_name = format!("rollout-2026-01-27T12-34-56-{uuid}.jsonl"); + let previous_path = codex_home + .join(crate::ROTATED_ROLLOUT_SEGMENTS_SUBDIR) + .join(id.to_string()) + .join(previous_segment_id.to_string()) + .join(&file_name); + let current_path = codex_home.join("sessions/2026/01/27").join(&file_name); + let session_meta = |segment_id| { + RolloutItem::SessionMeta(SessionMetaLine { + meta: SessionMeta { + id, + segment_id: Some(segment_id), + forked_from_id: None, + parent_thread_id: None, + timestamp: "2026-01-27T12:34:56Z".to_string(), + cwd: codex_home.to_path_buf(), + originator: "cli".to_string(), + cli_version: "0.0.0".to_string(), + source: SessionSource::default(), + thread_source: None, + agent_path: None, + agent_nickname: None, + agent_role: None, + model_provider: Some("openai".to_string()), + base_instructions: None, + dynamic_tools: None, + memory_mode: None, + multi_agent_version: None, + }, + git: None, + }) + }; + std::fs::create_dir_all(previous_path.parent().expect("previous parent")) + .expect("create previous parent"); + let mut previous_file = File::create(&previous_path).expect("create previous rollout"); + for item in [ + session_meta(previous_segment_id), + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "Referenced preview".to_string(), + ..Default::default() + })), + ] { + let line = RolloutLine { + timestamp: "2026-01-27T12:34:56Z".to_string(), + item, + }; + writeln!( + previous_file, + "{}", + serde_json::to_string(&line).expect("serialize previous line") + ) + .expect("write previous line"); + } + std::fs::create_dir_all(current_path.parent().expect("current parent")) + .expect("create current parent"); + let mut current_file = File::create(¤t_path).expect("create current rollout"); + for item in [ + session_meta(current_segment_id), + RolloutItem::RolloutReference(RolloutReferenceItem { + rollout_path: current_path.clone(), + thread_id: Some(id), + rollout_timestamp: Some("2026-01-27T12-34-56".to_string()), + segment_id: Some(previous_segment_id), + max_depth: 2, + nth_user_message: None, + compacted_replacement_history_filter_texts: None, + }), + ] { + let line = RolloutLine { + timestamp: "2026-01-27T12:34:56Z".to_string(), + item, + }; + writeln!( + current_file, + "{}", + serde_json::to_string(&line).expect("serialize current line") + ) + .expect("write current line"); + } + drop(current_file); + + let outcome = extract_metadata_from_rollout(¤t_path, "openai") + .await + .expect("extract referenced metadata"); + + assert_eq!( + outcome.metadata.first_user_message.as_deref(), + Some("Referenced preview") + ); + assert_eq!( + outcome.metadata.preview.as_deref(), + Some("Referenced preview") + ); +} + #[test] fn builder_from_items_falls_back_to_filename() { let dir = tempdir().expect("tempdir"); @@ -289,6 +398,46 @@ async fn backfill_sessions_preserves_existing_git_branch_and_fills_missing_git_f ); } +#[tokio::test] +async fn backfill_ignores_legacy_rotated_segment_when_live_rollout_exists() { + let dir = tempdir().expect("tempdir"); + let codex_home = dir.path().to_path_buf(); + let thread_uuid = Uuid::new_v4(); + let live_rollout_path = write_rollout_in_sessions( + codex_home.as_path(), + "2026-01-27T12-34-56", + "2026-01-27T12:34:56Z", + thread_uuid, + /*git*/ None, + ); + let legacy_rotated_path = codex_home + .join(ARCHIVED_SESSIONS_SUBDIR) + .join(live_rollout_path.file_name().expect("rollout file name")); + std::fs::create_dir_all( + legacy_rotated_path + .parent() + .expect("legacy rotated rollout parent"), + ) + .expect("create legacy rotated rollout parent"); + std::fs::copy(live_rollout_path.as_path(), legacy_rotated_path.as_path()) + .expect("copy legacy rotated rollout"); + + let runtime = codex_state::StateRuntime::init(codex_home.clone(), "test-provider".to_string()) + .await + .expect("initialize runtime"); + + backfill_sessions(runtime.as_ref(), codex_home.as_path(), "test-provider").await; + + let thread_id = ThreadId::from_string(&thread_uuid.to_string()).expect("thread id"); + let metadata = runtime + .get_thread(thread_id) + .await + .expect("get thread") + .expect("thread metadata"); + assert_eq!(metadata.rollout_path, live_rollout_path); + assert_eq!(metadata.archived_at, None); +} + #[tokio::test] async fn backfill_sessions_normalizes_cwd_before_upsert() { let dir = tempdir().expect("tempdir"); @@ -352,6 +501,7 @@ fn write_rollout_in_sessions_with_cwd( let path = sessions_dir.join(format!("rollout-{filename_ts}-{thread_uuid}.jsonl")); let session_meta = SessionMeta { id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp: event_ts.to_string(), diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 40ca77c97e..805b91f25f 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -7,6 +7,7 @@ pub fn is_persisted_rollout_item(item: &RolloutItem) -> bool { match item { RolloutItem::ResponseItem(item) => should_persist_response_item(item), RolloutItem::InterAgentCommunication(_) => true, + RolloutItem::RolloutReference(_) => true, RolloutItem::EventMsg(ev) => should_persist_event_msg(ev), // Persist Codex executive markers so we can analyze flows (e.g., compaction, API turns). RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => { diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 9ade474896..fe766060c0 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -31,6 +31,7 @@ use tracing::warn; use super::ARCHIVED_SESSIONS_SUBDIR; use super::SESSIONS_SUBDIR; use super::compression; +use super::list::ArchivedThreadRolloutDisposition; use super::list::Cursor; use super::list::SortDirection; use super::list::ThreadItem; @@ -489,6 +490,38 @@ impl RolloutRecorder { ) .await; if let Some(db_page) = db_page { + if archived { + repair_legacy_rotated_archived_rows( + codex_home, + state_db_ctx.as_deref(), + &db_page.items, + ) + .await?; + if search_term.is_some() + && let Some(repaired_db_page) = state_db::list_threads_db( + state_db_ctx.as_deref(), + codex_home, + page_size, + cursor, + sort_key, + sort_direction, + allowed_sources, + model_providers, + cwd_filters, + archived, + search_term, + ) + .await + { + return Ok(repaired_db_page.into()); + } + let page = page_from_filesystem_scan(fs_page, sort_direction, page_size, sort_key); + return Ok(fill_missing_thread_item_metadata_from_state_db( + state_db_ctx.as_deref(), + page, + ) + .await); + } if search_term.is_some() && (!db_page.items.is_empty() || cursor.is_some()) { for item in &db_page.items { state_db::reconcile_rollout( @@ -695,6 +728,7 @@ impl RolloutRecorder { let session_meta = SessionMeta { id: session_id, + segment_id: None, forked_from_id, parent_thread_id, timestamp, @@ -1182,20 +1216,64 @@ async fn list_threads_from_files_desc_unfiltered( ) -> std::io::Result { if archived { let root = codex_home.join(ARCHIVED_SESSIONS_SUBDIR); - get_threads_in_root( - root, - page_size, - cursor, - sort_key, - ThreadListConfig { - allowed_sources, - model_providers, - cwd_filters, - default_provider, - layout: ThreadListLayout::Flat, - }, - ) - .await + let mut canonical_items = Vec::with_capacity(page_size); + let mut num_scanned_files = 0usize; + let mut reached_scan_cap = false; + let mut page_cursor = cursor.cloned(); + + loop { + let page = get_threads_in_root( + root.clone(), + page_size, + page_cursor.as_ref(), + sort_key, + ThreadListConfig { + allowed_sources, + model_providers, + cwd_filters, + default_provider, + layout: ThreadListLayout::Flat, + }, + ) + .await?; + num_scanned_files = num_scanned_files.saturating_add(page.num_scanned_files); + reached_scan_cap |= page.reached_scan_cap; + for item in page.items { + if matches!( + super::list::classify_archived_thread_rollout( + codex_home, + item.path.as_path(), + /*state_db_ctx*/ None, + ) + .await?, + ArchivedThreadRolloutDisposition::CanonicalArchivedThread + ) { + canonical_items.push(item); + if canonical_items.len() == page_size { + break; + } + } + } + page_cursor = page.next_cursor; + if canonical_items.len() == page_size || page_cursor.is_none() || reached_scan_cap { + break; + } + } + + let more_matches_available = page_cursor.is_some() || reached_scan_cap; + let next_cursor = if more_matches_available { + canonical_items + .last() + .and_then(|item| cursor_from_thread_item(item, sort_key)) + } else { + None + }; + Ok(ThreadsPage { + items: canonical_items, + next_cursor, + num_scanned_files, + reached_scan_cap, + }) } else { get_threads( codex_home, @@ -1288,6 +1366,35 @@ async fn list_threads_from_files_asc( }) } +async fn repair_legacy_rotated_archived_rows( + codex_home: &Path, + state_db_ctx: Option<&StateRuntime>, + items: &[codex_state::ThreadMetadata], +) -> std::io::Result<()> { + for item in items { + let ArchivedThreadRolloutDisposition::LegacyRotatedSegment { live_rollout_path } = + super::list::classify_archived_thread_rollout( + codex_home, + item.rollout_path.as_path(), + state_db_ctx, + ) + .await? + else { + continue; + }; + if let Some(live_rollout_path) = live_rollout_path { + state_db::read_repair_rollout_path( + state_db_ctx, + Some(item.id), + Some(false), + live_rollout_path.as_path(), + ) + .await; + } + } + Ok(()) +} + async fn filter_thread_items_by_search_term( codex_home: &Path, items: &mut Vec, @@ -1763,6 +1870,7 @@ async fn resume_candidate_matches_cwd( && let Some(latest_turn_context_cwd) = items.iter().rev().find_map(|item| match item { RolloutItem::TurnContext(turn_context) => Some(turn_context.cwd.as_path()), RolloutItem::SessionMeta(_) + | RolloutItem::RolloutReference(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) | RolloutItem::Compacted(_) diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index 68896ad202..b37ef88267 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -84,6 +84,7 @@ async fn state_db_init_backfills_before_returning() -> anyhow::Result<()> { let session_meta_line = SessionMetaLine { meta: SessionMeta { id: thread_id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp: "2026-01-27T12:34:56Z".to_string(), @@ -962,6 +963,72 @@ async fn list_threads_metadata_filter_overlays_state_db_list_metadata() -> std:: Ok(()) } +#[tokio::test] +async fn archived_search_preserves_state_db_only_matches() -> std::io::Result<()> { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let uuid = Uuid::from_u128(9016); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let active_path = write_session_file(home.path(), "2025-01-03T17-00-00", uuid)?; + let archived_path = home + .path() + .join("archived_sessions") + .join(active_path.file_name().expect("rollout file name")); + fs::create_dir_all(archived_path.parent().expect("archived parent"))?; + fs::rename(active_path, &archived_path)?; + + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + runtime + .mark_backfill_complete(/*last_watermark*/ None) + .await + .expect("backfill should be complete"); + let created_at = chrono::Utc + .with_ymd_and_hms(2025, 1, 3, 17, 0, 0) + .single() + .expect("valid datetime"); + let mut builder = codex_state::ThreadMetadataBuilder::new( + thread_id, + archived_path.clone(), + created_at, + SessionSource::Cli, + ); + builder.model_provider = Some(config.model_provider_id.clone()); + builder.cwd = home.path().to_path_buf(); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.archived_at = Some(created_at); + metadata.preview = Some("SQLite-only archived match".to_string()); + metadata.first_user_message = metadata.preview.clone(); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let page = RolloutRecorder::list_archived_threads( + Some(runtime), + &config, + /*page_size*/ 10, + /*cursor*/ None, + ThreadSortKey::CreatedAt, + SortDirection::Desc, + &[], + /*model_providers*/ None, + /*cwd_filters*/ None, + config.model_provider_id.as_str(), + Some("SQLite-only"), + ) + .await?; + + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].thread_id, Some(thread_id)); + assert_eq!(page.items[0].path, archived_path); + Ok(()) +} + #[test] fn fill_missing_thread_item_metadata_preserves_identity_and_prefers_state_git_fields() { let filesystem_thread_id = ThreadId::new(); diff --git a/codex-rs/rollout/src/search.rs b/codex-rs/rollout/src/search.rs index 799e927170..67b79cef30 100644 --- a/codex-rs/rollout/src/search.rs +++ b/codex-rs/rollout/src/search.rs @@ -280,6 +280,7 @@ fn conversation_text_from_item(item: &RolloutItem) -> Option { } } RolloutItem::SessionMeta(_) + | RolloutItem::RolloutReference(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) diff --git a/codex-rs/rollout/src/session_index_tests.rs b/codex-rs/rollout/src/session_index_tests.rs index 97496d394e..e020842373 100644 --- a/codex-rs/rollout/src/session_index_tests.rs +++ b/codex-rs/rollout/src/session_index_tests.rs @@ -26,6 +26,7 @@ fn write_rollout_with_metadata(path: &Path, thread_id: ThreadId) -> std::io::Res item: RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { id: thread_id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp, diff --git a/codex-rs/rollout/src/state_db_tests.rs b/codex-rs/rollout/src/state_db_tests.rs index a84ede199b..41345ed75d 100644 --- a/codex-rs/rollout/src/state_db_tests.rs +++ b/codex-rs/rollout/src/state_db_tests.rs @@ -143,6 +143,7 @@ fn write_rollout_with_user_message( item: RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { id: thread_id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp: "2026-06-01T14:26:25Z".to_string(), diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs index e489eaba59..22d6d86850 100644 --- a/codex-rs/rollout/src/tests.rs +++ b/codex-rs/rollout/src/tests.rs @@ -28,12 +28,14 @@ use crate::list::get_threads; use crate::list::read_head_for_summary; use crate::rollout_date_parts; use anyhow::Result; +use codex_protocol::SegmentId; use codex_protocol::ThreadId; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; +use codex_protocol::protocol::RolloutReferenceItem; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; @@ -240,6 +242,264 @@ fn rollout_date_parts_extracts_directory_components() { ); } +#[tokio::test] +async fn rollout_reference_resolves_archived_file_by_stable_thread_and_timestamp() { + let temp = TempDir::new().expect("tempdir"); + let home = temp.path(); + let uuid = Uuid::new_v4(); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("thread id"); + let ts = "2025-01-03T13-00-00"; + let active_path = home.join(format!("sessions/2025/01/03/rollout-{ts}-{uuid}.jsonl")); + let archived_path = home.join(format!("archived_sessions/rollout-{ts}-{uuid}.jsonl")); + fs::create_dir_all(archived_path.parent().expect("archived parent")).unwrap(); + fs::write(&archived_path, "").unwrap(); + + let resolved = crate::resolve_rollout_reference_rollout_path( + home, + &RolloutReferenceItem { + rollout_path: active_path, + thread_id: Some(thread_id), + rollout_timestamp: Some(ts.to_string()), + segment_id: None, + max_depth: 2, + nth_user_message: None, + compacted_replacement_history_filter_texts: None, + }, + ) + .await + .expect("resolve rollout reference"); + + assert_eq!(resolved, archived_path); +} + +#[tokio::test] +async fn rollout_reference_prefers_segment_id_over_live_path_with_same_thread_timestamp() { + let temp = TempDir::new().expect("tempdir"); + let home = temp.path(); + let uuid = Uuid::new_v4(); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("thread id"); + let referenced_segment_id = SegmentId::new(); + let live_segment_id = SegmentId::new(); + let ts = "2025-01-03T13-00-00"; + let file_name = format!("rollout-{ts}-{uuid}.jsonl"); + let active_path = home.join(format!("sessions/2025/01/03/{file_name}")); + let archived_path = home + .join("archived_sessions") + .join(thread_id.to_string()) + .join(referenced_segment_id.to_string()) + .join(file_name); + write_session_meta(active_path.as_path(), thread_id, live_segment_id, ts); + write_session_meta( + archived_path.as_path(), + thread_id, + referenced_segment_id, + ts, + ); + + let resolved = crate::resolve_rollout_reference_rollout_path( + home, + &RolloutReferenceItem { + rollout_path: active_path, + thread_id: Some(thread_id), + rollout_timestamp: Some(ts.to_string()), + segment_id: Some(referenced_segment_id), + max_depth: 2, + nth_user_message: None, + compacted_replacement_history_filter_texts: None, + }, + ) + .await + .expect("resolve rollout reference"); + + assert_eq!(resolved, archived_path); +} + +#[tokio::test] +async fn rollout_reference_prefers_existing_prior_segment_path_without_segment_id() { + let temp = TempDir::new().expect("tempdir"); + let home = temp.path(); + let uuid = Uuid::new_v4(); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("thread id"); + let live_segment_id = SegmentId::new(); + let ts = "2025-01-03T13-00-00"; + let file_name = format!("rollout-{ts}-{uuid}.jsonl"); + let active_path = home.join(format!("sessions/2025/01/03/{file_name}")); + let legacy_rotated_path = home.join("archived_sessions").join(file_name); + write_session_meta(active_path.as_path(), thread_id, live_segment_id, ts); + write_session_meta( + legacy_rotated_path.as_path(), + thread_id, + SegmentId::new(), + ts, + ); + + let resolved = crate::resolve_rollout_reference_rollout_path( + home, + &RolloutReferenceItem { + rollout_path: legacy_rotated_path.clone(), + thread_id: Some(thread_id), + rollout_timestamp: Some(ts.to_string()), + segment_id: None, + max_depth: 2, + nth_user_message: None, + compacted_replacement_history_filter_texts: None, + }, + ) + .await + .expect("resolve rollout reference"); + + assert_eq!(resolved, legacy_rotated_path); +} + +#[tokio::test] +async fn rollout_reference_resolves_rotated_segment_file_by_segment_id() { + let temp = TempDir::new().expect("tempdir"); + let home = temp.path(); + let uuid = Uuid::new_v4(); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("thread id"); + let referenced_segment_id = SegmentId::new(); + let live_segment_id = SegmentId::new(); + let ts = "2025-01-03T13-00-00"; + let file_name = format!("rollout-{ts}-{uuid}.jsonl"); + let active_path = home.join(format!("sessions/2025/01/03/{file_name}")); + let rotated_segment_path = home + .join(crate::ROTATED_ROLLOUT_SEGMENTS_SUBDIR) + .join(thread_id.to_string()) + .join(referenced_segment_id.to_string()) + .join(file_name); + write_session_meta(active_path.as_path(), thread_id, live_segment_id, ts); + write_session_meta( + rotated_segment_path.as_path(), + thread_id, + referenced_segment_id, + ts, + ); + + let resolved = crate::resolve_rollout_reference_rollout_path( + home, + &RolloutReferenceItem { + rollout_path: active_path, + thread_id: Some(thread_id), + rollout_timestamp: Some(ts.to_string()), + segment_id: Some(referenced_segment_id), + max_depth: 2, + nth_user_message: None, + compacted_replacement_history_filter_texts: None, + }, + ) + .await + .expect("resolve rollout reference"); + + assert_eq!(resolved, rotated_segment_path); +} + +#[tokio::test] +async fn rollout_reference_resolves_compressed_segment_file_by_segment_id() { + let temp = TempDir::new().expect("tempdir"); + let home = temp.path(); + let uuid = Uuid::new_v4(); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("thread id"); + let referenced_segment_id = SegmentId::new(); + let live_segment_id = SegmentId::new(); + let ts = "2025-01-03T13-00-00"; + let file_name = format!("rollout-{ts}-{uuid}.jsonl"); + let active_path = home.join(format!("sessions/2025/01/03/{file_name}")); + let rotated_segment_path = home + .join(crate::ROTATED_ROLLOUT_SEGMENTS_SUBDIR) + .join(thread_id.to_string()) + .join(referenced_segment_id.to_string()) + .join(file_name); + write_session_meta(active_path.as_path(), thread_id, live_segment_id, ts); + write_session_meta( + rotated_segment_path.as_path(), + thread_id, + referenced_segment_id, + ts, + ); + let compressed_segment_path = compress_rollout(rotated_segment_path.as_path()); + + let resolved = crate::resolve_rollout_reference_rollout_path( + home, + &RolloutReferenceItem { + rollout_path: active_path, + thread_id: Some(thread_id), + rollout_timestamp: Some(ts.to_string()), + segment_id: Some(referenced_segment_id), + max_depth: 2, + nth_user_message: None, + compacted_replacement_history_filter_texts: None, + }, + ) + .await + .expect("resolve rollout reference"); + + assert_eq!(resolved, compressed_segment_path); +} + +fn write_session_meta(path: &Path, thread_id: ThreadId, segment_id: SegmentId, timestamp: &str) { + fs::create_dir_all(path.parent().expect("rollout parent")).expect("create rollout parent"); + let line = RolloutLine { + timestamp: timestamp.to_string(), + item: RolloutItem::SessionMeta(SessionMetaLine { + meta: SessionMeta { + id: thread_id, + segment_id: Some(segment_id), + forked_from_id: None, + parent_thread_id: None, + timestamp: timestamp.to_string(), + cwd: ".".into(), + originator: "test_originator".into(), + cli_version: "test_version".into(), + source: SessionSource::VSCode, + thread_source: None, + agent_path: None, + agent_nickname: None, + agent_role: None, + model_provider: Some("test-provider".into()), + base_instructions: None, + dynamic_tools: None, + memory_mode: None, + multi_agent_version: None, + }, + git: None, + }), + }; + fs::write( + path, + format!( + "{}\n", + serde_json::to_string(&line).expect("serialize meta") + ), + ) + .expect("write rollout"); +} + +fn append_rollout_item(path: &Path, timestamp: &str, item: RolloutItem) { + let line = RolloutLine { + timestamp: timestamp.to_string(), + item, + }; + let mut file = fs::OpenOptions::new() + .append(true) + .open(path) + .expect("open rollout"); + writeln!( + file, + "{}", + serde_json::to_string(&line).expect("serialize rollout line") + ) + .expect("append rollout line"); +} + +fn compress_rollout(path: &Path) -> std::path::PathBuf { + let compressed_path = crate::compression::compressed_rollout_path(path); + let input = File::open(path).expect("open rollout"); + let output = File::create(&compressed_path).expect("create compressed rollout"); + zstd::stream::copy_encode(input, output, 0).expect("compress rollout"); + fs::remove_file(path).expect("remove plain rollout"); + compressed_path +} + async fn assert_state_db_rollout_path( home: &Path, thread_id: ThreadId, @@ -932,6 +1192,202 @@ async fn test_list_threads_scans_past_head_for_user_event() { assert_eq!(page.items[0].thread_id, Some(thread_id_from_uuid(uuid))); } +#[tokio::test] +async fn test_list_threads_includes_compacted_replacement_history_with_user_message() { + let temp = TempDir::new().unwrap(); + let home = temp.path(); + + let uuid = Uuid::from_u128(102); + let thread_id = thread_id_from_uuid(uuid); + let ts = "2025-05-03T10-30-00"; + let path = home + .join("sessions/2025/05/03") + .join(format!("rollout-{ts}-{uuid}.jsonl")); + write_session_meta(path.as_path(), thread_id, SegmentId::new(), ts); + let mut file = fs::OpenOptions::new().append(true).open(path).unwrap(); + let compacted = serde_json::json!({ + "timestamp": ts, + "type": "compacted", + "payload": { + "message": "summary", + "replacement_history": [{ + "type": "message", + "role": "user", + "content": [{ + "type": "input_text", + "text": "Hello from compacted history", + }], + }], + }, + }); + writeln!(file, "{compacted}").unwrap(); + + let provider_filter = provider_vec(&[TEST_PROVIDER]); + let page = get_threads( + home, + /*page_size*/ 10, + /*cursor*/ None, + ThreadSortKey::CreatedAt, + INTERACTIVE_SESSION_SOURCES.as_slice(), + Some(provider_filter.as_slice()), + /*cwd_filters*/ None, + TEST_PROVIDER, + ) + .await + .unwrap(); + + assert_eq!(page.items.len(), 1); + let item = &page.items[0]; + assert_eq!(item.thread_id, Some(thread_id)); + assert_eq!(item.preview, None); + assert_eq!(item.first_user_message, None); +} + +#[tokio::test] +async fn test_list_threads_reads_preview_from_referenced_segment() { + let temp = TempDir::new().unwrap(); + let home = temp.path(); + + let uuid = Uuid::from_u128(103); + let thread_id = thread_id_from_uuid(uuid); + let previous_segment_id = SegmentId::new(); + let current_segment_id = SegmentId::new(); + let ts = "2025-05-04T10-30-00"; + let file_name = format!("rollout-{ts}-{uuid}.jsonl"); + let previous_path = home + .join(crate::ROTATED_ROLLOUT_SEGMENTS_SUBDIR) + .join(thread_id.to_string()) + .join(previous_segment_id.to_string()) + .join(&file_name); + let current_path = home.join("sessions/2025/05/04").join(file_name); + write_session_meta(previous_path.as_path(), thread_id, previous_segment_id, ts); + append_rollout_item( + previous_path.as_path(), + ts, + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "Hello from the previous segment".to_string(), + ..Default::default() + })), + ); + write_session_meta(current_path.as_path(), thread_id, current_segment_id, ts); + append_rollout_item( + current_path.as_path(), + ts, + RolloutItem::RolloutReference(RolloutReferenceItem { + rollout_path: current_path.clone(), + thread_id: Some(thread_id), + rollout_timestamp: Some(ts.to_string()), + segment_id: Some(previous_segment_id), + max_depth: 2, + nth_user_message: None, + compacted_replacement_history_filter_texts: None, + }), + ); + + let provider_filter = provider_vec(&[TEST_PROVIDER]); + let page = get_threads( + home, + /*page_size*/ 10, + /*cursor*/ None, + ThreadSortKey::CreatedAt, + INTERACTIVE_SESSION_SOURCES.as_slice(), + Some(provider_filter.as_slice()), + /*cwd_filters*/ None, + TEST_PROVIDER, + ) + .await + .unwrap(); + + assert_eq!(page.items.len(), 1); + assert_eq!( + page.items[0].preview.as_deref(), + Some("Hello from the previous segment") + ); + assert_eq!( + page.items[0].first_user_message.as_deref(), + Some("Hello from the previous segment") + ); +} + +#[tokio::test] +async fn test_list_threads_does_not_read_preview_from_fork_reference() { + let temp = TempDir::new().unwrap(); + let home = temp.path(); + + let parent_uuid = Uuid::from_u128(104); + let child_uuid = Uuid::from_u128(105); + let parent_thread_id = thread_id_from_uuid(parent_uuid); + let child_thread_id = thread_id_from_uuid(child_uuid); + let parent_segment_id = SegmentId::new(); + let child_segment_id = SegmentId::new(); + let ts = "2025-05-04T10-30-00"; + let parent_path = home + .join(crate::ROTATED_ROLLOUT_SEGMENTS_SUBDIR) + .join(parent_thread_id.to_string()) + .join(parent_segment_id.to_string()) + .join(format!("rollout-{ts}-{parent_uuid}.jsonl")); + let child_path = home + .join("sessions/2025/05/04") + .join(format!("rollout-{ts}-{child_uuid}.jsonl")); + write_session_meta( + parent_path.as_path(), + parent_thread_id, + parent_segment_id, + ts, + ); + append_rollout_item( + parent_path.as_path(), + ts, + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "Parent preview".to_string(), + ..Default::default() + })), + ); + write_session_meta(child_path.as_path(), child_thread_id, child_segment_id, ts); + append_rollout_item( + child_path.as_path(), + ts, + RolloutItem::RolloutReference(RolloutReferenceItem { + rollout_path: parent_path, + thread_id: Some(parent_thread_id), + rollout_timestamp: Some(ts.to_string()), + segment_id: Some(parent_segment_id), + max_depth: 2, + nth_user_message: Some(usize::MAX), + compacted_replacement_history_filter_texts: None, + }), + ); + append_rollout_item( + child_path.as_path(), + ts, + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "Child preview".to_string(), + ..Default::default() + })), + ); + + let provider_filter = provider_vec(&[TEST_PROVIDER]); + let page = get_threads( + home, + /*page_size*/ 10, + /*cursor*/ None, + ThreadSortKey::CreatedAt, + INTERACTIVE_SESSION_SOURCES.as_slice(), + Some(provider_filter.as_slice()), + /*cwd_filters*/ None, + TEST_PROVIDER, + ) + .await + .unwrap(); + + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].preview.as_deref(), Some("Child preview")); + assert_eq!( + page.items[0].first_user_message.as_deref(), + Some("Child preview") + ); +} + #[tokio::test] async fn test_list_threads_uses_goal_objective_as_preview() { let temp = TempDir::new().unwrap(); @@ -1259,6 +1715,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> { item: RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { id: conversation_id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp: ts.to_string(), diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index fbb63dceae..b9467fa134 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -23,7 +23,7 @@ pub fn apply_rollout_item( RolloutItem::EventMsg(event) => apply_event_msg(metadata, event), RolloutItem::ResponseItem(item) => apply_response_item(metadata, item), RolloutItem::InterAgentCommunication(_) => {} - RolloutItem::Compacted(_) => {} + RolloutItem::Compacted(_) | RolloutItem::RolloutReference(_) => {} } if metadata.model_provider.is_empty() { metadata.model_provider = default_provider.to_string(); @@ -40,6 +40,7 @@ pub fn rollout_item_affects_thread_metadata(item: &RolloutItem) -> bool { RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) + | RolloutItem::RolloutReference(_) | RolloutItem::Compacted(_) => false, } } @@ -321,6 +322,7 @@ mod tests { &RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { id: thread_id, + segment_id: None, forked_from_id: Some( ThreadId::from_string(&Uuid::now_v7().to_string()).expect("thread id"), ), @@ -490,6 +492,7 @@ mod tests { &RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { id: thread_id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp: "2026-02-26T00:00:00.000Z".to_string(), diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 40267f0466..a36429fca1 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -1079,6 +1079,7 @@ pub(super) fn extract_memory_mode(items: &[RolloutItem]) -> Option { RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) + | RolloutItem::RolloutReference(_) | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) => None, @@ -1753,6 +1754,7 @@ mod tests { let items = vec![RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { id: thread_id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp: metadata.created_at.to_rfc3339(), @@ -1814,6 +1816,7 @@ mod tests { let items = vec![RolloutItem::SessionMeta(SessionMetaLine { meta: SessionMeta { id: thread_id, + segment_id: None, forked_from_id: None, parent_thread_id: None, timestamp: created_at,