diff --git a/codex-rs/app-server-protocol/src/lib.rs b/codex-rs/app-server-protocol/src/lib.rs index 0797fcdf7e38..26842d50dce4 100644 --- a/codex-rs/app-server-protocol/src/lib.rs +++ b/codex-rs/app-server-protocol/src/lib.rs @@ -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; diff --git a/codex-rs/app-server-protocol/src/protocol/mod.rs b/codex-rs/app-server-protocol/src/protocol/mod.rs index 592944b35f21..3a90aa704512 100644 --- a/codex-rs/app-server-protocol/src/protocol/mod.rs +++ b/codex-rs/app-server-protocol/src/protocol/mod.rs @@ -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; diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs new file mode 100644 index 000000000000..38de714bcebc --- /dev/null +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs @@ -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; diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs new file mode 100644 index 000000000000..851b95ecbe9e --- /dev/null +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection_tests.rs @@ -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, + })) +} diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index 8c0d59efed63..38a94ab8121a 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs @@ -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, diff --git a/codex-rs/cli/src/doctor/thread_inventory.rs b/codex-rs/cli/src/doctor/thread_inventory.rs index c50e6debdc08..94f68a679e92 100644 --- a/codex-rs/cli/src/doctor/thread_inventory.rs +++ b/codex-rs/cli/src/doctor/thread_inventory.rs @@ -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(), diff --git a/codex-rs/core/tests/suite/client.rs b/codex-rs/core/tests/suite/client.rs index 9059a54850bd..92682592a59d 100644 --- a/codex-rs/core/tests/suite/client.rs +++ b/codex-rs/core/tests/suite/client.rs @@ -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(), diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index 726d74ebfdad..8759d83fa3c3 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -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(), diff --git a/codex-rs/memories/write/src/startup_tests.rs b/codex-rs/memories/write/src/startup_tests.rs index 4154b852bf82..1338e92b628a 100644 --- a/codex-rs/memories/write/src/startup_tests.rs +++ b/codex-rs/memories/write/src/startup_tests.rs @@ -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(), diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index f0ec2bf85092..59dec101e44e 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -3334,6 +3334,8 @@ impl Mul for TruncationPolicy { #[derive(Serialize, Deserialize, Clone, JsonSchema)] pub struct RolloutLine { pub timestamp: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub ordinal: Option, #[serde(flatten)] pub item: RolloutItem, } diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index 46c90d0120d3..3b8132534072 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/codex-rs/rollout/src/compression_tests.rs @@ -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() diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 43e7a9216f05..7cf140dc2082 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -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; diff --git a/codex-rs/rollout/src/metadata_tests.rs b/codex-rs/rollout/src/metadata_tests.rs index 2e1ea98f3aca..c3163b6ef1d6 100644 --- a/codex-rs/rollout/src/metadata_tests.rs +++ b/codex-rs/rollout/src/metadata_tests.rs @@ -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"); diff --git a/codex-rs/rollout/src/ordinal.rs b/codex-rs/rollout/src/ordinal.rs new file mode 100644 index 000000000000..f6e36dcbd24c --- /dev/null +++ b/codex-rs/rollout/src/ordinal.rs @@ -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 }, +} + +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> { + 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 { + 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::()? { + 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> { + 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) +} diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 735af8f0a493..8bfcadf45978 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -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 { - 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, cwd: PathBuf, rollout_path: PathBuf, + ordinal_state: RolloutOrdinalState, last_logged_error: Option, } impl RolloutWriterState { - fn new( - file: Option, - deferred_log_file_info: Option, - meta: Option, - 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) { 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, - deferred_log_file_info: Option, + mut state: RolloutWriterState, mut rx: mpsc::Receiver, - meta: Option, - 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, #[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, + ) -> 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 diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index 672902c46e5e..99e2339a1755 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -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::, _>>()? + .join("\n"); + fs::write(path, format!("{jsonl}\n")) +} + +fn read_rollout_lines(path: &Path) -> std::io::Result> { + 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 { 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![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::) + .collect::, _>>()?; + 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::) + .collect::, _>>()?; + 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![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::(line).ok()) + .map(|line| line.ordinal) + .collect::>(); + 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)) diff --git a/codex-rs/rollout/src/session_index_tests.rs b/codex-rs/rollout/src/session_index_tests.rs index 45aad8279cfb..b7008dfc8bde 100644 --- a/codex-rs/rollout/src/session_index_tests.rs +++ b/codex-rs/rollout/src/session_index_tests.rs @@ -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(), diff --git a/codex-rs/rollout/src/state_db_tests.rs b/codex-rs/rollout/src/state_db_tests.rs index 18566d887ebc..dd67a9759b23 100644 --- a/codex-rs/rollout/src/state_db_tests.rs +++ b/codex-rs/rollout/src/state_db_tests.rs @@ -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() diff --git a/codex-rs/rollout/src/tests.rs b/codex-rs/rollout/src/tests.rs index 23f72cb9473e..9285faacdbc0 100644 --- a/codex-rs/rollout/src/tests.rs +++ b/codex-rs/rollout/src/tests.rs @@ -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(),