github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-20 00:18:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Distinguish rollout IDs from thread IDs (#38127)

Browse Source 
## Why

`thread/revert` preserves a thread ID while creating a new immutable rollout. Rollout persistence and filesystem fallback therefore need to distinguish the logical thread from each physical history file.

## What changed

- Add a `RolloutId` alias and canonical filename parsing for both ordinary rollouts and reverted rollouts with a distinct rollout ID.
- Allow the recorder to create a replacement rollout without changing the thread ID, and add lookup by either the stable thread ID or an exact rollout ID.
- Index history references and compression safety checks by rollout ID so multiple rollouts for one thread remain distinct.
- Preserve the rollout path selected in SQLite when filesystem repair encounters another rollout for the same thread.

## Testing

Add coverage for filename round trips, replacement rollout metadata, thread and rollout lookup, reference indexing, compression, and database path preservation.

GitOrigin-RevId: 14b1d0b4a9432b341a1307afc0abadd507e7ae7d
This commit is contained in:
Owen Lin
2026-08-12 05:20:32 +00:00
committed by copyberry
parent 2230d64464
commit 4ef836f883
18 changed files with 513 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/protocol/src/lib.rs
+1
View File
@@ -8,6 +8,7 @@ mod tool_name;
pub use agent_path::AgentPath;
pub use response_item_id::ResponseItemId;
pub use session_id::SessionId;
pub use thread_id::RolloutId;
pub use thread_id::ThreadId;
pub use tool_name::DEFAULT_FUNCTION_NAMESPACE;
pub use tool_name::ToolName;
codex-rs/protocol/src/protocol.rs
+7 -1
View File
@@ -2840,9 +2840,15 @@ impl SessionContextWindow {
}
}
/// Exclusive position in another thread's paginated rollout history.
/// Exclusive position in another rollout's paginated history.
#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq, JsonSchema, TS)]
pub struct HistoryPosition {
/// Rollout ID for the immutable prefix file.
///
/// `HistoryPosition` predates `thread/revert`, so this field is named `thread_id`. Treat its
/// value as a `rollout_id`: ordinary rollouts use the thread ID as their rollout ID, while a
/// reverted thread's filename carries a distinct rollout ID. It is not necessarily
/// [`SessionMeta::id`], which remains the stable thread ID across revert.
pub thread_id: ThreadId,
/// First rollout ordinal not included from the prefix file.
pub end_ordinal_exclusive: u64,
codex-rs/protocol/src/thread_id.rs
+7
View File
@@ -17,6 +17,13 @@ pub struct ThreadId {
pub(crate) uuid: Uuid,
}
/// Identifier encoded in a rollout filename.
///
/// Rollout IDs use the same UUID representation as thread IDs. Ordinary rollout files use the
/// thread ID as their rollout ID; \`thread/revert\` creates a new rollout file with a distinct
/// rollout ID while preserving the thread ID.
pub type RolloutId = ThreadId;
impl ThreadId {
pub fn new() -> Self {
Self {
codex-rs/rollout/src/compression.rs
+6 -1
View File
@@ -491,12 +491,17 @@ mod worker {
continue;
}
let path = rollout_file.into_path();
let Some(rollout_id) = crate::rollout_id_from_path(path.as_path()) else {
stats.skipped = stats.skipped.saturating_add(1);
metrics::file("skipped_unreadable_meta");
continue;
};
let Ok(meta) = crate::read_session_meta_line(path.as_path()).await else {
stats.skipped = stats.skipped.saturating_add(1);
metrics::file("skipped_unreadable_meta");
continue;
};
if reference_index.reference_count(meta.meta.id) > 0 {
if reference_index.reference_count(rollout_id) > 0 {
stats.skipped = stats.skipped.saturating_add(1);
metrics::file("skipped_referenced");
continue;
codex-rs/rollout/src/compression_tests.rs
+5 -5
View File
@@ -333,20 +333,20 @@ async fn worker_waits_for_rollout_maintenance_before_compressing() -> anyhow::Re
#[tokio::test]
async fn worker_skips_archived_paginated_fork_pointer_chain() -> anyhow::Result<()> {
let home = TempDir::new()?;
let thread_id = ThreadId::from_string(&Uuid::from_u128(15).to_string())?;
let source_uuid = Uuid::from_u128(16);
let source_id = ThreadId::from_string(&source_uuid.to_string())?;
let source_rollout_id = ThreadId::from_string(&source_uuid.to_string())?;
let source_path = rollout_path(home.path(), "2025-01-03T12-00-00", source_uuid);
write_rollout(&source_path, source_id, "referenced source")?;
write_rollout(&source_path, thread_id, "referenced source")?;
set_old_mtime(&source_path)?;
let child_uuid = Uuid::from_u128(17);
let child_id = ThreadId::from_string(&child_uuid.to_string())?;
let child_path = archived_rollout_path(home.path(), "2025-01-03T12-00-01", child_uuid);
write_rollout(&child_path, child_id, "fork child")?;
write_rollout(&child_path, thread_id, "fork child")?;
set_history_base(
child_path.as_path(),
HistoryPosition {
thread_id: source_id,
thread_id: source_rollout_id,
end_ordinal_exclusive: 2,
end_byte_offset: std::fs::metadata(source_path.as_path())?.len(),
},
codex-rs/rollout/src/lib.rs
+3
View File
@@ -15,6 +15,7 @@ mod persistence_metrics;
pub(crate) mod policy;
pub(crate) mod recorder;
mod reverse_jsonl_scanner;
mod rollout_file_name;
mod rollout_reference_index;
pub(crate) mod search;
pub(crate) mod session_index;
@@ -64,6 +65,7 @@ pub use list::ThreadListLayout;
pub use list::ThreadSortKey;
pub use list::ThreadsPage;
pub use list::find_archived_thread_path_by_id_str;
pub use list::find_rollout_path_by_rollout_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;
@@ -77,6 +79,7 @@ pub use list::rollout_date_parts;
pub use maintenance::RolloutMaintenanceGuard;
pub use maintenance::try_acquire_rollout_maintenance_lock;
pub use metadata::builder_from_items;
pub use metadata::rollout_id_from_path;
pub use model_context::ModelContextScan;
pub use model_context::ModelContextScanProgress;
pub use persistence_metrics::RolloutPersistenceBatchMeasurement;
codex-rs/rollout/src/list.rs
+84 -27
View File
@@ -18,11 +18,13 @@ use uuid::Uuid;
use super::ARCHIVED_SESSIONS_SUBDIR;
use super::SESSIONS_SUBDIR;
use super::compression;
use super::rollout_file_name::RolloutFileName;
use crate::RolloutItem;
use crate::RolloutLine;
use crate::protocol::EventMsg;
use crate::state_db;
use codex_file_search as file_search;
use codex_protocol::RolloutId;
use codex_protocol::ThreadId;
use codex_protocol::items::TurnItem;
use codex_protocol::protocol::SessionMetaLine;
@@ -965,21 +967,9 @@ async fn collect_rollout_day_files(
}
pub(crate) fn parse_timestamp_uuid_from_filename(name: &str) -> Option<(OffsetDateTime, Uuid)> {
// Expected: rollout-YYYY-MM-DDThh-mm-ss-<uuid>.jsonl[.zst]
let name = compression::parse_rollout_file_name(name)?;
let core = name.strip_prefix("rollout-")?.strip_suffix(".jsonl")?;
// Scan from the right for a '-' such that the suffix parses as a UUID.
let (sep_idx, uuid) = core
.match_indices('-')
.rev()
.find_map(|(i, _)| Uuid::parse_str(&core[i + 1..]).ok().map(|u| (i, u)))?;
let ts_str = &core[..sep_idx];
let format: &[FormatItem] =
format_description!("[year]-[month]-[day]T[hour]-[minute]-[second]");
let ts = PrimitiveDateTime::parse(ts_str, format).ok()?.assume_utc();
Some((ts, uuid))
let file_name = RolloutFileName::parse(name)?;
let thread_id = Uuid::parse_str(&file_name.thread_id().to_string()).ok()?;
Some((file_name.timestamp(), thread_id))
}
struct ThreadCandidate {
@@ -1418,7 +1408,7 @@ async fn find_thread_path_by_id_str_in_subdir(
if !root.exists() {
return Ok(unverified_db_path);
}
let (filename_match, filename_scan_error) = match find_rollout_path_by_id_from_filenames(
let (filename_match, filename_scan_error) = match find_thread_path_by_id_from_filenames(
root.as_path(),
id_str,
)
@@ -1493,13 +1483,41 @@ async fn find_thread_path_by_id_str_in_subdir(
Ok(found.or(unverified_db_path))
}
async fn find_rollout_path_by_id_from_filenames(
async fn find_thread_path_by_id_from_filenames(
root: &Path,
id_str: &str,
) -> io::Result<Option<PathBuf>> {
let Ok(target) = Uuid::parse_str(id_str) else {
let Ok(target) = ThreadId::from_string(id_str) else {
return Ok(None);
};
let mut newest = None;
visit_rollout_filenames::<()>(root, |file_name, path| {
if file_name.thread_id() != target {
return ControlFlow::Continue(());
}
let rollout_id = Uuid::parse_str(&file_name.rollout_id().to_string()).ok();
let Some(rollout_id) = rollout_id else {
return ControlFlow::Continue(());
};
// Rollout filenames only encode timestamps to second precision, so use the UUIDv7
// rollout ID as a deterministic tie-breaker when multiple files are created in the same
// second.
let candidate = (file_name.timestamp(), rollout_id, path);
if newest.as_ref().is_none_or(|(timestamp, id, _)| {
candidate.0 > *timestamp || (candidate.0 == *timestamp && candidate.1 > *id)
}) {
newest = Some(candidate);
}
ControlFlow::Continue(())
})
.await?;
Ok(newest.map(|(_timestamp, _id, path)| path))
}
async fn visit_rollout_filenames<T>(
root: &Path,
mut visitor: impl FnMut(RolloutFileName, PathBuf) -> ControlFlow<T>,
) -> io::Result<Option<T>> {
let mut stack = vec![root.to_path_buf()];
while let Some(dir) = stack.pop() {
let mut read_dir = match tokio::fs::read_dir(dir.as_path()).await {
@@ -1520,22 +1538,40 @@ async fn find_rollout_path_by_id_from_filenames(
let Some(rollout_file) = compression::RolloutFile::from_path(path) else {
continue;
};
let Some((_ts, id)) =
parse_timestamp_uuid_from_filename(rollout_file.plain_file_name())
else {
let Some(file_name) = RolloutFileName::parse(rollout_file.plain_file_name()) else {
continue;
};
if id == target {
return Ok(Some(rollout_file.into_path()));
if let ControlFlow::Break(found) = visitor(file_name, rollout_file.into_path()) {
return Ok(Some(found));
}
}
}
Ok(None)
}
/// Locate a recorded thread rollout file by its UUID string using the existing
/// paginated listing implementation. Returns `Ok(Some(path))` if found, `Ok(None)` if not present
/// or the id is invalid.
async fn find_rollout_path_by_rollout_id_from_filenames(
root: &Path,
rollout_id: RolloutId,
) -> io::Result<Option<PathBuf>> {
visit_rollout_filenames(root, |file_name, path| {
if file_name.rollout_id() == rollout_id {
ControlFlow::Break(path)
} else {
ControlFlow::Continue(())
}
})
.await
}
/// Locate the newest rollout file owned by a thread ID.
///
/// A thread normally has one rollout file. `thread/revert` keeps the thread ID stable while
/// creating a new rollout file and switching the thread to it, so filesystem fallback matches the
/// stable thread ID encoded before any `_rollout-id` suffix and chooses the newest matching
/// filename.
///
/// SQLite can return its selected path directly. Returns `Ok(Some(path))` if found, `Ok(None)`
/// if not present or the ID is invalid.
pub async fn find_thread_path_by_id_str(
codex_home: &Path,
id_str: &str,
@@ -1544,7 +1580,7 @@ pub async fn find_thread_path_by_id_str(
find_thread_path_by_id_str_in_subdir(codex_home, SESSIONS_SUBDIR, id_str, state_db_ctx).await
}
/// Locate an archived thread rollout file by its UUID string.
/// Locate the newest archived rollout file owned by a thread ID.
pub async fn find_archived_thread_path_by_id_str(
codex_home: &Path,
id_str: &str,
@@ -1554,6 +1590,27 @@ pub async fn find_archived_thread_path_by_id_str(
.await
}
/// Locate one immutable rollout file by its rollout ID across unarchived and archived storage.
///
/// Unlike [`find_thread_path_by_id_str`], this does not consult SQLite or choose among several
/// rollout files owned by one thread. It is for following `SessionMeta.history_base`.
pub async fn find_rollout_path_by_rollout_id(
codex_home: &Path,
rollout_id: RolloutId,
) -> io::Result<Option<PathBuf>> {
for subdir in [SESSIONS_SUBDIR, ARCHIVED_SESSIONS_SUBDIR] {
let path = find_rollout_path_by_rollout_id_from_filenames(
codex_home.join(subdir).as_path(),
rollout_id,
)
.await?;
if path.is_some() {
return Ok(path);
}
}
Ok(None)
}
/// 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
+18 -6
View File
@@ -2,14 +2,14 @@ use crate::ARCHIVED_SESSIONS_SUBDIR;
use crate::RolloutItem;
use crate::SESSIONS_SUBDIR;
use crate::compression;
use crate::list::parse_timestamp_uuid_from_filename;
use crate::recorder::RolloutRecorder;
use crate::rollout_file_name::RolloutFileName;
use crate::state_db::normalize_cwd_for_state_db;
use chrono::DateTime;
use chrono::NaiveDateTime;
use chrono::Timelike;
use chrono::Utc;
use codex_protocol::ThreadId;
use codex_protocol::RolloutId;
use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionMetaLine;
@@ -82,19 +82,31 @@ pub fn builder_from_items(
}
let file_name = rollout_path.file_name()?.to_str()?;
let file_name = compression::parse_rollout_file_name(file_name)?;
let (created_ts, uuid) = parse_timestamp_uuid_from_filename(file_name)?;
let file_name = RolloutFileName::parse(file_name)?;
let created_ts = file_name.timestamp();
let created_at =
DateTime::<Utc>::from_timestamp(created_ts.unix_timestamp(), 0)?.with_nanosecond(0)?;
let id = ThreadId::from_string(&uuid.to_string()).ok()?;
Some(ThreadMetadataBuilder::new(
id,
file_name.thread_id(),
rollout_path.to_path_buf(),
created_at,
SessionSource::default(),
))
}
/// Returns the rollout ID encoded in a canonical rollout filename.
///
/// Normal rollouts use `rollout-<timestamp>-<thread-id>.jsonl`, where the thread ID and rollout ID
/// are the same. Threads that have been `reverted` use
/// `rollout-<timestamp>-<thread-id>_<rollout-id>.jsonl`, where this returns the ID after `_`.
///
/// This can differ from [`SessionMeta::id`] when `thread/revert` keeps the thread ID stable while
/// switching to a new immutable rollout file.
pub fn rollout_id_from_path(rollout_path: &Path) -> Option<RolloutId> {
let file_name = rollout_path.file_name()?.to_str()?;
Some(RolloutFileName::parse(file_name)?.rollout_id())
}
pub async fn extract_metadata_from_rollout(
rollout_path: &Path,
default_provider: &str,
codex-rs/rollout/src/recorder.rs
+43 -43
View File
@@ -14,6 +14,7 @@ use std::sync::Arc;
use std::sync::Mutex;
use chrono::SecondsFormat;
use codex_protocol::RolloutId;
use codex_protocol::SessionId;
use codex_protocol::ThreadId;
use codex_protocol::capabilities::SelectedCapabilityRoot;
@@ -51,6 +52,7 @@ use super::list::parse_timestamp_uuid_from_filename;
use super::metadata;
use super::ordinal::RolloutOrdinalState;
use super::ordinal::ordinal_state_for_rollout;
use super::rollout_file_name::RolloutFileName;
use super::session_index::find_thread_names_by_ids;
use crate::InitialHistory;
use crate::ResumedHistory;
@@ -94,6 +96,13 @@ pub enum RolloutRecorderParams {
Create {
session_id: SessionId,
conversation_id: ThreadId,
/// Overrides the rollout ID encoded in the filename.
///
/// Normally this is `None`, so the filename is
/// `rollout-<timestamp>-<conversation_id>.jsonl`. `thread/revert` sets it, producing
/// `rollout-<timestamp>-<conversation_id>_<rollout_id>.jsonl`, because revert keeps the
/// thread ID stable while creating a new immutable rollout file.
rollout_id_override: Option<RolloutId>,
forked_from_id: Option<ThreadId>,
parent_thread_id: Option<ThreadId>,
source: Box<SessionSource>,
@@ -189,6 +198,7 @@ impl RolloutRecorderParams {
Self::Create {
session_id: conversation_id.into(),
conversation_id,
rollout_id_override: None,
forked_from_id,
parent_thread_id,
source: Box::new(source),
@@ -212,6 +222,20 @@ impl RolloutRecorderParams {
self
}
/// Override the rollout ID while preserving the thread ID.
///
/// This is for creating a new immutable rollout file for an existing thread.
pub fn with_rollout_id(mut self, rollout_id: RolloutId) -> Self {
if let Self::Create {
rollout_id_override,
..
} = &mut self
{
*rollout_id_override = Some(rollout_id);
}
self
}
pub fn with_selected_capability_roots(
mut self,
selected_capability_roots: Vec<SelectedCapabilityRoot>,
@@ -800,6 +824,7 @@ impl RolloutRecorder {
RolloutRecorderParams::Create {
session_id,
conversation_id,
rollout_id_override,
forked_from_id,
parent_thread_id,
source,
@@ -816,10 +841,8 @@ impl RolloutRecorder {
} => {
let ordinal_state =
RolloutOrdinalState::for_new_rollout(history_mode, history_base);
let log_file_info = precompute_log_file_info(config, conversation_id)?;
let path = log_file_info.path.clone();
let thread_id = log_file_info.conversation_id;
let started_at = log_file_info.timestamp;
let (path, started_at) =
precompute_new_rollout_path(config, conversation_id, rollout_id_override)?;
let timestamp_format: &[FormatItem] = format_description!(
"[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
@@ -831,7 +854,7 @@ impl RolloutRecorder {
let session_meta = SessionMeta {
session_id,
id: thread_id,
id: conversation_id,
forked_from_id,
parent_thread_id,
timestamp,
@@ -861,7 +884,7 @@ impl RolloutRecorder {
RolloutWriterState {
writer: None,
deferred_log_file_info: Some(log_file_info),
deferred_creation: true,
pending_items: Vec::new(),
meta: Some(session_meta),
cwd: cwd.clone(),
@@ -874,7 +897,7 @@ impl RolloutRecorder {
let (path, file, ordinal_state) = open_rollout_for_append(path.as_path()).await?;
RolloutWriterState {
writer: Some(JsonlWriter { file }),
deferred_log_file_info: None,
deferred_creation: false,
pending_items: Vec::new(),
meta: None,
cwd: cwd.clone(),
@@ -1575,21 +1598,11 @@ fn cursor_from_thread_item(item: &ThreadItem, sort_key: ThreadSortKey) -> Option
}
}
struct LogFileInfo {
/// Full path to the rollout file.
path: PathBuf,
/// Session ID (also embedded in filename).
conversation_id: ThreadId,
/// Timestamp for the start of the session.
timestamp: OffsetDateTime,
}
fn precompute_log_file_info(
fn precompute_new_rollout_path(
config: &impl RolloutConfigView,
conversation_id: ThreadId,
) -> std::io::Result<LogFileInfo> {
thread_id: ThreadId,
rollout_id_override: Option<RolloutId>,
) -> std::io::Result<(PathBuf, OffsetDateTime)> {
// Resolve ~/.codex/sessions/YYYY/MM/DD path.
let timestamp = OffsetDateTime::now_local()
.map_err(|e| IoError::other(format!("failed to get local time: {e}")))?;
@@ -1599,23 +1612,14 @@ fn precompute_log_file_info(
dir.push(format!("{:02}", u8::from(timestamp.month())));
dir.push(format!("{:02}", timestamp.day()));
// Custom format for YYYY-MM-DDThh-mm-ss. Use `-` instead of `:` for
// compatibility with filesystems that do not allow colons in filenames.
let format: &[FormatItem] =
format_description!("[year]-[month]-[day]T[hour]-[minute]-[second]");
let date_str = timestamp
.format(format)
let rollout_id = rollout_id_override.unwrap_or(thread_id);
let filename = RolloutFileName::new(timestamp, thread_id, rollout_id)
.render()
.map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?;
let filename = format!("rollout-{date_str}-{conversation_id}.jsonl");
let path = dir.join(filename);
Ok(LogFileInfo {
path,
conversation_id,
timestamp,
})
Ok((path, timestamp))
}
fn open_log_file(path: &Path) -> std::io::Result<File> {
@@ -1647,7 +1651,8 @@ fn open_log_file(path: &Path) -> std::io::Result<File> {
/// unwritten suffix so the next barrier can reopen the file and retry.
struct RolloutWriterState {
writer: Option<JsonlWriter>,
deferred_log_file_info: Option<LogFileInfo>,
/// True until a newly created rollout is first materialized.
deferred_creation: bool,
pending_items: Vec<RolloutItem>,
meta: Option<SessionMeta>,
cwd: PathBuf,
@@ -1716,7 +1721,7 @@ impl RolloutWriterState {
}
fn is_deferred(&self) -> bool {
self.writer.is_none() && self.deferred_log_file_info.is_some()
self.writer.is_none() && self.deferred_creation
}
fn enter_recovery_mode(&mut self, err: &IoError) {
@@ -1739,16 +1744,11 @@ impl RolloutWriterState {
return Ok(());
}
let path = self
.deferred_log_file_info
.as_ref()
.map(|info| info.path.as_path())
.unwrap_or(self.rollout_path.as_path());
let file = open_log_file(path)?;
let file = open_log_file(self.rollout_path.as_path())?;
self.writer = Some(JsonlWriter {
file: tokio::fs::File::from_std(file),
});
self.deferred_log_file_info = None;
self.deferred_creation = false;
Ok(())
}
codex-rs/rollout/src/recorder_tests.rs
+45 -69
View File
@@ -707,6 +707,50 @@ async fn referenced_paginated_rollout_starts_at_history_cutoff_and_resumes() ->
resumed.shutdown().await
}
#[tokio::test]
async fn rollout_id_preserves_session_meta_thread_id() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let thread_id = ThreadId::new();
let rollout_id = ThreadId::new();
let recorder = RolloutRecorder::new(
&config,
RolloutRecorderParams::new(
thread_id,
/*forked_from_id*/ None,
/*parent_thread_id*/ None,
SessionSource::Exec,
/*thread_source*/ None,
"test_originator".to_string(),
BaseInstructions::default(),
Vec::new(),
)
.with_history_mode(ThreadHistoryMode::Paginated)
.with_rollout_id(rollout_id),
)
.await?;
let rollout_path = recorder.rollout_path().to_path_buf();
recorder.persist().await?;
recorder.shutdown().await?;
let replacement_suffix = format!("-{thread_id}_{rollout_id}.jsonl");
assert!(
rollout_path
.file_name()
.and_then(|name| name.to_str())
.is_some_and(|name| name.ends_with(replacement_suffix.as_str()))
);
assert_eq!(
crate::rollout_id_from_path(rollout_path.as_path()),
Some(rollout_id)
);
let RolloutItem::SessionMeta(meta_line) = &read_rollout_lines(&rollout_path)?[0].item else {
panic!("first rollout item should be session metadata");
};
assert_eq!(meta_line.meta.id, thread_id);
Ok(())
}
#[tokio::test]
async fn recorder_omits_ordinals_from_legacy_rollouts() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
@@ -829,7 +873,7 @@ async fn writer_state_retries_write_error_before_reporting_flush_success() -> st
writer: Some(JsonlWriter {
file: tokio::fs::File::from_std(read_only_file),
}),
deferred_log_file_info: None,
deferred_creation: false,
pending_items: Vec::new(),
meta: None,
cwd: home.path().to_path_buf(),
@@ -1144,74 +1188,6 @@ async fn list_threads_db_enabled_preserves_metadata_for_missing_rollout_paths()
Ok(())
}
#[tokio::test]
async fn list_threads_db_enabled_repairs_stale_rollout_paths() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let uuid = Uuid::from_u128(9011);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let real_path = write_session_file(home.path(), "2025-01-03T13-00-00", uuid)?;
let stale_path = home.path().join(format!(
"sessions/2099/01/01/rollout-2099-01-01T00-00-00-{uuid}.jsonl"
));
let runtime = codex_state::StateRuntime::init(
codex_state::SqliteConfig::new_for_testing(home.path().abs()),
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, 13, 0, 0)
.single()
.expect("valid datetime");
let mut builder = codex_state::ThreadMetadataBuilder::new(
thread_id,
stale_path,
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.first_user_message = Some("Hello from user".to_string());
metadata.preview = metadata.first_user_message.clone();
runtime
.upsert_thread(&metadata)
.await
.expect("state db upsert should succeed");
let default_provider = config.model_provider_id.clone();
let page = RolloutRecorder::list_threads(
Some(runtime.clone()),
&config,
/*page_size*/ 1,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
default_provider.as_str(),
/*search_term*/ None,
)
.await?;
assert_eq!(page.items.len(), 1);
assert_eq!(page.items[0].path, real_path);
let repaired_path = runtime
.find_rollout_path_by_id(thread_id, Some(false))
.await
.expect("state db lookup should succeed");
assert_eq!(repaired_path, Some(real_path));
Ok(())
}
#[tokio::test]
async fn list_threads_state_db_only_skips_jsonl_repair_scan() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
codex-rs/rollout/src/rollout_file_name.rs
+87
View File
@@ -0,0 +1,87 @@
use codex_protocol::RolloutId;
use codex_protocol::ThreadId;
use time::OffsetDateTime;
use time::PrimitiveDateTime;
use time::format_description::FormatItem;
use time::macros::format_description;
use crate::compression;
/// Parsed canonical rollout basename.
///
/// Ordinary rollout filenames encode one ID, which is both the thread ID and rollout ID.
/// Filenames for reverted threads append an underscore and a distinct rollout ID after the stable
/// thread ID.
#[derive(Debug, PartialEq, Eq)]
pub(crate) struct RolloutFileName {
timestamp: OffsetDateTime,
thread_id: ThreadId,
rollout_id: RolloutId,
}
#[cfg(test)]
#[path = "rollout_file_name_tests.rs"]
mod tests;
impl RolloutFileName {
pub(crate) fn new(
timestamp: OffsetDateTime,
thread_id: ThreadId,
rollout_id: RolloutId,
) -> Self {
Self {
timestamp,
thread_id,
rollout_id,
}
}
pub(crate) fn parse(name: &str) -> Option<Self> {
let name = compression::parse_rollout_file_name(name)?;
let core = name.strip_prefix("rollout-")?.strip_suffix(".jsonl")?;
let timestamp = core.get(..19)?;
if core.get(19..20)? != "-" {
return None;
}
let ids = core.get(20..)?;
let (thread_id, rollout_id) = ids.split_once('_').unwrap_or((ids, ids));
let thread_id = ThreadId::from_string(thread_id).ok()?;
let rollout_id = RolloutId::from_string(rollout_id).ok()?;
let format: &[FormatItem] =
format_description!("[year]-[month]-[day]T[hour]-[minute]-[second]");
let timestamp = PrimitiveDateTime::parse(timestamp, format)
.ok()?
.assume_utc();
Some(Self {
timestamp,
thread_id,
rollout_id,
})
}
pub(crate) fn render(&self) -> Result<String, time::error::Format> {
let format: &[FormatItem] =
format_description!("[year]-[month]-[day]T[hour]-[minute]-[second]");
let timestamp = self.timestamp.format(format)?;
Ok(if self.thread_id == self.rollout_id {
format!("rollout-{timestamp}-{}.jsonl", self.thread_id)
} else {
format!(
"rollout-{timestamp}-{}_{}.jsonl",
self.thread_id, self.rollout_id
)
})
}
pub(crate) fn timestamp(&self) -> OffsetDateTime {
self.timestamp
}
pub(crate) fn thread_id(&self) -> ThreadId {
self.thread_id
}
pub(crate) fn rollout_id(&self) -> RolloutId {
self.rollout_id
}
}
codex-rs/rollout/src/rollout_file_name_tests.rs
+29
View File
@@ -0,0 +1,29 @@
use codex_protocol::ThreadId;
use pretty_assertions::assert_eq;
use time::macros::datetime;
use super::RolloutFileName;
#[test]
fn canonical_rollout_file_names_round_trip() {
let timestamp = datetime!(2026-08-11 18:42:07 UTC);
let thread_id =
ThreadId::from_string("019ff1a2-b3c4-7d5e-8f60-112233445566").expect("valid thread id");
let rollout_id =
ThreadId::from_string("019ff1a2-b3c4-7d5e-8f60-667788990011").expect("valid rollout id");
for (file_name, expected) in [
(
RolloutFileName::new(timestamp, thread_id, thread_id),
"rollout-2026-08-11T18-42-07-019ff1a2-b3c4-7d5e-8f60-112233445566.jsonl",
),
(
RolloutFileName::new(timestamp, thread_id, rollout_id),
"rollout-2026-08-11T18-42-07-019ff1a2-b3c4-7d5e-8f60-112233445566_019ff1a2-b3c4-7d5e-8f60-667788990011.jsonl",
),
] {
let rendered = file_name.render().expect("timestamp should format");
assert_eq!(rendered, expected);
assert_eq!(RolloutFileName::parse(&rendered), Some(file_name));
}
}
codex-rs/rollout/src/rollout_reference_index.rs
+24 -22
View File
@@ -7,7 +7,7 @@ use std::path::Path;
use std::time::Duration;
use std::time::Instant;
use codex_protocol::ThreadId;
use codex_protocol::RolloutId;
use codex_protocol::protocol::HistoryPosition;
use crate::ARCHIVED_SESSIONS_SUBDIR;
@@ -16,12 +16,12 @@ use crate::compression::RolloutFile;
/// Direct history-base edges discovered from local rollout metadata.
///
/// This is a physical-rollout index, not a logical lineage resolver. Callers use it to answer
/// This indexes immutable rollout IDs, not a thread's selected lineage. Callers use it to answer
/// cheap inverse-reference questions without each reimplementing rollout discovery.
#[derive(Debug, Default)]
pub struct RolloutReferenceIndex {
history_base_by_thread: HashMap<ThreadId, HistoryPosition>,
reference_counts_by_thread: HashMap<ThreadId, usize>,
history_base_by_rollout: HashMap<RolloutId, HistoryPosition>,
reference_counts_by_rollout: HashMap<RolloutId, usize>,
}
impl RolloutReferenceIndex {
@@ -54,25 +54,25 @@ impl RolloutReferenceIndex {
.await
}
/// Returns how many other discovered rollouts directly reference thread_id.
pub fn reference_count(&self, thread_id: ThreadId) -> usize {
self.reference_counts_by_thread
.get(&thread_id)
/// Returns how many other discovered rollouts directly reference `rollout_id`.
pub fn reference_count(&self, rollout_id: RolloutId) -> usize {
self.reference_counts_by_rollout
.get(&rollout_id)
.copied()
.unwrap_or_default()
}
/// Returns the direct history-base edge for thread_id, if one was discovered.
pub fn history_base(&self, thread_id: ThreadId) -> Option<&HistoryPosition> {
self.history_base_by_thread.get(&thread_id)
/// Returns the direct history-base edge for `rollout_id`, if one was discovered.
pub fn history_base(&self, rollout_id: RolloutId) -> Option<&HistoryPosition> {
self.history_base_by_rollout.get(&rollout_id)
}
async fn scan_with_deadline(
codex_home: &Path,
deadline: ScanDeadline,
) -> io::Result<Option<Self>> {
let mut history_base_by_thread = HashMap::new();
let mut seen_thread_ids = HashSet::new();
let mut history_base_by_rollout = HashMap::new();
let mut seen_rollout_ids = HashSet::new();
let mut stack = vec![
codex_home.join(ARCHIVED_SESSIONS_SUBDIR),
codex_home.join(SESSIONS_SUBDIR),
@@ -105,31 +105,33 @@ impl RolloutReferenceIndex {
let Some(rollout_file) = RolloutFile::from_path(path) else {
continue;
};
let Some(rollout_id) = crate::rollout_id_from_path(rollout_file.path()) else {
continue;
};
let Ok(meta) = crate::read_session_meta_line(rollout_file.path()).await else {
continue;
};
let thread_id = meta.meta.id;
if !seen_thread_ids.insert(thread_id) {
if !seen_rollout_ids.insert(rollout_id) {
continue;
}
if let Some(history_base) = meta.meta.history_base {
history_base_by_thread.insert(thread_id, history_base);
history_base_by_rollout.insert(rollout_id, history_base);
}
}
}
let mut reference_counts_by_thread = HashMap::new();
for (thread_id, history_base) in &history_base_by_thread {
if history_base.thread_id == *thread_id {
let mut reference_counts_by_rollout = HashMap::new();
for (rollout_id, history_base) in &history_base_by_rollout {
if history_base.thread_id == *rollout_id {
continue;
}
*reference_counts_by_thread
*reference_counts_by_rollout
.entry(history_base.thread_id)
.or_default() += 1;
}
Ok(Some(Self {
history_base_by_thread,
reference_counts_by_thread,
history_base_by_rollout,
reference_counts_by_rollout,
}))
}
}
codex-rs/rollout/src/rollout_reference_index_tests.rs
+30 -2
View File
@@ -80,6 +80,34 @@ async fn active_duplicate_wins_without_double_counting() -> anyhow::Result<()> {
Ok(())
}
#[tokio::test]
async fn indexes_multiple_rollouts_for_the_same_thread() -> anyhow::Result<()> {
let home = TempDir::new()?;
let source_rollout_id = thread_id(Uuid::from_u128(21))?;
let replacement_rollout_id = thread_id(Uuid::from_u128(22))?;
let thread_id = thread_id(Uuid::from_u128(20))?;
let history_base = history_position(source_rollout_id);
write_rollout(
active_rollout_path(home.path(), Uuid::from_u128(21)),
thread_id,
Some(history_base),
)?;
write_rollout(
active_rollout_path(home.path(), Uuid::from_u128(22)),
thread_id,
Some(history_base),
)?;
let index = RolloutReferenceIndex::scan(home.path()).await?;
assert_eq!(index.history_base(source_rollout_id), Some(&history_base));
assert_eq!(
index.history_base(replacement_rollout_id),
Some(&history_base)
);
assert_eq!(index.reference_count(source_rollout_id), 1);
Ok(())
}
#[tokio::test]
async fn self_history_base_does_not_count_as_reference() -> anyhow::Result<()> {
let home = TempDir::new()?;
@@ -144,9 +172,9 @@ fn write_rollout(
Ok(())
}
fn history_position(thread_id: ThreadId) -> HistoryPosition {
fn history_position(rollout_id: ThreadId) -> HistoryPosition {
HistoryPosition {
thread_id,
thread_id: rollout_id,
end_ordinal_exclusive: 2,
end_byte_offset: 100,
}
codex-rs/rollout/src/state_db.rs
+15
View File
@@ -554,6 +554,15 @@ pub async fn reconcile_rollout(
let memory_mode = outcome.memory_mode.unwrap_or_else(|| "enabled".to_string());
metadata.cwd = normalize_cwd_for_state_db(&metadata.cwd);
let existing_metadata = ctx.get_thread(metadata.id).await.ok().flatten();
// Filesystem repair may seed a missing row, but it must not change an existing row's
// selected rollout path. After `thread/revert`, a scan can find multiple immutable
// rollouts for one thread and cannot know which one SQLite selected.
if existing_metadata
.as_ref()
.is_some_and(|existing| existing.rollout_path.as_path() != rollout_path)
{
return;
}
// Paginated metadata updates are SQLite-only. Use the rollout mode to seed a
// missing row, then keep the value from SQLite.
let restore_memory_mode_from_rollout =
@@ -607,6 +616,12 @@ pub async fn read_repair_rollout_path(
if let Some(thread_id) = thread_id
&& let Ok(Some(metadata)) = ctx.get_thread(thread_id).await
{
// Filesystem repair may seed a missing row, but it must not change an existing row's
// selected rollout path. After `thread/revert`, a scan can find multiple immutable
// rollouts for one thread and cannot know which one SQLite selected.
if metadata.rollout_path.as_path() != rollout_path {
return;
}
saw_existing_metadata = true;
let mut repaired = metadata.clone();
repaired.rollout_path = rollout_path.to_path_buf();
codex-rs/rollout/src/state_db_tests.rs
+60
View File
@@ -199,6 +199,66 @@ async fn reconcile_rollout_preserves_existing_explicit_title() -> anyhow::Result
Ok(())
}
#[tokio::test]
async fn filesystem_repair_preserves_existing_rollout_path() -> anyhow::Result<()> {
let home = TempDir::new().expect("temp dir");
let thread_id = ThreadId::new();
let initial_rollout_path = write_rollout_with_user_message(
home.path(),
thread_id,
"Old",
ThreadHistoryMode::Paginated,
)?;
let old_rollout_path = initial_rollout_path
.with_file_name(format!("rollout-2026-06-01T14-26-25-{thread_id}_old.jsonl"));
std::fs::rename(initial_rollout_path, old_rollout_path.as_path())?;
let active_rollout_path = write_rollout_with_user_message(
home.path(),
thread_id,
"Current",
ThreadHistoryMode::Paginated,
)?;
let runtime = codex_state::StateRuntime::init(
codex_state::SqliteConfig::new_for_testing(home.path().abs()),
"test-provider".to_string(),
)
.await?;
let active_metadata =
metadata::extract_metadata_from_rollout(active_rollout_path.as_path(), "test-provider")
.await?
.metadata;
runtime.upsert_thread(&active_metadata).await?;
let active_metadata = runtime
.get_thread(thread_id)
.await?
.expect("thread should exist");
read_repair_rollout_path(
Some(runtime.as_ref()),
Some(thread_id),
/*archived_only*/ None,
old_rollout_path.as_path(),
)
.await;
assert_eq!(
runtime.get_thread(thread_id).await?,
Some(active_metadata.clone())
);
reconcile_rollout(
Some(runtime.as_ref()),
old_rollout_path.as_path(),
"test-provider",
/*builder*/ None,
&[],
/*archived_only*/ None,
/*new_thread_memory_mode*/ None,
)
.await;
assert_eq!(runtime.get_thread(thread_id).await?, Some(active_metadata));
Ok(())
}
#[tokio::test]
async fn reconcile_rollout_preserves_existing_paginated_memory_mode() -> anyhow::Result<()> {
let home = TempDir::new().expect("temp dir");
codex-rs/rollout/src/tests.rs
+49 -2
View File
@@ -23,6 +23,7 @@ use crate::INTERACTIVE_SESSION_SOURCES;
use crate::ResponseItemEnvelope;
use crate::RolloutItem;
use crate::RolloutLine;
use crate::find_rollout_path_by_rollout_id;
use crate::find_thread_path_by_id_str;
use crate::list::Cursor;
use crate::list::ThreadItem;
@@ -132,7 +133,53 @@ async fn find_thread_path_falls_back_when_db_path_is_stale() {
.await
.expect("lookup should succeed");
assert_eq!(found, Some(fs_rollout_path.clone()));
assert_state_db_rollout_path(home, thread_id, Some(fs_rollout_path.as_path())).await;
assert_state_db_rollout_path(home, thread_id, Some(stale_db_path.as_path())).await;
}
#[tokio::test]
async fn filesystem_lookup_distinguishes_thread_ids_from_rollout_ids() {
let temp = TempDir::new().unwrap();
let home = temp.path();
let thread_uuid = Uuid::from_u128(401);
let rollout_uuid = Uuid::from_u128(402);
let original_ts = "2025-01-03T13-00-00";
let reverted_ts = "2025-01-04T13-00-00";
write_session_file(
home,
original_ts,
thread_uuid,
/*num_records*/ 1,
Some(SessionSource::Cli),
)
.unwrap();
write_session_file(
home,
reverted_ts,
thread_uuid,
/*num_records*/ 1,
Some(SessionSource::Cli),
)
.unwrap();
let reverted_source = home.join(format!(
"sessions/2025/01/04/rollout-{reverted_ts}-{thread_uuid}.jsonl"
));
let reverted_path = home.join(format!(
"sessions/2025/01/04/rollout-{reverted_ts}-{thread_uuid}_{rollout_uuid}.jsonl"
));
fs::rename(reverted_source, reverted_path.as_path()).unwrap();
assert_eq!(
find_thread_path_by_id_str(home, &thread_uuid.to_string(), /*state_db_ctx*/ None)
.await
.unwrap(),
Some(reverted_path.clone())
);
assert_eq!(
find_rollout_path_by_rollout_id(home, thread_id_from_uuid(rollout_uuid))
.await
.unwrap(),
Some(reverted_path)
);
}
#[tokio::test]
@@ -212,7 +259,7 @@ async fn find_thread_path_falls_back_when_db_path_points_to_another_thread() {
.await
.expect("lookup should succeed");
assert_eq!(found, Some(fs_rollout_path.clone()));
assert_state_db_rollout_path(home, thread_id, Some(fs_rollout_path.as_path())).await;
assert_state_db_rollout_path(home, thread_id, Some(stale_db_path.as_path())).await;
}
#[tokio::test]