diff --git a/codex-rs/thread-store/src/local/thread_history_materialization.rs b/codex-rs/thread-store/src/local/thread_history_materialization.rs index 34eeede51b..a543302c38 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization.rs @@ -1,3 +1,5 @@ +use std::io::Read; +use std::io::Seek; use std::io::SeekFrom; use std::path::Path; @@ -6,8 +8,6 @@ use codex_app_server_protocol::ThreadHistoryChangeSet; use codex_app_server_protocol::project_rollout_line; use codex_protocol::ThreadId; use codex_rollout::RolloutItem; -use tokio::io::AsyncReadExt; -use tokio::io::AsyncSeekExt; use tracing::warn; use super::LocalThreadStore; @@ -90,20 +90,39 @@ async fn read_projection_steps( subagent_history_start_ordinal: Option, ) -> ThreadStoreResult<(Vec, u64)> { let path = rollout_path.to_path_buf(); - let file = - tokio::task::spawn_blocking(move || codex_rollout::open_rollout_seekable_reader(&path)) - .await - .map_err(|err| ThreadStoreError::Internal { - message: format!("failed to join rollout projection read: {err}"), - })?; - let mut file = match file { - Ok(file) => tokio::fs::File::from_std(file), + let span = tracing::Span::current(); + tokio::task::spawn_blocking(move || { + let _entered = span.enter(); + read_projection_steps_sync( + &path, + start_offset, + expected_ordinal, + thread_id, + subagent_history_start_ordinal, + ) + }) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to join rollout projection read: {err}"), + })? +} + +/// Read and project the complete durable prefix in one blocking task. +fn read_projection_steps_sync( + rollout_path: &Path, + start_offset: u64, + expected_ordinal: u64, + thread_id: ThreadId, + subagent_history_start_ordinal: Option, +) -> ThreadStoreResult<(Vec, u64)> { + let mut file = match codex_rollout::open_rollout_seekable_reader(rollout_path) { + Ok(file) => file, Err(err) if err.kind() == std::io::ErrorKind::NotFound && start_offset == 0 => { return Ok((Vec::new(), 0)); } Err(err) => return Err(thread_store_io_error(err)), }; - let file_end_offset = file.metadata().await.map_err(thread_store_io_error)?.len(); + let file_end_offset = file.metadata().map_err(thread_store_io_error)?.len(); let byte_count = file_end_offset .checked_sub(start_offset) @@ -115,10 +134,8 @@ async fn read_projection_steps( })?; let mut bytes = vec![0; byte_count]; file.seek(SeekFrom::Start(start_offset)) - .await .map_err(thread_store_io_error)?; file.read_exact(bytes.as_mut_slice()) - .await .map_err(thread_store_io_error)?; // Only project the newline-terminated prefix; leave a trailing partial record for the next // pass.