//! Persist Codex session rollouts (.jsonl) so sessions can be replayed or inspected later. use std::collections::HashSet; use std::fs; use std::fs::File; use std::io::Error as IoError; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; use chrono::SecondsFormat; use codex_protocol::ThreadId; use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::models::BaseInstructions; use serde_json::Value; use time::OffsetDateTime; use time::format_description::FormatItem; use time::format_description::well_known::Rfc3339; use time::macros::format_description; use tokio::io::AsyncWriteExt; use tokio::sync::mpsc; use tokio::sync::mpsc::Sender; use tokio::sync::oneshot; use tokio::task::JoinHandle; use tracing::error; use tracing::info; use tracing::trace; use tracing::warn; use super::ARCHIVED_SESSIONS_SUBDIR; use super::SESSIONS_SUBDIR; use super::compression; use super::list::Cursor; use super::list::SortDirection; use super::list::ThreadItem; use super::list::ThreadListConfig; use super::list::ThreadListLayout; use super::list::ThreadSortKey; use super::list::ThreadsPage; use super::list::get_threads; 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::session_index::find_thread_names_by_ids; use crate::config::RolloutConfigView; use crate::default_client::originator; use crate::state_db; use crate::state_db::StateDbHandle; use codex_git_utils::collect_git_info; use codex_git_utils::get_git_repo_root; use codex_protocol::protocol::GitInfo as ProtocolGitInfo; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::protocol::ResumedHistory; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadSource; use codex_state::StateRuntime; use codex_utils_path as path_utils; /// Writes canonical session rollout items to JSONL. /// /// Rollouts are recorded as JSONL and can be inspected with tools such as: /// /// ```ignore /// $ jq -C . ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl /// $ fx ~/.codex/sessions/rollout-2025-05-07T17-24-21-5973b6c0-94b8-487b-a530-2aeb6098ae0e.jsonl /// ``` #[derive(Clone)] pub struct RolloutRecorder { tx: Sender, writer_task: Arc, pub(crate) rollout_path: PathBuf, } #[derive(Clone)] pub enum RolloutRecorderParams { Create { conversation_id: ThreadId, forked_from_id: Option, parent_thread_id: Option, source: SessionSource, thread_source: Option, base_instructions: BaseInstructions, dynamic_tools: Vec, multi_agent_version: Option, }, Resume { path: PathBuf, }, } enum RolloutCmd { AddItems(Vec), Persist { ack: oneshot::Sender>, }, /// Ensure all prior writes are processed; respond when flushed. Flush { ack: oneshot::Sender>, }, Shutdown { ack: oneshot::Sender>, }, } /// Observable state for the background rollout writer task. struct RolloutWriterTask { handle: Mutex>>, terminal_failure: Mutex>>, } impl RolloutWriterTask { /// Create task observability state before spawning the writer. fn new() -> Self { Self { handle: Mutex::new(None), terminal_failure: Mutex::new(None), } } /// Store the spawned task handle so it remains owned for the lifetime of recorder clones. fn set_handle(&self, handle: JoinHandle<()>) { let mut guard = self .handle .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); *guard = Some(handle); } /// Remember a terminal task failure for future recorder API calls. fn mark_failed(&self, err: &IoError) { let mut guard = self .terminal_failure .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); *guard = Some(Arc::new(clone_io_error(err))); } /// Return the terminal writer-task failure, if the task exited with an error. fn terminal_failure(&self) -> Option { let guard = self .terminal_failure .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); guard.as_ref().map(|err| clone_io_error(err.as_ref())) } } fn clone_io_error(err: &IoError) -> IoError { IoError::new(err.kind(), err.to_string()) } impl RolloutRecorderParams { pub fn new( conversation_id: ThreadId, forked_from_id: Option, parent_thread_id: Option, source: SessionSource, thread_source: Option, base_instructions: BaseInstructions, dynamic_tools: Vec, ) -> Self { Self::Create { conversation_id, forked_from_id, parent_thread_id, source, thread_source, base_instructions, dynamic_tools, multi_agent_version: None, } } pub fn with_multi_agent_version( mut self, multi_agent_version: Option, ) -> Self { if let Self::Create { multi_agent_version: version, .. } = &mut self { *version = multi_agent_version; } self } pub fn resume(path: PathBuf) -> Self { Self::Resume { path } } } #[derive(Clone, Copy)] enum ThreadListArchiveFilter { Active, Archived, } #[derive(Clone, Copy)] enum ThreadListRepairMode { ScanAndRepair, StateDbOnly, } impl RolloutRecorder { /// List threads (rollout files) under the provided Codex home directory. #[allow(clippy::too_many_arguments)] pub async fn list_threads( state_db_ctx: Option, config: &impl RolloutConfigView, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, sort_direction: SortDirection, allowed_sources: &[SessionSource], model_providers: Option<&[String]>, cwd_filters: Option<&[PathBuf]>, default_provider: &str, search_term: Option<&str>, ) -> std::io::Result { Self::list_threads_with_db_fallback( state_db_ctx, config, page_size, cursor, sort_key, sort_direction, allowed_sources, model_providers, cwd_filters, default_provider, ThreadListArchiveFilter::Active, ThreadListRepairMode::ScanAndRepair, search_term, ) .await } #[allow(clippy::too_many_arguments)] pub async fn list_threads_from_state_db( state_db_ctx: Option, config: &impl RolloutConfigView, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, sort_direction: SortDirection, allowed_sources: &[SessionSource], model_providers: Option<&[String]>, cwd_filters: Option<&[PathBuf]>, default_provider: &str, search_term: Option<&str>, ) -> std::io::Result { Self::list_threads_with_db_fallback( state_db_ctx, config, page_size, cursor, sort_key, sort_direction, allowed_sources, model_providers, cwd_filters, default_provider, ThreadListArchiveFilter::Active, ThreadListRepairMode::StateDbOnly, search_term, ) .await } /// List archived threads (rollout files) under the archived sessions directory. #[allow(clippy::too_many_arguments)] pub async fn list_archived_threads( state_db_ctx: Option, config: &impl RolloutConfigView, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, sort_direction: SortDirection, allowed_sources: &[SessionSource], model_providers: Option<&[String]>, cwd_filters: Option<&[PathBuf]>, default_provider: &str, search_term: Option<&str>, ) -> std::io::Result { Self::list_threads_with_db_fallback( state_db_ctx, config, page_size, cursor, sort_key, sort_direction, allowed_sources, model_providers, cwd_filters, default_provider, ThreadListArchiveFilter::Archived, ThreadListRepairMode::ScanAndRepair, search_term, ) .await } #[allow(clippy::too_many_arguments)] pub async fn list_archived_threads_from_state_db( state_db_ctx: Option, config: &impl RolloutConfigView, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, sort_direction: SortDirection, allowed_sources: &[SessionSource], model_providers: Option<&[String]>, cwd_filters: Option<&[PathBuf]>, default_provider: &str, search_term: Option<&str>, ) -> std::io::Result { Self::list_threads_with_db_fallback( state_db_ctx, config, page_size, cursor, sort_key, sort_direction, allowed_sources, model_providers, cwd_filters, default_provider, ThreadListArchiveFilter::Archived, ThreadListRepairMode::StateDbOnly, search_term, ) .await } #[allow(clippy::too_many_arguments)] async fn list_threads_with_db_fallback( state_db_ctx: Option, config: &impl RolloutConfigView, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, sort_direction: SortDirection, allowed_sources: &[SessionSource], model_providers: Option<&[String]>, cwd_filters: Option<&[PathBuf]>, default_provider: &str, archive_filter: ThreadListArchiveFilter, repair_mode: ThreadListRepairMode, search_term: Option<&str>, ) -> std::io::Result { let codex_home = config.codex_home(); let archived = match archive_filter { ThreadListArchiveFilter::Active => false, ThreadListArchiveFilter::Archived => true, }; if cwd_filters.is_some_and(<[std::path::PathBuf]>::is_empty) { return Ok(ThreadsPage::default()); } if matches!(repair_mode, ThreadListRepairMode::StateDbOnly) { return Ok(state_db::list_threads_db( state_db_ctx.as_deref(), codex_home, page_size, cursor, sort_key, sort_direction, allowed_sources, model_providers, cwd_filters, /*parent_thread_id*/ None, archived, search_term, ) .await .map(Into::into) .unwrap_or_default()); } let listing_has_metadata_filters = !allowed_sources.is_empty() || model_providers.is_some() || cwd_filters.is_some() || search_term.is_some(); // Filesystem-first listing intentionally overfetches so we can repair stale/missing // SQLite rows before returning the scan page for filtered listings or the DB page for // unfiltered listings. let fs_page = match sort_direction { SortDirection::Asc => { list_threads_from_files_asc( codex_home, page_size, cursor, sort_key, allowed_sources, model_providers, cwd_filters, default_provider, archived, search_term, ) .await? } SortDirection::Desc => { list_threads_from_files_desc( codex_home, page_size.saturating_mul(2), cursor, sort_key, allowed_sources, model_providers, cwd_filters, default_provider, archived, search_term, ) .await? } }; if state_db_ctx.is_none() { // Keep legacy behavior when SQLite is unavailable: return filesystem results // at the requested page size. codex_state::record_fallback( "list_threads", "db_unavailable", /*telemetry_override*/ None, ); return Ok(page_from_filesystem_scan( fs_page, sort_direction, page_size, sort_key, )); } // For metadata-filtered listings the filesystem page is the page we return. Track those // IDs so the later DB page only triggers full reconciliation for DB-only hits. let fs_page_thread_ids = fs_page .items .iter() .filter_map(|item| item.thread_id) .collect::>(); // Warm the DB by repairing every filesystem hit before querying SQLite. Source/provider/cwd // filters are already validated from rollout head metadata, so lightweight read-repair is // enough there. Search can depend on full title metadata, so keep full reconciliation. for item in &fs_page.items { if search_term.is_some() { state_db::reconcile_rollout( state_db_ctx.as_deref(), item.path.as_path(), default_provider, /*builder*/ None, &[], Some(archived), /*new_thread_memory_mode*/ None, ) .await; } else { state_db::read_repair_rollout_path( state_db_ctx.as_deref(), item.thread_id, Some(archived), item.path.as_path(), ) .await; } } let db_page = state_db::list_threads_db( state_db_ctx.as_deref(), codex_home, page_size, cursor, sort_key, sort_direction, allowed_sources, model_providers, cwd_filters, /*parent_thread_id*/ None, archived, search_term, ) .await; if let Some(db_page) = db_page { if search_term.is_some() && (!db_page.items.is_empty() || cursor.is_some()) { for item in &db_page.items { state_db::reconcile_rollout( state_db_ctx.as_deref(), item.rollout_path.as_path(), default_provider, /*builder*/ None, &[], Some(archived), /*new_thread_memory_mode*/ None, ) .await; } if let Some(repaired_db_page) = state_db::list_threads_db( state_db_ctx.as_deref(), codex_home, page_size, cursor, sort_key, sort_direction, allowed_sources, model_providers, cwd_filters, /*parent_thread_id*/ None, archived, search_term, ) .await { return Ok(repaired_db_page.into()); } return Ok(db_page.into()); } if listing_has_metadata_filters { for item in &db_page.items { // Rows that also appeared in the filesystem page were just validated from the // rollout head. Rows only found by SQLite may be stale filter matches, so fully // reconcile those before returning the filesystem-backed page. if fs_page_thread_ids.contains(&item.id) { continue; } state_db::reconcile_rollout( state_db_ctx.as_deref(), item.rollout_path.as_path(), default_provider, /*builder*/ None, &[], Some(archived), /*new_thread_memory_mode*/ None, ) .await; } codex_state::record_fallback( "list_threads", "metadata_filter", /*telemetry_override*/ None, ); let page = page_from_filesystem_scan(fs_page, sort_direction, page_size, sort_key); return Ok(fill_missing_thread_item_metadata_from_state_db( state_db_ctx.as_deref(), page, ) .await); } return Ok(db_page.into()); } if listing_has_metadata_filters { let page = page_from_filesystem_scan(fs_page, sort_direction, page_size, sort_key); codex_state::record_fallback( "list_threads", "db_error", /*telemetry_override*/ None, ); return Ok(fill_missing_thread_item_metadata_from_state_db( state_db_ctx.as_deref(), page, ) .await); } // If SQLite listing still fails, return the filesystem page rather than failing the list. tracing::error!("Falling back on rollout system"); tracing::warn!("state db discrepancy during list_threads_with_db_fallback: falling_back"); codex_state::record_fallback("list_threads", "db_error", /*telemetry_override*/ None); Ok(page_from_filesystem_scan( fs_page, sort_direction, page_size, sort_key, )) } /// Find the newest recorded thread path, optionally filtering to a matching cwd. #[allow(clippy::too_many_arguments)] pub async fn find_latest_thread_path( state_db_ctx: Option, config: &impl RolloutConfigView, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, allowed_sources: &[SessionSource], model_providers: Option<&[String]>, default_provider: &str, filter_cwd: Option<&Path>, ) -> std::io::Result> { let codex_home = config.codex_home(); let cwd_filter = filter_cwd.map(Path::to_path_buf); let mut fallback_reason = state_db_ctx.is_none().then_some("db_unavailable"); if state_db_ctx.is_some() { let mut db_cursor = cursor.cloned(); loop { let Some(db_page) = state_db::list_threads_db( state_db_ctx.as_deref(), codex_home, page_size, db_cursor.as_ref(), sort_key, SortDirection::Desc, allowed_sources, model_providers, cwd_filter.as_ref().map(std::slice::from_ref), /*parent_thread_id*/ None, /*archived*/ false, /*search_term*/ None, ) .await else { fallback_reason = Some("db_error"); break; }; if let Some(path) = select_resume_path_from_db_page(&db_page, filter_cwd, default_provider).await { return Ok(Some(path)); } db_cursor = db_page.next_anchor.map(Into::into); if db_cursor.is_none() { fallback_reason = Some("missing_row"); break; } } } if let Some(reason) = fallback_reason { codex_state::record_fallback( "find_latest_thread_path", reason, /*telemetry_override*/ None, ); } let mut cursor = cursor.cloned(); loop { let page = get_threads( codex_home, page_size, cursor.as_ref(), sort_key, allowed_sources, model_providers, cwd_filter.as_ref().map(std::slice::from_ref), default_provider, ) .await?; if let Some(path) = select_resume_path(&page, filter_cwd, default_provider).await { return Ok(Some(path)); } cursor = page.next_cursor; if cursor.is_none() { return Ok(None); } } } /// Attempt to create a new [`RolloutRecorder`]. /// /// For newly created sessions, this precomputes path/metadata and defers /// file creation/open until an explicit `persist()` call. /// /// For resumed sessions, this immediately opens the existing rollout file. pub async fn new( config: &impl RolloutConfigView, params: RolloutRecorderParams, ) -> std::io::Result { let (file, deferred_log_file_info, rollout_path, meta) = match params { RolloutRecorderParams::Create { conversation_id, forked_from_id, parent_thread_id, source, thread_source, base_instructions, dynamic_tools, multi_agent_version, } => { let log_file_info = precompute_log_file_info(config, conversation_id)?; let path = log_file_info.path.clone(); let session_id = log_file_info.conversation_id; let started_at = log_file_info.timestamp; let timestamp_format: &[FormatItem] = format_description!( "[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z" ); let timestamp = started_at .to_offset(time::UtcOffset::UTC) .format(timestamp_format) .map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?; let session_meta = SessionMeta { id: session_id, forked_from_id, parent_thread_id, timestamp, cwd: config.cwd().to_path_buf(), originator: originator().value, cli_version: env!("CARGO_PKG_VERSION").to_string(), agent_nickname: source.get_nickname(), agent_role: source.get_agent_role(), agent_path: source.get_agent_path().map(Into::into), source, thread_source, model_provider: Some(config.model_provider_id().to_string()), base_instructions: Some(base_instructions), dynamic_tools: if dynamic_tools.is_empty() { None } else { Some(dynamic_tools) }, memory_mode: (!config.generate_memories()).then_some("disabled".to_string()), multi_agent_version, }; (None, Some(log_file_info), path, Some(session_meta)) } RolloutRecorderParams::Resume { path } => { let path = compression::materialize_rollout_for_append(path.as_path()).await?; ( Some( tokio::fs::OpenOptions::new() .append(true) .open(&path) .await?, ), None, path, None, ) } }; // Clone the cwd for the spawned task to collect git info asynchronously let cwd = config.cwd().to_path_buf(); // 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 // perform *blocking* I/O on the caller's thread. let (tx, rx) = mpsc::channel::(256); // Spawn a Tokio task that owns the file handle and performs async // writes. Using `tokio::fs::File` keeps everything on the async I/O // driver instead of blocking the runtime. let writer_task = Arc::new(RolloutWriterTask::new()); 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; 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. error!( "rollout writer task failed for {}: {err}; error_kind={:?}; raw_os_error={:?}", rollout_path_for_spawn.display(), err.kind(), err.raw_os_error() ); writer_task_for_spawn.mark_failed(&err); } }); writer_task.set_handle(handle); Ok(Self { tx, writer_task, rollout_path, }) } pub fn rollout_path(&self) -> &Path { self.rollout_path.as_path() } pub async fn record_canonical_items(&self, items: &[RolloutItem]) -> std::io::Result<()> { if items.is_empty() { return Ok(()); } self.tx .send(RolloutCmd::AddItems(items.to_vec())) .await .map_err(|e| { self.writer_task.terminal_failure().unwrap_or_else(|| { IoError::other(format!("failed to queue rollout items: {e}")) }) }) } /// Materialize the rollout file and persist all buffered items. /// /// This is idempotent. If materialization fails, the recorder keeps all pending items in memory /// and a later `persist()` or `flush()` can retry opening and writing the rollout file. pub async fn persist(&self) -> std::io::Result<()> { let (tx, rx) = oneshot::channel(); self.tx .send(RolloutCmd::Persist { ack: tx }) .await .map_err(|e| { self.writer_task.terminal_failure().unwrap_or_else(|| { IoError::other(format!("failed to queue rollout persist: {e}")) }) })?; rx.await.map_err(|e| { self.writer_task.terminal_failure().unwrap_or_else(|| { IoError::other(format!("failed waiting for rollout persist: {e}")) }) })? } /// Flush all queued writes and wait until they are committed by the writer task. /// /// If the first writer attempt fails, the writer drops and reopens the file handle before /// retrying. This returns an error only when that retry also fails or the writer task is gone. pub async fn flush(&self) -> std::io::Result<()> { let (tx, rx) = oneshot::channel(); self.tx .send(RolloutCmd::Flush { ack: tx }) .await .map_err(|e| { self.writer_task.terminal_failure().unwrap_or_else(|| { IoError::other(format!("failed to queue rollout flush: {e}")) }) })?; rx.await.map_err(|e| { self.writer_task .terminal_failure() .unwrap_or_else(|| IoError::other(format!("failed waiting for rollout flush: {e}"))) })? } pub async fn load_rollout_items( path: &Path, ) -> std::io::Result<(Vec, Option, usize)> { trace!("Resuming rollout from {path:?}"); let mut items: Vec = Vec::new(); let mut thread_id: Option = None; let mut parse_errors = 0usize; let mut reader = compression::open_rollout_line_reader(path).await?; let mut saw_non_empty_line = false; while let Some(line) = reader.next_line().await? { if line.trim().is_empty() { continue; } saw_non_empty_line = true; let mut v: Value = match serde_json::from_str(&line) { Ok(v) => v, Err(e) => { warn!("failed to parse line as JSON: {line:?}, error: {e}"); parse_errors = parse_errors.saturating_add(1); continue; } }; if strip_legacy_ghost_snapshot_rollout_line(&mut v) { trace!("skipping legacy ghost_snapshot rollout line"); continue; } // Parse the rollout line structure match serde_json::from_value::(v.clone()) { Ok(rollout_line) => { let item = rollout_line.item; // Use the FIRST SessionMeta encountered in the file as the canonical // thread id and main session information. Keep all items intact. if thread_id.is_none() && let RolloutItem::SessionMeta(session_meta_line) = &item { thread_id = Some(session_meta_line.meta.id); } items.push(item); } Err(e) => { trace!("failed to parse rollout line: {e}"); parse_errors = parse_errors.saturating_add(1); } } } if !saw_non_empty_line { return Err(IoError::other("empty session file")); } tracing::debug!( "Resumed rollout with {} items, thread ID: {:?}, parse errors: {}", items.len(), thread_id, parse_errors, ); Ok((items, thread_id, parse_errors)) } pub async fn get_rollout_history(path: &Path) -> std::io::Result { let (items, thread_id, _parse_errors) = Self::load_rollout_items(path).await?; let conversation_id = thread_id .ok_or_else(|| IoError::other("failed to parse thread ID from rollout file"))?; if items.is_empty() { return Ok(InitialHistory::New); } info!("Resumed rollout successfully from {path:?}"); Ok(InitialHistory::Resumed(ResumedHistory { conversation_id, history: items, rollout_path: Some(compression::plain_rollout_path(path)), })) } /// Drain pending items before stopping the writer task. /// /// If draining fails, the writer stays alive so callers can continue retrying flush/shutdown. pub async fn shutdown(&self) -> std::io::Result<()> { let (tx_done, rx_done) = oneshot::channel(); match self.tx.send(RolloutCmd::Shutdown { ack: tx_done }).await { Ok(_) => rx_done.await.map_err(|e| { self.writer_task.terminal_failure().unwrap_or_else(|| { IoError::other(format!("failed waiting for rollout shutdown: {e}")) }) })??, Err(e) => { if let Some(err) = self.writer_task.terminal_failure() { warn!( "failed to send rollout shutdown command because writer task failed: {err}" ); return Err(err); } warn!("failed to send rollout shutdown command: {e}"); return Err(IoError::other(format!( "failed to send rollout shutdown command: {e}" ))); } }; Ok(()) } } fn strip_legacy_ghost_snapshot_rollout_line(value: &mut Value) -> bool { match value.get("type").and_then(Value::as_str) { Some("response_item") => value .get("payload") .is_some_and(is_legacy_ghost_snapshot_response_item), Some("compacted") => { if let Some(replacement_history) = value .get_mut("payload") .and_then(|payload| payload.get_mut("replacement_history")) .and_then(Value::as_array_mut) { replacement_history.retain(|item| !is_legacy_ghost_snapshot_response_item(item)); } false } _ => false, } } fn is_legacy_ghost_snapshot_response_item(value: &Value) -> bool { value.get("type").and_then(Value::as_str) == Some("ghost_snapshot") } fn truncate_fs_page( mut page: ThreadsPage, page_size: usize, sort_key: ThreadSortKey, ) -> ThreadsPage { if page.items.len() <= page_size { return page; } page.items.truncate(page_size); page.next_cursor = page.items.last().and_then(|item| { let file_name = item.path.file_name()?.to_str()?; let (created_at, _id) = parse_timestamp_uuid_from_filename(file_name)?; let cursor_token = match sort_key { ThreadSortKey::CreatedAt => created_at.format(&Rfc3339).ok()?, ThreadSortKey::UpdatedAt => item.updated_at.as_deref()?.to_string(), }; parse_cursor(cursor_token.as_str()) }); page } fn page_from_filesystem_scan( page: ThreadsPage, sort_direction: SortDirection, page_size: usize, sort_key: ThreadSortKey, ) -> ThreadsPage { match sort_direction { SortDirection::Asc => page, SortDirection::Desc => truncate_fs_page(page, page_size, sort_key), } } async fn fill_missing_thread_item_metadata_from_state_db( state_db_ctx: Option<&StateRuntime>, mut page: ThreadsPage, ) -> ThreadsPage { let Some(state_db_ctx) = state_db_ctx else { return page; }; for item in &mut page.items { let Some(thread_id) = item.thread_id else { continue; }; let metadata = match state_db_ctx.get_thread(thread_id).await { Ok(Some(metadata)) => metadata, Ok(None) => continue, Err(err) => { warn!( "state db get_thread failed while overlaying filesystem scan thread metadata: {err}" ); continue; } }; fill_missing_thread_item_metadata(item, thread_item_from_state_metadata(metadata)); } page } fn fill_missing_thread_item_metadata(item: &mut ThreadItem, state_item: ThreadItem) { let ThreadItem { path: _state_path, thread_id: _state_thread_id, first_user_message, preview, cwd, git_branch, git_sha, git_origin_url, source, parent_thread_id, agent_nickname, agent_role, model_provider, cli_version, created_at, updated_at, } = state_item; if item.first_user_message.is_none() { item.first_user_message = first_user_message; } if item.preview.is_none() { item.preview = preview; } if item.cwd.is_none() { item.cwd = cwd; } if git_branch.is_some() { item.git_branch = git_branch; } if git_sha.is_some() { item.git_sha = git_sha; } if git_origin_url.is_some() { item.git_origin_url = git_origin_url; } if item.source.is_none() { item.source = source; } if item.parent_thread_id.is_none() { item.parent_thread_id = parent_thread_id; } if item.agent_nickname.is_none() { item.agent_nickname = agent_nickname; } if item.agent_role.is_none() { item.agent_role = agent_role; } if item.model_provider.is_none() { item.model_provider = model_provider; } if item.cli_version.is_none() { item.cli_version = cli_version; } if item.created_at.is_none() { item.created_at = created_at; } if item.updated_at.is_none() { item.updated_at = updated_at; } } #[allow(clippy::too_many_arguments)] async fn list_threads_from_files_desc( codex_home: &Path, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, allowed_sources: &[SessionSource], model_providers: Option<&[String]>, cwd_filters: Option<&[PathBuf]>, default_provider: &str, archived: bool, search_term: Option<&str>, ) -> std::io::Result { if let Some(search_term) = search_term { let mut matching_items = Vec::new(); let mut scanned_files = 0usize; let mut reached_scan_cap = false; let mut page_cursor = cursor.cloned(); let scan_page_size = page_size.saturating_mul(8).clamp(256, 2048); loop { let mut page = list_threads_from_files_desc_unfiltered( codex_home, scan_page_size, page_cursor.as_ref(), sort_key, allowed_sources, model_providers, cwd_filters, default_provider, archived, ) .await?; scanned_files = scanned_files.saturating_add(page.num_scanned_files); reached_scan_cap |= page.reached_scan_cap; filter_thread_items_by_search_term(codex_home, &mut page.items, Some(search_term)) .await?; matching_items.extend(page.items); page_cursor = page.next_cursor; if matching_items.len() > page_size || page_cursor.is_none() { break; } } let more_matches_available = matching_items.len() > page_size || page_cursor.is_some() || reached_scan_cap; matching_items.truncate(page_size); let next_cursor = if more_matches_available { matching_items .last() .and_then(|item| cursor_from_thread_item(item, sort_key)) } else { None }; return Ok(ThreadsPage { items: matching_items, next_cursor, num_scanned_files: scanned_files, reached_scan_cap, }); } list_threads_from_files_desc_unfiltered( codex_home, page_size, cursor, sort_key, allowed_sources, model_providers, cwd_filters, default_provider, archived, ) .await } #[allow(clippy::too_many_arguments)] async fn list_threads_from_files_desc_unfiltered( codex_home: &Path, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, allowed_sources: &[SessionSource], model_providers: Option<&[String]>, cwd_filters: Option<&[PathBuf]>, default_provider: &str, archived: bool, ) -> std::io::Result { if archived { let root = codex_home.join(ARCHIVED_SESSIONS_SUBDIR); get_threads_in_root( root, page_size, cursor, sort_key, ThreadListConfig { allowed_sources, model_providers, cwd_filters, default_provider, layout: ThreadListLayout::Flat, }, ) .await } else { get_threads( codex_home, page_size, cursor, sort_key, allowed_sources, model_providers, cwd_filters, default_provider, ) .await } } #[allow(clippy::too_many_arguments)] async fn list_threads_from_files_asc( codex_home: &Path, page_size: usize, cursor: Option<&Cursor>, sort_key: ThreadSortKey, allowed_sources: &[SessionSource], model_providers: Option<&[String]>, cwd_filters: Option<&[PathBuf]>, default_provider: &str, archived: bool, search_term: Option<&str>, ) -> std::io::Result { let mut all_items = Vec::new(); let mut scanned_files = 0usize; let mut reached_scan_cap = false; let mut page_cursor = None; let scan_page_size = page_size.saturating_mul(8).clamp(256, 2048); loop { let page = list_threads_from_files_desc( codex_home, scan_page_size, page_cursor.as_ref(), sort_key, allowed_sources, model_providers, cwd_filters, default_provider, archived, /*search_term*/ None, ) .await?; scanned_files = scanned_files.saturating_add(page.num_scanned_files); reached_scan_cap |= page.reached_scan_cap; all_items.extend(page.items); page_cursor = page.next_cursor; if page_cursor.is_none() { break; } } filter_thread_items_by_search_term(codex_home, &mut all_items, search_term).await?; let mut keyed_items = all_items .into_iter() .filter_map(|item| thread_item_sort_key(&item, sort_key).map(|key| (key, item))) .collect::>(); keyed_items.sort_by_key(|(key, _)| *key); let mut all_items = keyed_items .into_iter() .map(|(_, item)| item) .collect::>(); if let Some(cursor) = cursor { let anchor = cursor.timestamp(); all_items .retain(|item| thread_item_sort_key(item, sort_key).is_some_and(|key| key.0 > anchor)); } let more_matches_available = all_items.len() > page_size || reached_scan_cap; all_items.truncate(page_size); let next_cursor = if more_matches_available { all_items .last() .and_then(|item| cursor_from_thread_item(item, sort_key)) } else { None }; Ok(ThreadsPage { items: all_items, next_cursor, num_scanned_files: scanned_files, reached_scan_cap, }) } async fn filter_thread_items_by_search_term( codex_home: &Path, items: &mut Vec, search_term: Option<&str>, ) -> std::io::Result<()> { let Some(search_term) = search_term else { return Ok(()); }; // The file-backed fallback only has the thread title in the sidecar session index. // Match the SQLite path's title substring filter so search pagination behaves the same // whether the state DB is available or not. let thread_ids = items .iter() .filter_map(|item| item.thread_id) .collect::>(); let thread_names = find_thread_names_by_ids(codex_home, &thread_ids).await?; items.retain(|item| { item.thread_id .and_then(|thread_id| thread_names.get(&thread_id)) .is_some_and(|title| title.contains(search_term)) }); Ok(()) } fn thread_item_sort_key( item: &ThreadItem, sort_key: ThreadSortKey, ) -> Option<(OffsetDateTime, uuid::Uuid)> { let file_name = item.path.file_name()?.to_str()?; let (created_at, id) = parse_timestamp_uuid_from_filename(file_name)?; let timestamp = match sort_key { ThreadSortKey::CreatedAt => created_at, ThreadSortKey::UpdatedAt => { let updated_at = item.updated_at.as_deref().or(item.created_at.as_deref())?; OffsetDateTime::parse(updated_at, &Rfc3339).ok()? } }; Some((timestamp, id)) } fn cursor_from_thread_item(item: &ThreadItem, sort_key: ThreadSortKey) -> Option { let (timestamp, _id) = thread_item_sort_key(item, sort_key)?; let cursor_token = timestamp.format(&Rfc3339).ok()?; parse_cursor(cursor_token.as_str()) } 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( config: &impl RolloutConfigView, conversation_id: ThreadId, ) -> std::io::Result { // Resolve ~/.codex/sessions/YYYY/MM/DD path. let timestamp = OffsetDateTime::now_local() .map_err(|e| IoError::other(format!("failed to get local time: {e}")))?; let mut dir = config.codex_home().to_path_buf(); dir.push(SESSIONS_SUBDIR); dir.push(timestamp.year().to_string()); 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) .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, }) } fn open_log_file(path: &Path) -> std::io::Result { let path = compression::materialize_rollout_for_append_blocking(path)?; let Some(parent) = path.parent() else { return Err(IoError::other(format!( "rollout path has no parent: {}", path.display() ))); }; fs::create_dir_all(parent)?; std::fs::OpenOptions::new() .append(true) .create(true) .open(path) } /// Mutable state owned by the background rollout writer. /// /// Items are first appended to `pending_items`; persist/flush/shutdown remove each item from that /// queue only after it is written successfully. I/O failures drop the file handle but keep the /// unwritten suffix so the next barrier can reopen the file and retry. struct RolloutWriterState { writer: Option, deferred_log_file_info: Option, pending_items: Vec, meta: Option, cwd: PathBuf, rollout_path: PathBuf, 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); } async fn flush_if_materialized(&mut self) { if self.is_deferred() { return; } if let Err(err) = self.flush().await { self.enter_recovery_mode(&err); } } async fn persist(&mut self) -> std::io::Result<()> { self.write_pending_with_recovery("persist").await } async fn flush(&mut self) -> std::io::Result<()> { if self.is_deferred() && self.pending_items.is_empty() { return Ok(()); } self.write_pending_with_recovery("flush").await } async fn shutdown(&mut self) -> std::io::Result<()> { if self.is_deferred() && self.pending_items.is_empty() { return Ok(()); } self.write_pending_with_recovery("shutdown").await } async fn write_pending_with_recovery(&mut self, operation: &str) -> std::io::Result<()> { match self.write_pending_once().await { Ok(()) => { self.last_logged_error = None; Ok(()) } Err(first_err) => { self.enter_recovery_mode(&first_err); warn!("failed to {operation} rollout writer; reopening and retrying: {first_err}"); match self.write_pending_once().await { Ok(()) => { self.last_logged_error = None; Ok(()) } Err(second_err) => { self.enter_recovery_mode(&second_err); warn!( "retrying rollout writer {operation} failed; first error: \ {first_err}; final error: {second_err}" ); Err(second_err) } } } } } fn is_deferred(&self) -> bool { self.writer.is_none() && self.deferred_log_file_info.is_some() } fn enter_recovery_mode(&mut self, err: &IoError) { let message = err.to_string(); if self.last_logged_error.as_ref() != Some(&message) { error!( "rollout writer failed for {}; buffered rollout items will be retried: {err}; \ error_kind={:?}; raw_os_error={:?}", self.rollout_path.display(), err.kind(), err.raw_os_error() ); } self.last_logged_error = Some(message); self.writer = None; } async fn ensure_writer_open(&mut self) -> std::io::Result<()> { if self.writer.is_some() { 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)?; self.writer = Some(JsonlWriter { file: tokio::fs::File::from_std(file), }); self.deferred_log_file_info = None; Ok(()) } async fn write_session_meta_if_needed(&mut self) -> std::io::Result<()> { let Some(session_meta) = self.meta.as_ref().cloned() else { return Ok(()); }; write_session_meta(self.writer.as_mut(), session_meta, &self.cwd).await?; self.meta = None; Ok(()) } async fn write_pending_once(&mut self) -> std::io::Result<()> { self.ensure_writer_open().await?; self.write_session_meta_if_needed().await?; self.write_pending_items_once().await?; if let Some(writer) = self.writer.as_mut() { writer.file.flush().await?; } Ok(()) } async fn write_pending_items_once(&mut self) -> std::io::Result<()> { let Some(writer) = self.writer.as_mut() else { return Err(IoError::other("rollout writer is not open")); }; 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; } written_count += 1; } if written_count > 0 { self.pending_items.drain(..written_count); } write_result } } async fn rollout_writer( file: Option, deferred_log_file_info: Option, 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 { RolloutCmd::AddItems(items) => { state.add_items(items); state.flush_if_materialized().await; } RolloutCmd::Persist { ack } => { let _ = ack.send(state.persist().await); } RolloutCmd::Flush { ack } => { let _ = ack.send(state.flush().await); } RolloutCmd::Shutdown { ack } => match state.shutdown().await { Ok(()) => { let _ = ack.send(Ok(())); break; } Err(err) => { let _ = ack.send(Err(err)); } }, } } Ok(()) } async fn write_session_meta( mut writer: Option<&mut JsonlWriter>, session_meta: SessionMeta, cwd: &Path, ) -> std::io::Result<()> { let git_info = if get_git_repo_root(cwd).is_some() { collect_git_info(cwd).await.map(|info| ProtocolGitInfo { commit_hash: info.commit_hash, branch: info.branch, repository_url: info.repository_url, }) } else { None }; let session_meta_line = SessionMetaLine { meta: session_meta, git: git_info, }; let rollout_item = RolloutItem::SessionMeta(session_meta_line); if let Some(writer) = writer.as_mut() { writer.write_rollout_item(&rollout_item).await?; } Ok(()) } /// Append one already-filtered rollout item to an existing rollout JSONL file. /// /// This is for metadata updates to unloaded threads. Live sessions should use /// `RolloutRecorder::record_canonical_items` so rollout writes remain ordered /// with the rest of the session stream. pub async fn append_rollout_item_to_path( rollout_path: &Path, item: &RolloutItem, ) -> std::io::Result<()> { let rollout_path = compression::materialize_rollout_for_append(rollout_path).await?; let file = tokio::fs::OpenOptions::new() .append(true) .open(rollout_path) .await?; let mut writer = JsonlWriter { file }; writer.write_rollout_item(item).await } struct JsonlWriter { file: tokio::fs::File, } #[derive(serde::Serialize)] struct RolloutLineRef<'a> { timestamp: String, #[serde(flatten)] item: &'a RolloutItem, } impl JsonlWriter { async fn write_rollout_item(&mut self, rollout_item: &RolloutItem) -> std::io::Result<()> { let timestamp_format: &[FormatItem] = format_description!( "[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond digits:3]Z" ); let timestamp = OffsetDateTime::now_utc() .format(timestamp_format) .map_err(|e| IoError::other(format!("failed to format timestamp: {e}")))?; let line = RolloutLineRef { timestamp, item: rollout_item, }; self.write_line(&line).await } async fn write_line(&mut self, item: &impl serde::Serialize) -> std::io::Result<()> { let mut json = serde_json::to_string(item)?; json.push('\n'); self.file.write_all(json.as_bytes()).await?; self.file.flush().await?; Ok(()) } } impl From for ThreadsPage { fn from(db_page: codex_state::ThreadsPage) -> Self { let items = db_page .items .into_iter() .map(thread_item_from_state_metadata) .collect(); Self { items, next_cursor: db_page.next_anchor.map(Into::into), num_scanned_files: db_page.num_scanned_rows, reached_scan_cap: false, } } } fn thread_item_from_state_metadata(item: codex_state::ThreadMetadata) -> ThreadItem { ThreadItem { path: item.rollout_path, thread_id: Some(item.id), first_user_message: item.first_user_message, preview: item.preview, cwd: Some(item.cwd), git_branch: item.git_branch, git_sha: item.git_sha, git_origin_url: item.git_origin_url, source: Some( serde_json::from_str(item.source.as_str()) .or_else(|_| serde_json::from_value(Value::String(item.source))) .unwrap_or(SessionSource::Unknown), ), parent_thread_id: None, agent_nickname: item.agent_nickname, agent_role: item.agent_role, model_provider: Some(item.model_provider), cli_version: Some(item.cli_version), created_at: Some(item.created_at.to_rfc3339_opts(SecondsFormat::Secs, true)), updated_at: Some(item.updated_at.to_rfc3339_opts(SecondsFormat::Millis, true)), } } async fn select_resume_path( page: &ThreadsPage, filter_cwd: Option<&Path>, default_provider: &str, ) -> Option { match filter_cwd { Some(cwd) => { for item in &page.items { if resume_candidate_matches_cwd( item.path.as_path(), item.cwd.as_deref(), cwd, default_provider, ) .await { return Some(item.path.clone()); } } None } None => page.items.first().map(|item| item.path.clone()), } } async fn resume_candidate_matches_cwd( rollout_path: &Path, cached_cwd: Option<&Path>, cwd: &Path, default_provider: &str, ) -> bool { if cached_cwd.is_some_and(|session_cwd| cwd_matches(session_cwd, cwd)) { return true; } if let Ok((items, _, _)) = RolloutRecorder::load_rollout_items(rollout_path).await && let Some(latest_turn_context_cwd) = items.iter().rev().find_map(|item| match item { RolloutItem::TurnContext(turn_context) => Some(turn_context.cwd.as_path()), RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) | RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) => None, }) { return cwd_matches(latest_turn_context_cwd, cwd); } metadata::extract_metadata_from_rollout(rollout_path, default_provider) .await .is_ok_and(|outcome| cwd_matches(outcome.metadata.cwd.as_path(), cwd)) } async fn select_resume_path_from_db_page( page: &codex_state::ThreadsPage, filter_cwd: Option<&Path>, default_provider: &str, ) -> Option { match filter_cwd { Some(cwd) => { for item in &page.items { if resume_candidate_matches_cwd( item.rollout_path.as_path(), Some(item.cwd.as_path()), cwd, default_provider, ) .await { return Some(item.rollout_path.clone()); } } None } None => page.items.first().map(|item| item.rollout_path.clone()), } } fn cwd_matches(session_cwd: &Path, cwd: &Path) -> bool { path_utils::paths_match_after_normalization(session_cwd, cwd) } #[cfg(test)] #[path = "recorder_tests.rs"] mod tests;