openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-08 20:40:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Read rollout reference chains

Browse Source
This commit is contained in:
Friel
2026-06-12 21:20:09 +00:00
parent 0f8d24a68a
commit ecf4823d12
14 changed files with 1243 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/rollout/src/compression_tests.rs
+1
View File
@@ -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(),
codex-rs/rollout/src/lib.rs
+5
View File
@@ -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<Vec<SessionSource>> = 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;
codex-rs/rollout/src/list.rs
+391 -6
View File
@@ -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<ThreadId>,
first_user_message: Option<String>,
preview: Option<String>,
@@ -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<PathBuf> },
}
/// 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-<uuid>.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<HeadTailSummary> {
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<String>, Option<String>)> {
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<codex_protocol::protocol::RolloutReferenceItem>,
)> {
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<HeadTai
summary.saw_session_meta = true;
}
}
RolloutItem::RolloutReference(item) => {
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<HeadTai
RolloutItem::TurnContext(_) => {
// 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<HeadTai
}
}
Ok(summary)
Ok((summary, reference))
}
/// Read up to `HEAD_RECORD_LIMIT` records from the start of the rollout file at `path`.
@@ -1184,7 +1305,8 @@ pub async fn read_head_for_summary(path: &Path) -> io::Result<Vec<serde_json::Va
head.push(value);
}
}
RolloutItem::Compacted(_)
RolloutItem::RolloutReference(_)
| RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_) => {}
}
@@ -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<Option<PathBuf>> {
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<Option<PathBuf>> {
// 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<Option<PathBuf>> {
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<ArchivedThreadRolloutDisposition> {
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<Option<PathBuf>> {
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<Option<PathBuf>> {
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<PathBuf> {
// 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<PathBuf> {
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();
codex-rs/rollout/src/metadata.rs
+39 -5
View File
@@ -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!(
codex-rs/rollout/src/metadata_tests.rs
+150
View File
@@ -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(&current_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(&current_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(),
codex-rs/rollout/src/policy.rs
+1
View File
@@ -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(_) => {
codex-rs/rollout/src/recorder.rs
+122 -14
View File
@@ -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<ThreadsPage> {
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<ThreadItem>,
@@ -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(_)
codex-rs/rollout/src/recorder_tests.rs
+67
View File
@@ -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();
codex-rs/rollout/src/search.rs
+1
View File
@@ -280,6 +280,7 @@ fn conversation_text_from_item(item: &RolloutItem) -> Option<String> {
}
}
RolloutItem::SessionMeta(_)
| RolloutItem::RolloutReference(_)
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_)
| RolloutItem::ResponseItem(_)
codex-rs/rollout/src/session_index_tests.rs
+1
View File
@@ -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,
codex-rs/rollout/src/state_db_tests.rs
+1
View File
@@ -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(),
codex-rs/rollout/src/tests.rs
+457
View File
@@ -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(),
codex-rs/state/src/extract.rs
+4 -1
View File
@@ -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(),
codex-rs/state/src/runtime/threads.rs
+3
View File
@@ -1079,6 +1079,7 @@ pub(super) fn extract_memory_mode(items: &[RolloutItem]) -> Option<String> {
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,