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

Add ordinals to paginated rollout records (#32332)

Browse Source 
## Why

Paginated thread history needs durable ordering so consumers can process a rollout suffix without rebuilding all earlier history.

## What changed

- Add optional, zero-based ordinals to `RolloutLine` records in paginated rollouts while leaving legacy rollout serialization unchanged.
- Continue ordinals from the last valid record when appending or resuming, including after gaps or an incomplete tail, and reject overflow without appending.
- Add a stateless `project_rollout_line` helper that maps canonical turn lifecycle and completed-item records into thread-history change sets.

## Testing

- Cover ordinal assignment, legacy compatibility, resume and tail recovery, overflow handling, and thread-history projection for completed, failed, and interrupted turns.

GitOrigin-RevId: 3a9bb6cd2a1f674a9f154342e96e9aa3d8330781
This commit is contained in:
Owen Lin
2026-07-11 04:15:42 +00:00
committed by copyberry
parent f93c18ed0f
commit 5c19155cbd
19 changed files with 764 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/src/lib.rs
+1
View File
@@ -16,6 +16,7 @@ pub use protocol::common::*;
pub use protocol::event_mapping::*;
pub use protocol::item_builders::*;
pub use protocol::thread_history::*;
pub use protocol::thread_history_projection::*;
pub use protocol::v1::ApplyPatchApprovalParams;
pub use protocol::v1::ApplyPatchApprovalResponse;
pub use protocol::v1::ClientInfo;
codex-rs/app-server-protocol/src/protocol/mod.rs
+1
View File
@@ -7,5 +7,6 @@ pub mod item_builders;
mod mappers;
mod serde_helpers;
pub mod thread_history;
pub mod thread_history_projection;
pub mod v1;
pub mod v2;
codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs
+89
View File
@@ -0,0 +1,89 @@
//! Stateless projection from canonical paginated rollout records to thread-history changes.
//!
//! This module is only for the new paginated rollout format that persists canonical
//! `ItemCompleted(TurnItem)` records, not legacy event-only rollouts.
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use crate::protocol::thread_history::ThreadHistoryChangeSet;
use crate::protocol::thread_history::ThreadHistoryItemChange;
use crate::protocol::thread_history::ThreadHistoryTurnChange;
use crate::protocol::v2::ThreadItem;
use crate::protocol::v2::TurnError;
use crate::protocol::v2::TurnStatus;
/// Project one durable rollout line without reconstructing earlier history.
///
/// Callers that replay a JSONL suffix should invoke it once per line, in ordinal order, so storage
/// can preserve the first and latest timestamps for repeated item snapshots independently.
pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet {
match &line.item {
RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
turn_id: event.turn_id.clone(),
status: TurnStatus::InProgress,
error: None,
started_at: event.started_at,
completed_at: None,
duration_ms: None,
}],
..Default::default()
},
RolloutItem::EventMsg(EventMsg::TurnComplete(event)) => ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
turn_id: event.turn_id.clone(),
status: if event.error.is_some() {
TurnStatus::Failed
} else {
TurnStatus::Completed
},
error: event.error.as_ref().map(|error| TurnError {
message: error.message.clone(),
codex_error_info: error.codex_error_info.clone().map(Into::into),
additional_details: None,
}),
started_at: event.started_at,
completed_at: event.completed_at,
duration_ms: event.duration_ms,
}],
..Default::default()
},
RolloutItem::EventMsg(EventMsg::TurnAborted(event)) => {
let Some(turn_id) = event.turn_id.as_ref() else {
return ThreadHistoryChangeSet::default();
};
ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
turn_id: turn_id.clone(),
status: TurnStatus::Interrupted,
error: None,
started_at: event.started_at,
completed_at: event.completed_at,
duration_ms: event.duration_ms,
}],
..Default::default()
}
}
RolloutItem::EventMsg(EventMsg::ItemCompleted(event)) => ThreadHistoryChangeSet {
changed_items: vec![ThreadHistoryItemChange {
turn_id: event.turn_id.clone(),
item: ThreadItem::from(event.item.clone()),
}],
..Default::default()
},
RolloutItem::SessionMeta(_)
| RolloutItem::ResponseItem(_)
| RolloutItem::InterAgentCommunication(_)
| RolloutItem::InterAgentCommunicationMetadata { .. }
| RolloutItem::Compacted(_)
| RolloutItem::TurnContext(_)
| RolloutItem::WorldState(_)
| RolloutItem::EventMsg(_) => ThreadHistoryChangeSet::default(),
}
}
#[cfg(test)]
#[path = "thread_history_projection_tests.rs"]
mod tests;
codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs
+211
View File
@@ -0,0 +1,211 @@
use codex_protocol::ThreadId;
use codex_protocol::items::AgentMessageContent;
use codex_protocol::items::AgentMessageItem;
use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::protocol::CompactedItem;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::protocol::TurnAbortedEvent;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnStartedEvent;
use codex_protocol::user_input::UserInput;
use pretty_assertions::assert_eq;
use super::*;
use crate::protocol::v2::ThreadItem;
use crate::protocol::v2::TurnError;
#[test]
fn projects_turn_lifecycle_without_prior_builder_state() {
let started = project(RolloutItem::EventMsg(EventMsg::TurnStarted(
TurnStartedEvent {
turn_id: "turn-1".to_string(),
trace_id: None,
started_at: Some(10),
model_context_window: None,
collaboration_mode_kind: Default::default(),
},
)));
let completed = project(RolloutItem::EventMsg(EventMsg::TurnComplete(
TurnCompleteEvent {
turn_id: "turn-1".to_string(),
last_agent_message: None,
error: None,
started_at: Some(10),
completed_at: Some(20),
duration_ms: Some(10_000),
time_to_first_token_ms: None,
},
)));
assert_eq!(started.changed_turns.len(), 1);
assert_eq!(started.changed_turns[0].turn_id, "turn-1");
assert_eq!(started.changed_turns[0].status, TurnStatus::InProgress);
assert_eq!(started.changed_turns[0].started_at, Some(10));
assert_eq!(
completed,
ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
turn_id: "turn-1".to_string(),
status: TurnStatus::Completed,
error: None,
started_at: Some(10),
completed_at: Some(20),
duration_ms: Some(10_000),
}],
..Default::default()
}
);
}
#[test]
fn projects_failed_turn_completion_as_snapshot() {
let error = ErrorEvent {
message: "request failed".to_string(),
codex_error_info: None,
};
let changes = project(RolloutItem::EventMsg(EventMsg::TurnComplete(
TurnCompleteEvent {
turn_id: "turn-1".to_string(),
last_agent_message: None,
error: Some(error),
started_at: Some(10),
completed_at: Some(20),
duration_ms: Some(10_000),
time_to_first_token_ms: None,
},
)));
assert_eq!(
changes,
ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
turn_id: "turn-1".to_string(),
status: TurnStatus::Failed,
error: Some(TurnError {
message: "request failed".to_string(),
codex_error_info: None,
additional_details: None,
}),
started_at: Some(10),
completed_at: Some(20),
duration_ms: Some(10_000),
}],
..Default::default()
}
);
}
#[test]
fn projects_completed_canonical_turn_items() {
let thread_id = ThreadId::default();
let user_item = TurnItem::UserMessage(UserMessageItem {
id: "user-1".to_string(),
client_id: None,
content: vec![UserInput::Text {
text: "hello".to_string(),
text_elements: Vec::new(),
}],
});
let agent_item = TurnItem::AgentMessage(AgentMessageItem {
id: "agent-1".to_string(),
content: vec![AgentMessageContent::Text {
text: "done".to_string(),
}],
phase: None,
memory_citation: None,
});
let user_changes = project(item_completed(thread_id, "turn-1", user_item.clone()));
let agent_changes = project(item_completed(thread_id, "turn-1", agent_item.clone()));
assert_eq!(
user_changes.changed_items,
vec![ThreadHistoryItemChange {
turn_id: "turn-1".to_string(),
item: ThreadItem::from(user_item),
}]
);
assert_eq!(
agent_changes.changed_items,
vec![ThreadHistoryItemChange {
turn_id: "turn-1".to_string(),
item: ThreadItem::from(agent_item),
}]
);
}
#[test]
fn ignores_legacy_abort_without_turn_id_and_context_only_records() {
let aborted = project(RolloutItem::EventMsg(EventMsg::TurnAborted(
TurnAbortedEvent {
turn_id: None,
reason: TurnAbortReason::Interrupted,
started_at: None,
completed_at: None,
duration_ms: None,
},
)));
let compacted = project(RolloutItem::Compacted(CompactedItem {
message: String::new(),
replacement_history: None,
window_number: None,
first_window_id: None,
previous_window_id: None,
window_id: None,
}));
assert!(aborted.is_empty());
assert!(compacted.is_empty());
}
#[test]
fn projects_identified_turn_aborts() {
let changes = project(RolloutItem::EventMsg(EventMsg::TurnAborted(
TurnAbortedEvent {
turn_id: Some("turn-1".to_string()),
reason: TurnAbortReason::Interrupted,
started_at: Some(10),
completed_at: Some(20),
duration_ms: Some(10_000),
},
)));
assert_eq!(
changes,
ThreadHistoryChangeSet {
changed_turns: vec![ThreadHistoryTurnChange {
turn_id: "turn-1".to_string(),
status: TurnStatus::Interrupted,
error: None,
started_at: Some(10),
completed_at: Some(20),
duration_ms: Some(10_000),
}],
..Default::default()
}
);
}
fn project(item: RolloutItem) -> ThreadHistoryChangeSet {
project_rollout_line(&RolloutLine {
timestamp: "2026-07-09T00:00:00.000Z".to_string(),
ordinal: Some(7),
item,
})
}
fn item_completed(thread_id: ThreadId, turn_id: &str, item: TurnItem) -> RolloutItem {
RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
thread_id,
turn_id: turn_id.to_string(),
item,
completed_at_ms: 123,
}))
}
codex-rs/app-server/src/request_processors/thread_processor_tests.rs
+3
View File
@@ -1015,6 +1015,7 @@ mod thread_processor_behavior_tests {
let line = RolloutLine {
timestamp: timestamp.clone(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: session_meta.clone(),
git: None,
@@ -1082,6 +1083,7 @@ mod thread_processor_behavior_tests {
let line = RolloutLine {
timestamp,
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: session_meta,
git: None,
@@ -1124,6 +1126,7 @@ mod thread_processor_behavior_tests {
let line = RolloutLine {
timestamp,
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: session_meta,
git: None,
codex-rs/cli/src/doctor/thread_inventory.rs
+1
View File
@@ -819,6 +819,7 @@ mod tests {
let parsed_thread_id = ThreadId::from_string(thread_id).expect("thread id");
let rollout_line = RolloutLine {
timestamp: timestamp.to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(codex_protocol::protocol::SessionMetaLine {
meta: codex_protocol::protocol::SessionMeta {
session_id: parsed_thread_id.into(),
codex-rs/core/tests/suite/client.rs
+11
View File
@@ -371,6 +371,7 @@ async fn synthetic_call_output_id_is_stable_across_resumes() -> anyhow::Result<(
let rollout = vec![
RolloutLine {
timestamp: "2024-01-01T00:00:00.000Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
@@ -388,6 +389,7 @@ async fn synthetic_call_output_id_is_stable_across_resumes() -> anyhow::Result<(
},
RolloutLine {
timestamp: "2024-01-01T00:00:01.000Z".to_string(),
ordinal: None,
item: RolloutItem::ResponseItem(ResponseItem::FunctionCall {
id: Some(ResponseItemId::with_suffix("fc", "existing")),
name: "do_it".to_string(),
@@ -896,6 +898,7 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() {
let rollout = vec![
RolloutLine {
timestamp: "2024-01-01T00:00:00.000Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
@@ -913,10 +916,12 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() {
},
RolloutLine {
timestamp: "2024-01-01T00:00:01.000Z".to_string(),
ordinal: None,
item: RolloutItem::ResponseItem(legacy_custom_tool_call),
},
RolloutLine {
timestamp: "2024-01-01T00:00:02.000Z".to_string(),
ordinal: None,
item: RolloutItem::ResponseItem(ResponseItem::CustomToolCallOutput {
id: None,
call_id: "legacy-js-call".to_string(),
@@ -927,6 +932,7 @@ async fn resume_replays_legacy_js_repl_image_rollout_shapes() {
},
RolloutLine {
timestamp: "2024-01-01T00:00:03.000Z".to_string(),
ordinal: None,
item: RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "user".to_string(),
@@ -1032,6 +1038,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
let rollout = vec![
RolloutLine {
timestamp: "2024-01-01T00:00:00.000Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
@@ -1049,6 +1056,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
},
RolloutLine {
timestamp: "2024-01-01T00:00:01.000Z".to_string(),
ordinal: None,
item: RolloutItem::ResponseItem(ResponseItem::FunctionCall {
id: None,
name: "view_image".to_string(),
@@ -1060,6 +1068,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
},
RolloutLine {
timestamp: "2024-01-01T00:00:01.500Z".to_string(),
ordinal: None,
item: RolloutItem::ResponseItem(ResponseItem::FunctionCallOutput {
id: None,
call_id: function_call_id.to_string(),
@@ -1074,6 +1083,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
},
RolloutLine {
timestamp: "2024-01-01T00:00:02.000Z".to_string(),
ordinal: None,
item: RolloutItem::ResponseItem(ResponseItem::CustomToolCall {
id: None,
status: Some("completed".to_string()),
@@ -1086,6 +1096,7 @@ async fn resume_replays_image_tool_outputs_with_detail() {
},
RolloutLine {
timestamp: "2024-01-01T00:00:02.500Z".to_string(),
ordinal: None,
item: RolloutItem::ResponseItem(ResponseItem::CustomToolCallOutput {
id: None,
call_id: custom_call_id.to_string(),
codex-rs/core/tests/suite/sqlite_state.rs
+2
View File
@@ -384,10 +384,12 @@ async fn backfill_scans_existing_rollouts() -> Result<()> {
let lines = [
RolloutLine {
timestamp: "2026-01-27T12:00:00Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(session_meta_line),
},
RolloutLine {
timestamp: "2026-01-27T12:00:01Z".to_string(),
ordinal: None,
item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "hello from backfill".to_string(),
codex-rs/memories/write/src/startup_tests.rs
+1
View File
@@ -738,6 +738,7 @@ async fn seed_stage1_candidate(
let rollout_path = codex_home.join(format!("rollout-{thread_id}.jsonl"));
let line = RolloutLine {
timestamp: updated_at.to_rfc3339(),
ordinal: None,
item: RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "user".to_string(),
codex-rs/protocol/src/protocol.rs
+2
View File
@@ -3334,6 +3334,8 @@ impl Mul<f64> for TruncationPolicy {
#[derive(Serialize, Deserialize, Clone, JsonSchema)]
pub struct RolloutLine {
pub timestamp: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub ordinal: Option<u64>,
#[serde(flatten)]
pub item: RolloutItem,
}
codex-rs/rollout/src/compression_tests.rs
+2
View File
@@ -483,10 +483,12 @@ fn write_rollout(path: &std::path::Path, thread_id: ThreadId, message: &str) ->
let lines = [
RolloutLine {
timestamp: "2025-01-03T12:00:00Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(session_meta_line),
},
RolloutLine {
timestamp: "2025-01-03T12:00:01Z".to_string(),
ordinal: None,
item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: message.to_string(),
..Default::default()
codex-rs/rollout/src/lib.rs
+1
View File
@@ -8,6 +8,7 @@ pub(crate) mod compression;
pub(crate) mod config;
pub(crate) mod list;
pub(crate) mod metadata;
mod ordinal;
mod persistence_metrics;
pub(crate) mod policy;
pub(crate) mod recorder;
codex-rs/rollout/src/metadata_tests.rs
+5
View File
@@ -62,6 +62,7 @@ async fn extract_metadata_from_rollout_uses_session_meta() {
};
let rollout_line = RolloutLine {
timestamp: "2026-01-27T12:34:56Z".to_string(),
ordinal: Some(0),
item: RolloutItem::SessionMeta(session_meta_line.clone()),
};
let json = serde_json::to_string(&rollout_line).expect("rollout json");
@@ -93,6 +94,7 @@ async fn extract_metadata_from_rollout_rejects_unknown_history_mode() {
.join(format!("rollout-2026-01-27T12-34-56-{uuid}.jsonl"));
let mut rollout_line = serde_json::to_value(RolloutLine {
timestamp: "2026-01-27T12:34:56Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: id.into(),
@@ -158,6 +160,7 @@ async fn extract_metadata_from_rollout_returns_latest_memory_mode() {
let lines = vec![
RolloutLine {
timestamp: "2026-01-27T12:34:56Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: session_meta,
git: None,
@@ -165,6 +168,7 @@ async fn extract_metadata_from_rollout_returns_latest_memory_mode() {
},
RolloutLine {
timestamp: "2026-01-27T12:35:00Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: polluted_meta,
git: None,
@@ -427,6 +431,7 @@ fn write_rollout_in_sessions_with_cwd(
};
let rollout_line = RolloutLine {
timestamp: event_ts.to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(session_meta_line),
};
let json = serde_json::to_string(&rollout_line).expect("serialize rollout");
codex-rs/rollout/src/ordinal.rs
+108
View File
@@ -0,0 +1,108 @@
use std::fs::File;
use std::io;
use std::io::BufRead;
use std::io::BufReader;
use std::io::Seek;
use std::io::SeekFrom;
use std::path::Path;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::ThreadHistoryMode;
use crate::reverse_jsonl_scanner::ReverseJsonlScanner;
use crate::reverse_jsonl_scanner::ScanOutcome;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum RolloutOrdinalState {
Legacy,
Paginated { next: Option<u64> },
}
impl RolloutOrdinalState {
pub(crate) fn for_new_rollout(history_mode: ThreadHistoryMode) -> Self {
match history_mode {
ThreadHistoryMode::Legacy => Self::Legacy,
ThreadHistoryMode::Paginated => Self::Paginated { next: Some(0) },
}
}
pub(crate) fn current(&self) -> io::Result<Option<u64>> {
match self {
Self::Legacy => Ok(None),
Self::Paginated { next } => {
let ordinal = (*next)
.ok_or_else(|| io::Error::other("paginated rollout record ordinal overflow"))?;
Ok(Some(ordinal))
}
}
}
pub(crate) fn advance(&mut self) {
if let Self::Paginated { next } = self
&& let Some(ordinal) = *next
{
*next = ordinal.checked_add(1);
}
}
}
pub(crate) fn ordinal_state_for_rollout(
file: &mut File,
path: &Path,
) -> io::Result<RolloutOrdinalState> {
let Some(history_mode) = read_history_mode(file, path)? else {
return Ok(RolloutOrdinalState::Legacy);
};
if matches!(history_mode, ThreadHistoryMode::Legacy) {
return Ok(RolloutOrdinalState::Legacy);
}
let mut scanner = ReverseJsonlScanner::new(file)?;
let record = loop {
match scanner.scan_next::<RolloutLine>()? {
Some(ScanOutcome::Parsed(record)) => break record,
Some(ScanOutcome::Rejected(_)) => continue,
None => {
return Err(io::Error::other(format!(
"rollout at {} contains no valid records",
path.display()
)));
}
}
};
let ordinal = record.ordinal.ok_or_else(|| {
io::Error::other(format!(
"final paginated rollout record at {} is missing an ordinal",
path.display()
))
})?;
Ok(RolloutOrdinalState::Paginated {
next: ordinal.checked_add(1),
})
}
fn read_history_mode(file: &mut File, path: &Path) -> io::Result<Option<ThreadHistoryMode>> {
file.seek(SeekFrom::Start(0))?;
let reader = BufReader::new(file);
for line in reader.lines() {
let line = line?;
if line.trim().is_empty() {
continue;
}
let record: RolloutLine = serde_json::from_str(line.as_str()).map_err(|error| {
io::Error::other(format!(
"failed to parse first rollout record at {}: {error}",
path.display()
))
})?;
let RolloutItem::SessionMeta(session_meta) = record.item else {
return Err(io::Error::other(format!(
"rollout at {} does not start with session metadata",
path.display()
)));
};
return Ok(Some(session_meta.meta.history_mode));
}
Ok(None)
}
codex-rs/rollout/src/recorder.rs
+74 -55
View File
@@ -49,6 +49,8 @@ use super::list::get_threads_in_root;
use super::list::parse_cursor;
use super::list::parse_timestamp_uuid_from_filename;
use super::metadata;
use super::ordinal::RolloutOrdinalState;
use super::ordinal::ordinal_state_for_rollout;
use super::session_index::find_thread_names_by_ids;
use crate::config::RolloutConfigView;
use crate::state_db;
@@ -755,7 +757,9 @@ impl RolloutRecorder {
config: &impl RolloutConfigView,
params: RolloutRecorderParams,
) -> std::io::Result<Self> {
let (file, deferred_log_file_info, rollout_path, meta) = match params {
// Clone the cwd for the spawned task to collect git info asynchronously.
let cwd = config.cwd().to_path_buf();
let state = match params {
RolloutRecorderParams::Create {
session_id,
conversation_id,
@@ -771,6 +775,7 @@ impl RolloutRecorder {
history_mode,
initial_window_id,
} => {
let ordinal_state = RolloutOrdinalState::for_new_rollout(history_mode);
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;
@@ -790,7 +795,7 @@ impl RolloutRecorder {
forked_from_id,
parent_thread_id,
timestamp,
cwd: config.cwd().to_path_buf(),
cwd: cwd.clone(),
originator,
cli_version: env!("CARGO_PKG_VERSION").to_string(),
agent_nickname: source.get_nickname(),
@@ -812,16 +817,32 @@ impl RolloutRecorder {
context_window: initial_window_id.map(SessionContextWindow::new),
};
(None, Some(log_file_info), path, Some(session_meta))
RolloutWriterState {
writer: None,
deferred_log_file_info: Some(log_file_info),
pending_items: Vec::new(),
meta: Some(session_meta),
cwd: cwd.clone(),
rollout_path: path,
ordinal_state,
last_logged_error: None,
}
}
RolloutRecorderParams::Resume { path } => {
let (path, file) = open_rollout_for_append(path.as_path()).await?;
(Some(file), None, path, None)
let (path, file, ordinal_state) = open_rollout_for_append(path.as_path()).await?;
RolloutWriterState {
writer: Some(JsonlWriter { file }),
deferred_log_file_info: None,
pending_items: Vec::new(),
meta: None,
cwd: cwd.clone(),
rollout_path: path,
ordinal_state,
last_logged_error: None,
}
}
};
// Clone the cwd for the spawned task to collect git info asynchronously
let cwd = config.cwd().to_path_buf();
let rollout_path = state.rollout_path.clone();
// A reasonably-sized bounded channel. If the buffer fills up the send
// future will yield, which is fine – we only need to ensure we do not
@@ -834,15 +855,7 @@ impl RolloutRecorder {
let writer_task_for_spawn = Arc::clone(&writer_task);
let rollout_path_for_spawn = rollout_path.clone();
let handle = tokio::task::spawn(async move {
let result = rollout_writer(
file,
deferred_log_file_info,
rx,
meta,
cwd,
rollout_path_for_spawn.clone(),
)
.await;
let result = rollout_writer(state, rx).await;
if let Err(err) = result {
// This is the terminal background-task failure path. Normal I/O failures stay inside
// `rollout_writer`, are reported through command acks, and leave items buffered for retry.
@@ -1551,28 +1564,11 @@ struct RolloutWriterState {
meta: Option<SessionMeta>,
cwd: PathBuf,
rollout_path: PathBuf,
ordinal_state: RolloutOrdinalState,
last_logged_error: Option<String>,
}
impl RolloutWriterState {
fn new(
file: Option<tokio::fs::File>,
deferred_log_file_info: Option<LogFileInfo>,
meta: Option<SessionMeta>,
cwd: PathBuf,
rollout_path: PathBuf,
) -> Self {
Self {
writer: file.map(|file| JsonlWriter { file }),
deferred_log_file_info,
pending_items: Vec::new(),
meta,
cwd,
rollout_path,
last_logged_error: None,
}
}
fn add_items(&mut self, items: Vec<RolloutItem>) {
self.pending_items.extend(items);
}
@@ -1672,7 +1668,13 @@ impl RolloutWriterState {
let Some(session_meta) = self.meta.as_ref().cloned() else {
return Ok(());
};
write_session_meta(self.writer.as_mut(), session_meta, &self.cwd).await?;
write_session_meta(
self.writer.as_mut(),
&mut self.ordinal_state,
session_meta,
&self.cwd,
)
.await?;
self.meta = None;
Ok(())
}
@@ -1697,9 +1699,18 @@ impl RolloutWriterState {
let mut written_count = 0usize;
let mut write_result = Ok(());
for item in &self.pending_items {
if let Err(err) = writer.write_rollout_item(item).await {
write_result = Err(err);
break;
match self.ordinal_state.current() {
Ok(ordinal) => match writer.write_rollout_item(item, ordinal).await {
Ok(()) => self.ordinal_state.advance(),
Err(err) => {
write_result = Err(err);
break;
}
},
Err(err) => {
write_result = Err(err);
break;
}
}
written_count += 1;
}
@@ -1713,15 +1724,9 @@ impl RolloutWriterState {
}
async fn rollout_writer(
file: Option<tokio::fs::File>,
deferred_log_file_info: Option<LogFileInfo>,
mut state: RolloutWriterState,
mut rx: mpsc::Receiver<RolloutCmd>,
meta: Option<SessionMeta>,
cwd: PathBuf,
rollout_path: PathBuf,
) -> std::io::Result<()> {
let mut state = RolloutWriterState::new(file, deferred_log_file_info, meta, cwd, rollout_path);
// Process rollout commands
while let Some(cmd) = rx.recv().await {
match cmd {
@@ -1752,6 +1757,7 @@ async fn rollout_writer(
async fn write_session_meta(
mut writer: Option<&mut JsonlWriter>,
ordinal_state: &mut RolloutOrdinalState,
session_meta: SessionMeta,
cwd: &Path,
) -> std::io::Result<()> {
@@ -1771,7 +1777,9 @@ async fn write_session_meta(
let rollout_item = RolloutItem::SessionMeta(session_meta_line);
if let Some(writer) = writer.as_mut() {
writer.write_rollout_item(&rollout_item).await?;
let ordinal = ordinal_state.current()?;
writer.write_rollout_item(&rollout_item, ordinal).await?;
ordinal_state.advance();
}
Ok(())
}
@@ -1785,25 +1793,29 @@ pub async fn append_rollout_item_to_path(
rollout_path: &Path,
item: &RolloutItem,
) -> std::io::Result<()> {
let (_rollout_path, file) = open_rollout_for_append(rollout_path).await?;
let (_rollout_path, file, ordinal_state) = open_rollout_for_append(rollout_path).await?;
let ordinal = ordinal_state.current()?;
let mut writer = JsonlWriter { file };
writer.write_rollout_item(item).await
writer.write_rollout_item(item, ordinal).await
}
async fn open_rollout_for_append(path: &Path) -> std::io::Result<(PathBuf, tokio::fs::File)> {
async fn open_rollout_for_append(
path: &Path,
) -> std::io::Result<(PathBuf, tokio::fs::File, RolloutOrdinalState)> {
let path = compression::materialize_rollout_for_append(path).await?;
let path_for_open = path.clone();
let file = tokio::task::spawn_blocking(move || {
let (file, ordinal_state) = tokio::task::spawn_blocking(move || {
let mut file = File::options()
.read(true)
.append(true)
.open(path_for_open)?;
.open(path_for_open.as_path())?;
ensure_rollout_is_newline_terminated(&mut file)?;
Ok::<_, std::io::Error>(file)
let ordinal_state = ordinal_state_for_rollout(&mut file, path_for_open.as_path())?;
Ok::<_, std::io::Error>((file, ordinal_state))
})
.await
.map_err(IoError::other)??;
Ok((path, tokio::fs::File::from_std(file)))
Ok((path, tokio::fs::File::from_std(file), ordinal_state))
}
fn ensure_rollout_is_newline_terminated(file: &mut File) -> std::io::Result<()> {
@@ -1828,12 +1840,18 @@ struct JsonlWriter {
#[derive(serde::Serialize)]
struct RolloutLineRef<'a> {
timestamp: String,
#[serde(skip_serializing_if = "Option::is_none")]
ordinal: Option<u64>,
#[serde(flatten)]
item: &'a RolloutItem,
}
impl JsonlWriter {
async fn write_rollout_item(&mut self, rollout_item: &RolloutItem) -> std::io::Result<()> {
async fn write_rollout_item(
&mut self,
rollout_item: &RolloutItem,
ordinal: Option<u64>,
) -> std::io::Result<()> {
let timestamp_format: &[FormatItem] = format_description!(
"[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z"
);
@@ -1843,6 +1861,7 @@ impl JsonlWriter {
let line = RolloutLineRef {
timestamp,
ordinal,
item: rollout_item,
};
self.write_line(&line).await
codex-rs/rollout/src/recorder_tests.rs
+246 -7
View File
@@ -38,6 +38,67 @@ fn test_config(codex_home: &Path) -> RolloutConfig {
}
}
fn paginated_session_meta_item(thread_id: ThreadId, cwd: &Path) -> RolloutItem {
RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
id: thread_id,
timestamp: "2026-07-09T00:00:00Z".to_string(),
cwd: cwd.to_path_buf(),
originator: "test".to_string(),
cli_version: "test".to_string(),
source: SessionSource::Exec,
history_mode: ThreadHistoryMode::Paginated,
..SessionMeta::default()
},
git: None,
})
}
fn agent_message_item(message: &str) -> RolloutItem {
RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent {
message: message.to_string(),
phase: None,
memory_citation: None,
}))
}
fn write_paginated_rollout(
path: &Path,
thread_id: ThreadId,
subsequent_ordinals: &[u64],
) -> std::io::Result<()> {
let mut records = vec![RolloutLine {
timestamp: "2026-07-09T00:00:00Z".to_string(),
ordinal: Some(0),
item: paginated_session_meta_item(thread_id, path.parent().unwrap_or(path)),
}];
records.extend(
subsequent_ordinals
.iter()
.enumerate()
.map(|(index, ordinal)| RolloutLine {
timestamp: format!("2026-07-09T00:00:{:02}Z", index + 1),
ordinal: Some(*ordinal),
item: agent_message_item(format!("message-{index}").as_str()),
}),
);
let jsonl = records
.iter()
.map(serde_json::to_string)
.collect::<Result<Vec<_>, _>>()?
.join("\n");
fs::write(path, format!("{jsonl}\n"))
}
fn read_rollout_lines(path: &Path) -> std::io::Result<Vec<RolloutLine>> {
fs::read_to_string(path)?
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str(line).map_err(std::io::Error::other))
.collect()
}
fn write_session_file(root: &Path, ts: &str, uuid: Uuid) -> std::io::Result<PathBuf> {
let day_dir = root.join("sessions/2025/01/03");
fs::create_dir_all(&day_dir)?;
@@ -125,10 +186,12 @@ async fn state_db_init_backfills_before_returning() -> anyhow::Result<()> {
let lines = [
RolloutLine {
timestamp: "2026-01-27T12:34:56Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(session_meta_line),
},
RolloutLine {
timestamp: "2026-01-27T12:34:57Z".to_string(),
ordinal: None,
item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "hello from startup backfill".to_string(),
@@ -488,6 +551,11 @@ async fn recorder_materializes_on_flush_with_pending_items() -> std::io::Result<
assert!(rollout_path.exists(), "rollout file should be materialized");
let text = std::fs::read_to_string(&rollout_path)?;
let lines = read_rollout_lines(&rollout_path)?;
assert_eq!(
lines.iter().map(|line| line.ordinal).collect::<Vec<_>>(),
vec![Some(0), Some(1), Some(2)]
);
let first_line = text.lines().next().expect("session metadata line");
let session_meta: RolloutLine = serde_json::from_str(first_line)?;
let RolloutItem::SessionMeta(session_meta) = session_meta.item else {
@@ -519,6 +587,63 @@ async fn recorder_materializes_on_flush_with_pending_items() -> std::io::Result<
Ok(())
}
#[tokio::test]
async fn recorder_omits_ordinals_from_legacy_rollouts() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let recorder = RolloutRecorder::new(
&config,
RolloutRecorderParams::new(
ThreadId::new(),
/*forked_from_id*/ None,
/*parent_thread_id*/ None,
SessionSource::Exec,
/*thread_source*/ None,
"test_originator".to_string(),
BaseInstructions::default(),
Vec::new(),
),
)
.await?;
recorder
.record_canonical_items(&[agent_message_item("legacy")])
.await?;
recorder.flush().await?;
let text = fs::read_to_string(recorder.rollout_path())?;
let values = text
.lines()
.map(serde_json::from_str::<serde_json::Value>)
.collect::<Result<Vec<_>, _>>()?;
assert!(values.iter().all(|value| value.get("ordinal").is_none()));
recorder.shutdown().await
}
#[tokio::test]
async fn resumed_empty_rollout_omits_ordinals() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let rollout_path = home.path().join("rollout.jsonl");
File::create(&rollout_path)?;
let recorder =
RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())).await?;
recorder
.record_canonical_items(&[agent_message_item("legacy")])
.await?;
recorder.flush().await?;
let text = fs::read_to_string(rollout_path)?;
let values = text
.lines()
.map(serde_json::from_str::<serde_json::Value>)
.collect::<Result<Vec<_>, _>>()?;
assert!(values.iter().all(|value| value.get("ordinal").is_none()));
recorder.shutdown().await
}
#[tokio::test]
async fn persist_reports_filesystem_error_and_retries_buffered_items() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
@@ -580,13 +705,18 @@ async fn writer_state_retries_write_error_before_reporting_flush_success() -> st
let rollout_path = home.path().join("rollout.jsonl");
File::create(&rollout_path)?;
let read_only_file = std::fs::OpenOptions::new().read(true).open(&rollout_path)?;
let mut state = RolloutWriterState::new(
Some(tokio::fs::File::from_std(read_only_file)),
/*deferred_log_file_info*/ None,
/*meta*/ None,
home.path().to_path_buf(),
rollout_path.clone(),
);
let mut state = RolloutWriterState {
writer: Some(JsonlWriter {
file: tokio::fs::File::from_std(read_only_file),
}),
deferred_log_file_info: None,
pending_items: Vec::new(),
meta: None,
cwd: home.path().to_path_buf(),
rollout_path: rollout_path.clone(),
ordinal_state: RolloutOrdinalState::Legacy,
last_logged_error: None,
};
state.add_items(vec![RolloutItem::EventMsg(EventMsg::AgentMessage(
AgentMessageEvent {
message: "queued-after-writer-error".to_string(),
@@ -604,6 +734,114 @@ async fn writer_state_retries_write_error_before_reporting_flush_success() -> st
Ok(())
}
#[tokio::test]
async fn resumed_paginated_rollout_continues_after_ordinal_gap() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let rollout_path = home.path().join("rollout.jsonl");
write_paginated_rollout(&rollout_path, ThreadId::new(), &[4])?;
let recorder =
RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())).await?;
recorder
.record_canonical_items(&[agent_message_item("after-resume")])
.await?;
recorder.flush().await?;
let lines = read_rollout_lines(&rollout_path)?;
assert_eq!(
lines.iter().map(|line| line.ordinal).collect::<Vec<_>>(),
vec![Some(0), Some(4), Some(5)]
);
recorder.shutdown().await
}
#[tokio::test]
async fn resumed_paginated_rollout_repairs_unsafe_tail() -> std::io::Result<()> {
let valid_unterminated = serde_json::to_string(&RolloutLine {
timestamp: "2026-07-09T00:00:05Z".to_string(),
ordinal: Some(5),
item: agent_message_item("valid unterminated"),
})?;
for (name, tail, expected_ordinals) in [
(
"valid unterminated",
valid_unterminated,
vec![Some(0), Some(4), Some(5), Some(6)],
),
(
"invalid unterminated",
"{\"timestamp\":\"unterminated\"".to_string(),
vec![Some(0), Some(4), Some(5)],
),
] {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let rollout_path = home.path().join("rollout.jsonl");
write_paginated_rollout(&rollout_path, ThreadId::new(), &[4])?;
let mut file = fs::OpenOptions::new().append(true).open(&rollout_path)?;
write!(file, "{tail}")?;
drop(file);
let recorder =
RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone()))
.await?;
recorder
.record_canonical_items(&[agent_message_item("after-tail-repair")])
.await?;
recorder.flush().await?;
let contents = fs::read_to_string(&rollout_path)?;
assert!(contents.ends_with('\n'), "{name} tail should be terminated");
let ordinals = contents
.lines()
.filter_map(|line| serde_json::from_str::<RolloutLine>(line).ok())
.map(|line| line.ordinal)
.collect::<Vec<_>>();
assert_eq!(
ordinals, expected_ordinals,
"unexpected ordinals after repairing {name} tail"
);
recorder.shutdown().await?;
}
Ok(())
}
#[tokio::test]
async fn paginated_ordinal_overflow_fails_without_appending() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let rollout_path = home.path().join("rollout.jsonl");
write_paginated_rollout(&rollout_path, ThreadId::new(), &[u64::MAX])?;
let before = fs::read(&rollout_path)?;
let recorder =
RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path.clone())).await?;
recorder
.record_canonical_items(&[agent_message_item("overflow")])
.await?;
let err = recorder
.flush()
.await
.expect_err("ordinal overflow should fail the append");
assert!(err.to_string().contains("overflow"));
assert_eq!(fs::read(&rollout_path)?, before);
Ok(())
}
#[tokio::test]
async fn append_rollout_item_to_path_assigns_next_paginated_ordinal() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let rollout_path = home.path().join("rollout.jsonl");
write_paginated_rollout(&rollout_path, ThreadId::new(), &[4])?;
append_rollout_item_to_path(&rollout_path, &agent_message_item("offline")).await?;
let lines = read_rollout_lines(&rollout_path)?;
assert_eq!(lines.last().and_then(|line| line.ordinal), Some(5));
Ok(())
}
#[tokio::test]
async fn list_threads_db_disabled_does_not_skip_paginated_items() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
@@ -1216,6 +1454,7 @@ async fn resume_candidate_matches_cwd_reads_latest_turn_context() -> std::io::Re
let mut file = std::fs::OpenOptions::new().append(true).open(&path)?;
let turn_context = RolloutLine {
timestamp: "2025-01-03T13:00:01Z".to_string(),
ordinal: None,
item: RolloutItem::TurnContext(TurnContextItem {
turn_id: Some("turn-1".to_string()),
cwd: serde_json::from_value(serde_json::json!(&latest_cwd))
codex-rs/rollout/src/session_index_tests.rs
+1
View File
@@ -23,6 +23,7 @@ fn write_rollout_with_metadata(path: &Path, thread_id: ThreadId) -> std::io::Res
let timestamp = "2024-01-01T00-00-00Z".to_string();
let line = RolloutLine {
timestamp: timestamp.clone(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
codex-rs/rollout/src/state_db_tests.rs
+2
View File
@@ -156,6 +156,7 @@ fn write_rollout_with_user_message(
let lines = [
RolloutLine {
timestamp: "2026-06-01T14:26:25Z".to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: thread_id.into(),
@@ -185,6 +186,7 @@ fn write_rollout_with_user_message(
},
RolloutLine {
timestamp: "2026-06-01T14:26:26Z".to_string(),
ordinal: None,
item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: message.to_string(),
..Default::default()
codex-rs/rollout/src/tests.rs
+3
View File
@@ -1314,6 +1314,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
let conversation_id = ThreadId::from_string(&uuid.to_string())?;
let meta_line = RolloutLine {
timestamp: ts.to_string(),
ordinal: None,
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
session_id: conversation_id.into(),
@@ -1345,6 +1346,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
let user_event_line = RolloutLine {
timestamp: ts.to_string(),
ordinal: None,
item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
client_id: None,
message: "hello".into(),
@@ -1360,6 +1362,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
for idx in 0..total_messages {
let response_line = RolloutLine {
timestamp: format!("{ts}-{idx:02}"),
ordinal: None,
item: RolloutItem::ResponseItem(ResponseItem::Message {
id: None,
role: "assistant".into(),