codex/codex-rs/thread-store/src/local/mod.rs

mod archive_thread;
mod create_thread;
mod delete_thread;
mod helpers;
mod history_revision;
mod list_threads;
mod live_writer;
mod model_context;
mod move_thread_to_section;
mod paginated_fork;
mod pending_thread_metadata;
mod projects;
mod read_thread;
mod revert_thread;
mod rollout_migration;
// This lands before the reader PRs that consume the shared lineage resolver.
#[allow(dead_code)]
mod rollout_lineage;
mod search_threads;
mod thread_attachments;
mod thread_history;
mod thread_history_materialization;
mod thread_rollout_resolver;
mod thread_sections;
mod unarchive_thread;
mod update_thread_metadata;

#[cfg(test)]
#[path = "compression_writer_tests.rs"]
mod compression_writer_tests;
#[cfg(test)]
#[path = "daybreak_metadata_tests.rs"]
mod daybreak_metadata_tests;
#[cfg(test)]
#[path = "pending_thread_metadata_tests.rs"]
mod pending_thread_metadata_tests;
#[cfg(test)]
#[path = "read_thread_tests.rs"]
mod read_thread_tests;
#[cfg(test)]
mod test_support;
#[cfg(test)]
#[path = "timestamp_metadata_tests.rs"]
mod timestamp_metadata_tests;

use codex_protocol::ThreadId;
use codex_protocol::protocol::ThreadHistoryMode;
use codex_rollout::RolloutItem;
use codex_rollout::RolloutRecorder;
use codex_rollout::StateDbHandle;
use codex_rollout::WriterLockCoordinator;
use codex_state::SqliteConfig;
use std::collections::HashMap;
use std::collections::hash_map::Entry;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::sync::OnceCell;
use tokio::sync::OwnedMutexGuard;
use tokio::sync::OwnedRwLockReadGuard;
use tokio::sync::OwnedRwLockWriteGuard;
use tokio::sync::RwLock;

use crate::AddThreadAttachmentOutcome;
use crate::AddThreadAttachmentParams;
use crate::AppendThreadItemsParams;
use crate::ArchiveThreadParams;
use crate::ArchiveThreadsParams;
use crate::CreateProjectParams;
use crate::CreateThreadParams;
use crate::CreateThreadSectionParams;
use crate::CreatedProject;
use crate::DeleteThreadParams;
use crate::DeleteThreadSectionParams;
use crate::DeleteThreadsParams;
use crate::DeletedProject;
use crate::ItemPage;
use crate::ListItemsParams;
use crate::ListProjectsParams;
use crate::ListThreadAttachmentThreadsParams;
use crate::ListThreadAttachmentsParams;
use crate::ListThreadSectionsParams;
use crate::ListThreadsParams;
use crate::ListTimelineParams;
use crate::ListTurnsParams;
use crate::LoadThreadHistoryParams;
use crate::MoveProjectParams;
use crate::MoveThreadToSectionParams;
use crate::PersistContext;
use crate::PrepareForkParams;
use crate::PreparedFork;
use crate::ProjectMoveOutcome;
use crate::ReadThreadByRolloutPathParams;
use crate::ReadThreadParams;
use crate::RemoveThreadAttachmentOutcome;
use crate::RemoveThreadAttachmentParams;
use crate::RenameThreadSectionParams;
use crate::ResumeThreadParams;
use crate::RevertThreadParams;
use crate::SearchThreadOccurrencesParams;
use crate::SearchThreadsParams;
use crate::StoredModelContext;
use crate::StoredProject;
use crate::StoredProjectsPage;
use crate::StoredThread;
use crate::StoredThreadHistory;
use crate::StoredThreadSection;
use crate::StoredThreadSectionsPage;
use crate::ThreadAttachmentOwnerPage;
use crate::ThreadAttachmentPage;
use crate::ThreadMetadataPatch;
use crate::ThreadOccurrenceSearchPage;
use crate::ThreadPage;
use crate::ThreadSearchPage;
use crate::ThreadStore;
use crate::ThreadStoreError;
use crate::ThreadStoreFuture;
use crate::ThreadStoreResult;
use crate::TimelinePage;
use crate::TurnPage;
use crate::UpdateProjectParams;
use crate::UpdateThreadMetadataParams;
use crate::UpdatedProject;

pub use rollout_migration::RolloutMigrationFailureReason;
pub use rollout_migration::RolloutMigrationMode;
pub use rollout_migration::RolloutMigrationOptions;
pub use rollout_migration::RolloutMigrationOutcome;
pub use rollout_migration::RolloutMigrationProgress;
pub use rollout_migration::RolloutMigrationReport;
pub use rollout_migration::RolloutMigrationStatus;

/// Local filesystem/SQLite-backed implementation of [`ThreadStore`].
///
/// Local storage has two compatibility surfaces. Rollout JSONL files are the
/// durable replay format and remain readable without SQLite, including older
/// files that encode metadata in `SessionMeta` items and name-index entries.
/// The SQLite state DB, when available, is the queryable metadata index used by
/// list/read paths for fast lookup.
///
/// Live appends still write canonical JSONL history, but append-derived
/// metadata is observed above the store and applied through
/// [`ThreadStore::update_thread_metadata`]. This implementation applies that
/// patch literally to SQLite while keeping the JSONL/name-index compatibility
/// behavior needed for SQLite-less reads, repair, and old local rollout files.
#[derive(Clone)]
pub struct LocalThreadStore {
    pub(super) config: LocalThreadStoreConfig,
    live_recorders: Arc<Mutex<HashMap<ThreadId, LiveRecorderEntry>>>,
    pending_thread_metadata: pending_thread_metadata::PendingThreadMetadataRegistry,
    live_writer_locks: Arc<LiveWriterLocks>,
    writer_lock_coordinator: Arc<WriterLockCoordinator>,
    state_db: Option<StateDbHandle>,
    thread_history_db: Arc<OnceCell<sqlx::SqlitePool>>,
    thread_data_cleanup: Option<Arc<ThreadDataCleanup>>,
    #[cfg(test)]
    startup_migration_retry_gate: Option<Arc<StartupMigrationRetryGate>>,
}

type ThreadDataCleanup = dyn Fn(Vec<ThreadId>) -> ThreadStoreFuture<'static, ()> + Send + Sync;

type WriterLockGuard = Arc<codex_rollout::WriterLockGuard>;

struct LiveRecorderEntry {
    recorder: RolloutRecorder,
    // Rollout projection rows are keyed by immutable rollout ID, not the stable thread ID used
    // to find this live writer.
    rollout_id: ThreadId,
    // Local rollout files are materialized lazily, but metadata updates can arrive before the
    // canonical SessionMeta is durable. Retain the mode captured when live persistence was opened
    // so missing SQLite rows can still be seeded.
    history_mode: ThreadHistoryMode,
    writer_lock: WriterLockGuard,
}

#[derive(Default)]
struct LiveWriterLocks {
    // Keep per-thread locks after a writer goes idle. Removing one while another caller is about
    // to acquire it could let two operations for the same thread run at once.
    by_thread: Mutex<HashMap<ThreadId, Arc<ThreadCoordination>>>,
}

#[derive(Default)]
struct ThreadCoordination {
    // Serialize writes and capture consistent fork snapshots.
    writer: Arc<Mutex<()>>,
    // Forks hold a shared lease until their child reference is durable; deletion, archive, and
    // unarchive require exclusive access. Keeping this separate from `writer` lets the source
    // accept writes during child initialization, including MCP startup that can take 30 seconds.
    // Operations that need both locks must acquire `lifecycle` before `writer`.
    lifecycle: Arc<RwLock<()>>,
}

impl LiveWriterLocks {
    async fn coordination(&self, thread_id: ThreadId) -> Arc<ThreadCoordination> {
        self.by_thread
            .lock()
            .await
            .entry(thread_id)
            .or_default()
            .clone()
    }

    async fn lock(&self, thread_id: ThreadId) -> OwnedMutexGuard<()> {
        self.coordination(thread_id)
            .await
            .writer
            .clone()
            .lock_owned()
            .await
    }

    async fn reserve_lifecycle(&self, thread_id: ThreadId) -> OwnedRwLockReadGuard<()> {
        self.coordination(thread_id)
            .await
            .lifecycle
            .clone()
            .read_owned()
            .await
    }

    async fn lock_lifecycle(&self, thread_id: ThreadId) -> OwnedRwLockWriteGuard<()> {
        self.coordination(thread_id)
            .await
            .lifecycle
            .clone()
            .write_owned()
            .await
    }
}

// Per-store synchronization for exercising startup contention through the public entry point.
#[cfg(test)]
#[derive(Default)]
struct StartupMigrationRetryGate {
    retry_started: tokio::sync::Notify,
    retry_allowed: tokio::sync::Notify,
}

/// Process-scoped configuration for local thread storage.
///
/// This describes where local storage lives. New-thread rollout metadata such
/// as cwd, provider, and memory mode is supplied when live persistence is opened.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct LocalThreadStoreConfig {
    pub codex_home: PathBuf,
    pub sqlite: SqliteConfig,
    /// Provider used only when older local metadata does not contain one.
    pub default_model_provider_id: String,
}

impl LocalThreadStoreConfig {
    pub fn from_config(config: &impl codex_rollout::RolloutConfigView) -> Self {
        Self {
            codex_home: config.codex_home().to_path_buf(),
            sqlite: config.sqlite_config().clone(),
            default_model_provider_id: config.model_provider_id().to_string(),
        }
    }
}

impl std::fmt::Debug for LocalThreadStore {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("LocalThreadStore")
            .field("config", &self.config)
            .finish_non_exhaustive()
    }
}

impl LocalThreadStore {
    /// Create a local store using an already initialized state DB handle.
    pub fn new(config: LocalThreadStoreConfig, state_db: Option<StateDbHandle>) -> Self {
        let writer_lock_coordinator = Arc::new(WriterLockCoordinator::new(&config.codex_home));
        Self {
            config,
            live_recorders: Arc::new(Mutex::new(HashMap::new())),
            pending_thread_metadata:
                pending_thread_metadata::PendingThreadMetadataRegistry::default(),
            live_writer_locks: Arc::new(LiveWriterLocks::default()),
            writer_lock_coordinator,
            state_db,
            thread_history_db: Arc::new(OnceCell::new()),
            thread_data_cleanup: None,
            #[cfg(test)]
            startup_migration_retry_gate: None,
        }
    }

    /// Adds cleanup for host-owned durable data when threads are permanently deleted.
    /// Runs after reference and writer checks, under the lifecycle locks, before deleting rollouts.
    /// Cleanup must be idempotent: later deletion steps can fail and the caller may retry.
    pub fn with_thread_data_cleanup(
        mut self,
        cleanup: impl Fn(Vec<ThreadId>) -> ThreadStoreFuture<'static, ()> + Send + Sync + 'static,
    ) -> Self {
        self.thread_data_cleanup = Some(Arc::new(cleanup));
        self
    }

    /// Return the state DB handle used by local rollout writers.
    pub async fn state_db(&self) -> Option<StateDbHandle> {
        self.state_db.clone()
    }

    async fn thread_history_db(&self) -> ThreadStoreResult<&sqlx::SqlitePool> {
        if self.state_db.is_none() {
            return Err(ThreadStoreError::Unsupported {
                operation: "paginated_history",
            });
        }
        self.thread_history_db
            .get_or_try_init(|| async {
                codex_state::open_thread_history_db(&self.config.sqlite).await
            })
            .await
            .map_err(|err| ThreadStoreError::Internal {
                message: format!("failed to open thread history database: {err}"),
            })
    }

    /// Read a local rollout-backed thread by path.
    pub async fn read_thread_by_rollout_path(
        &self,
        rollout_path: PathBuf,
        include_archived: bool,
        include_history: bool,
    ) -> ThreadStoreResult<StoredThread> {
        read_thread::read_thread_by_rollout_path(
            self,
            rollout_path,
            include_archived,
            include_history,
        )
        .await
    }

    /// Return the live local rollout path for legacy local-only code paths.
    pub async fn live_rollout_path(&self, thread_id: ThreadId) -> ThreadStoreResult<PathBuf> {
        live_writer::rollout_path(self, thread_id).await
    }

    /// Select only materialized rollout inputs without creating a file for untouched threads.
    pub(crate) async fn feedback_rollout_path(
        &self,
        thread_id: ThreadId,
    ) -> ThreadStoreResult<Option<PathBuf>> {
        let (recorder, _, _) = live_writer::live_writer_parts(self, thread_id).await?;
        if recorder
            .unpersisted_session_meta()
            .await
            .map_err(|error| ThreadStoreError::Internal {
                message: error.to_string(),
            })?
            .is_some()
        {
            return Ok(None);
        }
        // Keep the path even if a materialized file was deleted: collection must report that error.
        Ok(Some(recorder.rollout_path().to_path_buf()))
    }

    pub(super) async fn ensure_live_recorder_absent(
        &self,
        thread_id: ThreadId,
    ) -> ThreadStoreResult<()> {
        if self.live_recorders.lock().await.contains_key(&thread_id) {
            return Err(ThreadStoreError::InvalidRequest {
                message: format!("thread {thread_id} already has a live local writer"),
            });
        }
        Ok(())
    }

    fn acquire_writer_lock(&self, thread_id: ThreadId) -> ThreadStoreResult<WriterLockGuard> {
        self.writer_lock_coordinator
            .acquire(thread_id)
            .map(Arc::new)
            .map_err(|err| {
                if err.kind() == std::io::ErrorKind::WouldBlock {
                    ThreadStoreError::Conflict {
                        message: err.to_string(),
                    }
                } else {
                    ThreadStoreError::Internal {
                        message: err.to_string(),
                    }
                }
            })
    }

    async fn acquire_writer_locks(
        &self,
        thread_ids: &[ThreadId],
    ) -> ThreadStoreResult<Vec<WriterLockGuard>> {
        let mut writer_locks = Vec::with_capacity(thread_ids.len());
        for &thread_id in thread_ids {
            if self.live_recorders.lock().await.contains_key(&thread_id) {
                continue;
            }
            writer_locks.push(self.acquire_writer_lock(thread_id)?);
        }
        Ok(writer_locks)
    }

    async fn insert_live_recorder(
        &self,
        thread_id: ThreadId,
        recorder: RolloutRecorder,
        rollout_id: ThreadId,
        history_mode: ThreadHistoryMode,
        writer_lock: WriterLockGuard,
    ) -> ThreadStoreResult<()> {
        match self.live_recorders.lock().await.entry(thread_id) {
            Entry::Occupied(entry) => Err(ThreadStoreError::InvalidRequest {
                message: format!("thread {} already has a live local writer", entry.key()),
            }),
            Entry::Vacant(entry) => {
                entry.insert(LiveRecorderEntry {
                    recorder,
                    rollout_id,
                    history_mode,
                    writer_lock,
                });
                Ok(())
            }
        }
    }

    async fn load_history(
        &self,
        params: LoadThreadHistoryParams,
    ) -> ThreadStoreResult<StoredThreadHistory> {
        if let Ok(rollout_path) = live_writer::rollout_path(self, params.thread_id).await {
            if !params.include_archived
                && helpers::rollout_path_is_archived(
                    self.config.codex_home.as_path(),
                    rollout_path.as_path(),
                )
            {
                return Err(ThreadStoreError::InvalidRequest {
                    message: format!("thread {} is archived", params.thread_id),
                });
            }
            return read_thread::read_thread_by_rollout_path(
                self,
                rollout_path,
                /*include_archived*/ true,
                /*include_history*/ true,
            )
            .await?
            .history
            .ok_or_else(|| ThreadStoreError::Internal {
                message: format!("failed to load history for thread {}", params.thread_id),
            });
        }

        read_thread::read_thread(
            self,
            ReadThreadParams {
                thread_id: params.thread_id,
                include_archived: params.include_archived,
                include_history: true,
            },
        )
        .await?
        .history
        .ok_or_else(|| ThreadStoreError::Internal {
            message: format!("failed to load history for thread {}", params.thread_id),
        })
    }

    async fn read_thread_by_rollout_path_params(
        &self,
        params: ReadThreadByRolloutPathParams,
    ) -> ThreadStoreResult<StoredThread> {
        read_thread::read_thread_by_rollout_path(
            self,
            params.rollout_path,
            params.include_archived,
            params.include_history,
        )
        .await
    }

    /// Lists projection-backed turns without enabling app-server routing yet.
    pub async fn list_turns(&self, params: ListTurnsParams) -> ThreadStoreResult<TurnPage> {
        thread_history::list_turns(self, params).await
    }

    /// Lists projection-backed items without enabling app-server routing yet.
    pub async fn list_items(&self, params: ListItemsParams) -> ThreadStoreResult<ItemPage> {
        thread_history::list_items(self, params).await
    }

    /// Lists bounded ordinary and realtime items from the rollout projection.
    pub async fn list_timeline(
        &self,
        params: ListTimelineParams,
    ) -> ThreadStoreResult<TimelinePage> {
        thread_history::list_timeline(self, params).await
    }

    /// Searches projection-backed visible messages within one paginated thread.
    pub async fn search_thread_occurrences(
        &self,
        params: SearchThreadOccurrencesParams,
    ) -> ThreadStoreResult<ThreadOccurrenceSearchPage> {
        thread_history::search_thread_occurrences(self, params).await
    }
}

impl ThreadStore for LocalThreadStore {
    fn default_history_mode(&self) -> ThreadHistoryMode {
        ThreadHistoryMode::Paginated
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    fn create_thread(&self, params: CreateThreadParams) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move { live_writer::create_thread(self, params).await })
    }

    fn stage_pending_thread_metadata(
        &self,
        thread_id: ThreadId,
        patch: ThreadMetadataPatch,
    ) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move {
            if self.state_db.is_none() {
                return Err(ThreadStoreError::InvalidRequest {
                    message: "pending thread metadata requires a state db".to_string(),
                });
            }
            if patch.rollout_path.is_some() {
                return Err(ThreadStoreError::InvalidRequest {
                    message: "pending thread metadata cannot set rollout_path".to_string(),
                });
            }
            self.pending_thread_metadata.stage(thread_id, patch).await
        })
    }

    fn read_pending_thread_metadata(
        &self,
        thread_id: ThreadId,
    ) -> ThreadStoreFuture<'_, Option<ThreadMetadataPatch>> {
        Box::pin(async move {
            Ok(self
                .pending_thread_metadata
                .lock(thread_id)
                .await
                .and_then(|metadata| metadata.clone()))
        })
    }

    fn remove_pending_thread_metadata(&self, thread_id: ThreadId) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move {
            self.pending_thread_metadata.remove(thread_id).await;
            Ok(())
        })
    }

    fn resume_thread(
        &self,
        params: ResumeThreadParams,
    ) -> ThreadStoreFuture<'_, Arc<Vec<RolloutItem>>> {
        Box::pin(async move { live_writer::resume_thread(self, params).await })
    }

    fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move { live_writer::append_items(self, params).await })
    }

    fn persist_thread(
        &self,
        thread_id: ThreadId,
        context: PersistContext,
    ) -> ThreadStoreFuture<'_, ()> {
        if context == PersistContext::SubagentSpawn {
            return Box::pin(async { Ok(()) });
        }
        Box::pin(async move {
            if context == PersistContext::ThreadPreparation {
                live_writer::flush_thread(self, thread_id).await
            } else {
                live_writer::persist_thread(self, thread_id).await
            }
        })
    }

    fn flush_thread(&self, thread_id: ThreadId) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move { live_writer::flush_thread(self, thread_id).await })
    }

    fn shutdown_thread(&self, thread_id: ThreadId) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move { live_writer::shutdown_thread(self, thread_id).await })
    }

    fn discard_thread(&self, thread_id: ThreadId) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move { live_writer::discard_thread(self, thread_id).await })
    }

    fn load_history(
        &self,
        params: LoadThreadHistoryParams,
    ) -> ThreadStoreFuture<'_, StoredThreadHistory> {
        Box::pin(LocalThreadStore::load_history(self, params))
    }

    fn load_latest_model_context(
        &self,
        params: LoadThreadHistoryParams,
    ) -> ThreadStoreFuture<'_, StoredModelContext> {
        Box::pin(async move { model_context::load_latest_model_context(self, params).await })
    }

    fn is_model_context_current(
        &self,
        params: LoadThreadHistoryParams,
        revision: String,
    ) -> ThreadStoreFuture<'_, bool> {
        Box::pin(async move {
            let resolved = if params.include_archived {
                thread_rollout_resolver::resolve_current_including_archived(self, params.thread_id)
                    .await?
            } else {
                thread_rollout_resolver::resolve_current(self, params.thread_id).await?
            };
            let Some(resolved) = resolved else {
                return Ok(false);
            };
            Ok(history_revision::read(&resolved.path).await.as_ref() == Some(&revision))
        })
    }

    fn prepare_fork(&self, params: PrepareForkParams) -> ThreadStoreFuture<'_, PreparedFork> {
        Box::pin(async move { paginated_fork::prepare(self, params).await })
    }

    fn revert_thread(&self, params: RevertThreadParams) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move { revert_thread::revert(self, params).await })
    }

    fn read_thread(&self, params: ReadThreadParams) -> ThreadStoreFuture<'_, StoredThread> {
        Box::pin(async move { read_thread::read_thread(self, params).await })
    }

    fn read_thread_by_rollout_path(
        &self,
        params: ReadThreadByRolloutPathParams,
    ) -> ThreadStoreFuture<'_, StoredThread> {
        Box::pin(LocalThreadStore::read_thread_by_rollout_path_params(
            self, params,
        ))
    }

    fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreFuture<'_, ThreadPage> {
        Box::pin(async move { list_threads::list_threads(self, params).await })
    }

    fn supports_thread_sections(&self) -> bool {
        self.state_db.is_some()
    }

    fn list_thread_sections(
        &self,
        params: ListThreadSectionsParams,
    ) -> ThreadStoreFuture<'_, StoredThreadSectionsPage> {
        Box::pin(async move { thread_sections::list_thread_sections(self, params).await })
    }

    fn create_thread_section(
        &self,
        params: CreateThreadSectionParams,
    ) -> ThreadStoreFuture<'_, StoredThreadSection> {
        Box::pin(async move { thread_sections::create_thread_section(self, params).await })
    }

    fn rename_thread_section(
        &self,
        params: RenameThreadSectionParams,
    ) -> ThreadStoreFuture<'_, Option<StoredThreadSection>> {
        Box::pin(async move { thread_sections::rename_thread_section(self, params).await })
    }

    fn delete_thread_section(
        &self,
        params: DeleteThreadSectionParams,
    ) -> ThreadStoreFuture<'_, bool> {
        Box::pin(async move { thread_sections::delete_thread_section(self, params).await })
    }

    fn supports_thread_attachments(&self) -> bool {
        self.state_db.is_some()
    }

    fn copy_thread_attachments(
        &self,
        source_thread_id: ThreadId,
        destination_thread_id: ThreadId,
    ) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move {
            thread_attachments::copy_thread_attachments(
                self,
                source_thread_id,
                destination_thread_id,
            )
            .await
        })
    }

    fn add_thread_attachment(
        &self,
        params: AddThreadAttachmentParams,
    ) -> ThreadStoreFuture<'_, AddThreadAttachmentOutcome> {
        Box::pin(async move { thread_attachments::add_thread_attachment(self, params).await })
    }

    fn list_thread_attachments(
        &self,
        params: ListThreadAttachmentsParams,
    ) -> ThreadStoreFuture<'_, ThreadAttachmentPage> {
        Box::pin(async move { thread_attachments::list_thread_attachments(self, params).await })
    }

    fn list_thread_attachment_threads(
        &self,
        params: ListThreadAttachmentThreadsParams,
    ) -> ThreadStoreFuture<'_, ThreadAttachmentOwnerPage> {
        Box::pin(
            async move { thread_attachments::list_thread_attachment_threads(self, params).await },
        )
    }

    fn remove_thread_attachment(
        &self,
        params: RemoveThreadAttachmentParams,
    ) -> ThreadStoreFuture<'_, RemoveThreadAttachmentOutcome> {
        Box::pin(async move { thread_attachments::remove_thread_attachment(self, params).await })
    }

    fn supports_projects(&self) -> bool {
        self.state_db.is_some()
    }

    fn list_projects(
        &self,
        params: ListProjectsParams,
    ) -> ThreadStoreFuture<'_, StoredProjectsPage> {
        Box::pin(async move { projects::list_projects(self, params).await })
    }

    fn read_project(&self, project_id: String) -> ThreadStoreFuture<'_, Option<StoredProject>> {
        Box::pin(async move { projects::read_project(self, project_id).await })
    }

    fn create_project(&self, params: CreateProjectParams) -> ThreadStoreFuture<'_, CreatedProject> {
        Box::pin(async move { projects::create_project(self, params).await })
    }

    fn update_project(
        &self,
        params: UpdateProjectParams,
    ) -> ThreadStoreFuture<'_, Option<UpdatedProject>> {
        Box::pin(async move { projects::update_project(self, params).await })
    }

    fn move_project(
        &self,
        params: MoveProjectParams,
    ) -> ThreadStoreFuture<'_, Option<ProjectMoveOutcome>> {
        Box::pin(async move { projects::move_project(self, params).await })
    }

    fn delete_project(&self, project_id: String) -> ThreadStoreFuture<'_, Option<DeletedProject>> {
        Box::pin(async move { projects::delete_project(self, project_id).await })
    }

    fn supports_paginated_history_lists(&self) -> bool {
        self.state_db.is_some()
    }

    fn list_turns(&self, params: ListTurnsParams) -> ThreadStoreFuture<'_, TurnPage> {
        Box::pin(LocalThreadStore::list_turns(self, params))
    }

    fn list_items(&self, params: ListItemsParams) -> ThreadStoreFuture<'_, ItemPage> {
        Box::pin(LocalThreadStore::list_items(self, params))
    }

    fn list_timeline(&self, params: ListTimelineParams) -> ThreadStoreFuture<'_, TimelinePage> {
        Box::pin(LocalThreadStore::list_timeline(self, params))
    }

    fn search_threads(
        &self,
        params: SearchThreadsParams,
    ) -> ThreadStoreFuture<'_, ThreadSearchPage> {
        Box::pin(async move { search_threads::search_threads(self, params).await })
    }

    fn search_thread_occurrences(
        &self,
        params: SearchThreadOccurrencesParams,
    ) -> ThreadStoreFuture<'_, ThreadOccurrenceSearchPage> {
        Box::pin(LocalThreadStore::search_thread_occurrences(self, params))
    }

    fn update_thread_metadata(
        &self,
        params: UpdateThreadMetadataParams,
    ) -> ThreadStoreFuture<'_, Option<StoredThread>> {
        Box::pin(async move {
            update_thread_metadata::update_thread_metadata(self, params)
                .await
                .map(Some)
        })
    }

    fn move_thread_to_section(
        &self,
        params: MoveThreadToSectionParams,
    ) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move { move_thread_to_section::move_thread_to_section(self, params).await })
    }

    fn archive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move {
            archive_thread::archive_threads(
                self,
                ArchiveThreadsParams {
                    thread_ids: vec![params.thread_id],
                    writer_lock_thread_ids: Vec::new(),
                },
            )
            .await
            .map(|_| ())
        })
    }

    fn archive_threads(
        &self,
        params: ArchiveThreadsParams,
    ) -> ThreadStoreFuture<'_, Vec<ThreadId>> {
        Box::pin(async move { archive_thread::archive_threads(self, params).await })
    }

    fn unarchive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, StoredThread> {
        Box::pin(async move {
            unarchive_thread::unarchive_threads(self, params, Vec::new())
                .await
                .map(|(thread, _)| thread)
        })
    }

    fn unarchive_threads(
        &self,
        params: ArchiveThreadParams,
        descendant_thread_ids: Vec<ThreadId>,
    ) -> ThreadStoreFuture<'_, (StoredThread, Vec<ThreadId>)> {
        Box::pin(async move {
            unarchive_thread::unarchive_threads(self, params, descendant_thread_ids).await
        })
    }

    fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move { delete_thread::delete_thread(self, params).await })
    }

    fn delete_threads(&self, params: DeleteThreadsParams) -> ThreadStoreFuture<'_, ()> {
        Box::pin(async move { delete_thread::delete_threads(self, params).await })
    }
}

#[cfg(test)]
mod tests {
    #[path = "acquisition_tests.rs"]
    mod acquisition_tests;
    #[path = "delegated_preview_tests.rs"]
    mod delegated_preview_tests;
    use std::sync::Arc;

    use codex_protocol::ThreadId;
    use codex_protocol::config_types::ReasoningSummary;
    use codex_protocol::items::TurnItem;
    use codex_protocol::items::UserMessageItem;
    use codex_protocol::models::BaseInstructions;
    use codex_protocol::models::ContentItem;
    use codex_protocol::models::FunctionCallOutputPayload;
    use codex_protocol::models::MessagePhase;
    use codex_protocol::models::ResponseItem;
    use codex_protocol::protocol::AgentMessageEvent;
    use codex_protocol::protocol::AskForApproval;
    use codex_protocol::protocol::EventMsg;
    use codex_protocol::protocol::ItemCompletedEvent;
    use codex_protocol::protocol::SandboxPolicy;
    use codex_protocol::protocol::SessionSource;
    use codex_protocol::protocol::ThreadHistoryMode;
    use codex_protocol::protocol::ThreadMemoryMode;
    use codex_protocol::protocol::TurnCompleteEvent;
    use codex_protocol::protocol::TurnContextItem;
    use codex_protocol::protocol::TurnStartedEvent;
    use codex_protocol::protocol::UserMessageEvent;
    use codex_rollout::RolloutItem;
    use tempfile::TempDir;

    use super::*;
    use crate::LiveThread;
    use crate::ThreadPersistenceMetadata;
    use crate::local::test_support::test_config;
    use crate::local::test_support::write_archived_session_file;
    use crate::local::test_support::write_session_file;
    use crate::local::test_support::write_session_file_with_history_mode;

    #[tokio::test]
    async fn live_writer_lifecycle_writes_and_closes() {
        let home = TempDir::new().expect("temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let thread_id = ThreadId::default();

        store
            .create_thread(create_thread_params(thread_id))
            .await
            .expect("create live thread");
        let rollout_path = store
            .live_rollout_path(thread_id)
            .await
            .expect("load rollout path");

        store
            .append_items(AppendThreadItemsParams {
                thread_id,
                items: vec![user_message_item("first live write")],
            })
            .await
            .expect("append live item");
        store
            .persist_thread(thread_id, PersistContext::Standard)
            .await
            .expect("persist live thread");
        store
            .flush_thread(thread_id)
            .await
            .expect("flush live thread");

        assert_rollout_contains_message(rollout_path.as_path(), "first live write").await;

        store
            .shutdown_thread(thread_id)
            .await
            .expect("shutdown live thread");
        let err = store
            .append_items(AppendThreadItemsParams {
                thread_id,
                items: vec![user_message_item("write after shutdown")],
            })
            .await
            .expect_err("shutdown should remove the live thread writer");
        assert!(
            matches!(err, ThreadStoreError::ThreadNotFound { thread_id: missing } if missing == thread_id)
        );
    }

    #[tokio::test]
    async fn raw_append_items_does_not_update_sqlite_metadata() {
        // This pins the ThreadStore contract: raw appends are history-only. Callers that need
        // metadata updates must use LiveThread or call update_thread_metadata explicitly.
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let store = LocalThreadStore::new(config, Some(runtime.clone()));
        let thread_id = ThreadId::default();

        store
            .create_thread(create_thread_params(thread_id))
            .await
            .expect("create live thread");
        store
            .append_items(AppendThreadItemsParams {
                thread_id,
                items: vec![user_message_item("raw append")],
            })
            .await
            .expect("append raw item");
        store.flush_thread(thread_id).await.expect("flush thread");

        assert_eq!(
            runtime
                .get_thread(thread_id)
                .await
                .expect("sqlite metadata read"),
            None
        );
    }

    #[tokio::test]
    async fn live_thread_observes_appended_items_into_sqlite_metadata() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone())));
        let thread_id = ThreadId::default();
        let live_thread = LiveThread::create(store.clone(), create_thread_params(thread_id))
            .await
            .expect("create live thread");

        live_thread
            .append_items(&[user_message_item("observed append")])
            .await
            .expect("append observed item");
        live_thread.flush().await.expect("flush thread");

        let metadata = runtime
            .get_thread(thread_id)
            .await
            .expect("sqlite metadata read")
            .expect("sqlite metadata");
        assert_eq!(
            metadata.first_user_message.as_deref(),
            Some("observed append")
        );
        assert_eq!(metadata.preview.as_deref(), Some("observed append"));
        assert_eq!(metadata.title, "observed append");
    }

    #[tokio::test]
    async fn paginated_resume_prefers_explicit_rollout_path_over_stale_sqlite_path() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone())));
        let uuid = uuid::Uuid::from_u128(228);
        let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
        let rollout_path = write_session_file_with_history_mode(
            home.path(),
            "2025-01-03T12-00-00",
            uuid,
            ThreadHistoryMode::Paginated,
        )
        .expect("paginated session file");
        let stale_rollout_path = home.path().join("stale-rollout.jsonl");
        tokio::fs::write(&stale_rollout_path, "malformed session metadata\n")
            .await
            .expect("write stale rollout");
        let mut builder = codex_state::ThreadMetadataBuilder::new(
            thread_id,
            stale_rollout_path,
            chrono::Utc::now(),
            SessionSource::Cli,
        );
        builder.history_mode = ThreadHistoryMode::Paginated;
        builder.cwd = home.path().to_path_buf();
        let mut metadata = builder.build("test-provider");
        metadata.preview = Some("original user message".to_string());
        metadata.first_user_message = Some("original user message".to_string());
        metadata.title = "original user message".to_string();
        runtime
            .upsert_thread(&metadata)
            .await
            .expect("update stale sqlite rollout path");

        let history = Arc::new(vec![user_message_item("bounded suffix")]);
        let (resumed, supplied_history) = LiveThread::resume(
            store,
            ThreadHistoryMode::Paginated,
            ResumeThreadParams {
                history_revision: None,
                thread_id,
                rollout_path: Some(rollout_path.clone()),
                history: Some(Arc::clone(&history)),
                include_archived: false,
                metadata: ThreadPersistenceMetadata {
                    cwd: Some(home.path().to_path_buf()),
                    model_provider: "test-provider".to_string(),
                    memory_mode: ThreadMemoryMode::Enabled,
                },
            },
            /*resume_preparation*/ None,
        )
        .await
        .expect("resume paginated thread from its requested rollout");
        assert!(Arc::ptr_eq(&supplied_history, &history));
        assert_eq!(
            resumed.local_rollout_path().await.expect("live rollout"),
            Some(std::fs::canonicalize(rollout_path).expect("canonical rollout path"))
        );
        resumed.shutdown().await.expect("shutdown resumed writer");

        let metadata = runtime
            .get_thread(thread_id)
            .await
            .expect("sqlite metadata read")
            .expect("sqlite metadata");
        assert_eq!(
            (
                metadata.preview.as_deref(),
                metadata.title.as_str(),
                metadata.first_user_message.as_deref(),
            ),
            (
                Some("original user message"),
                "original user message",
                Some("original user message"),
            )
        );
    }

    #[tokio::test]
    async fn live_thread_does_not_derive_metadata_from_inherited_items() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone())));
        let thread_id = ThreadId::default();
        let mut params = create_thread_params(thread_id);
        params.history_mode = ThreadHistoryMode::Paginated;
        let cwd = std::env::current_dir().expect("current directory");
        let turn_context = |model: &str, approval_policy| {
            RolloutItem::TurnContext(TurnContextItem {
                turn_id: Some("turn-1".to_string()),
                root_turn_id: None,
                direct_only_tool_namespaces: None,
                disabled_plugin_ids: None,
                cwd: serde_json::from_value(serde_json::json!(cwd)).expect("absolute cwd"),
                approval_policy,
                approvals_reviewer: None,
                sandbox_policy: SandboxPolicy::DangerFullAccess,
                permission_profile: None,
                active_permission_profile: None,
                file_system_sandbox_policy: None,
                model: model.to_string(),
                comp_hash: None,
                collaboration_mode: None,
                multi_agent_version: None,
                realtime_active: None,
                cyber_access_program: None,
                effort: None,
                summary: Some(ReasoningSummary::Auto),
            })
        };

        let mut guard = crate::LiveThreadInitGuard::default();
        let live_thread = LiveThread::create_with_inherited_model_context(
            store,
            params,
            &[turn_context("parent-model", AskForApproval::Never)],
            &mut guard,
        )
        .await
        .expect("create live thread with inherited context");
        guard.commit();
        live_thread
            .persist(PersistContext::Standard)
            .await
            .expect("persist thread");
        let inherited_metadata = runtime
            .get_thread(thread_id)
            .await
            .expect("sqlite metadata read")
            .expect("sqlite metadata");
        assert_eq!(inherited_metadata.model, None);

        live_thread
            .append_items(&[turn_context("child-model", AskForApproval::OnRequest)])
            .await
            .expect("append child context");
        let child_metadata = runtime
            .get_thread(thread_id)
            .await
            .expect("sqlite metadata read")
            .expect("sqlite metadata");
        assert_eq!(child_metadata.model.as_deref(), Some("child-model"));
        assert_eq!(child_metadata.approval_mode, "on-request");
    }

    #[tokio::test]
    async fn live_thread_output_advances_updated_at_but_not_recency_at() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone())));
        let thread_id = ThreadId::default();
        let live_thread = LiveThread::create(store, create_thread_params(thread_id))
            .await
            .expect("create live thread");

        live_thread
            .append_items(&[user_message_item("start thread")])
            .await
            .expect("append initial user message");
        live_thread.flush().await.expect("flush thread");
        let before_turn_start = runtime
            .get_thread(thread_id)
            .await
            .expect("sqlite metadata read")
            .expect("sqlite metadata");

        live_thread
            .append_items(&[RolloutItem::EventMsg(EventMsg::TurnStarted(
                TurnStartedEvent {
                    turn_attribution: None,
                    turn_id: "turn-1".to_string(),
                    root_turn_id: None,
                    trace_id: None,
                    started_at: None,
                    model_context_window: None,
                    collaboration_mode_kind: Default::default(),
                },
            ))])
            .await
            .expect("append turn start");
        live_thread.flush().await.expect("flush thread");
        let after_turn_start = runtime
            .get_thread(thread_id)
            .await
            .expect("sqlite metadata read")
            .expect("sqlite metadata");
        assert!(after_turn_start.recency_at > before_turn_start.recency_at);

        live_thread
            .append_items(&[
                RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent {
                    message: "commentary".to_string(),
                    phase: Some(MessagePhase::Commentary),
                    memory_citation: None,
                    delivery: None,
                    questions: None,
                })),
                RolloutItem::ResponseItem(
                    ResponseItem::FunctionCallOutput {
                        id: None,
                        call_id: Some("call-1".to_string()),
                        name: None,
                        namespace: None,
                        output: FunctionCallOutputPayload::from_text("tool output".to_string()),
                        internal_chat_message_metadata_passthrough: None,
                    }
                    .into(),
                ),
                RolloutItem::EventMsg(EventMsg::TokenCount(
                    codex_protocol::protocol::TokenCountEvent {
                        info: None,
                        rate_limits: None,
                    },
                )),
                RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
                    root_turn_id: None,
                    turn_id: "turn-1".to_string(),
                    started_at: None,
                    last_agent_message: None,
                    error: None,
                    completed_at: None,
                    duration_ms: None,
                    time_to_first_token_ms: None,
                })),
            ])
            .await
            .expect("append post-start items");
        live_thread.flush().await.expect("flush thread");
        let completed = runtime
            .get_thread(thread_id)
            .await
            .expect("sqlite metadata read")
            .expect("sqlite metadata");

        assert!(completed.updated_at > after_turn_start.updated_at);
        assert_eq!(completed.recency_at, after_turn_start.recency_at);
    }

    #[tokio::test]
    async fn live_thread_shutdown_does_not_materialize_empty_thread_metadata() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone())));
        let thread_id = ThreadId::default();
        let live_thread = LiveThread::create(store.clone(), create_thread_params(thread_id))
            .await
            .expect("create live thread");
        let rollout_path = store
            .live_rollout_path(thread_id)
            .await
            .expect("live rollout path");

        live_thread.shutdown().await.expect("shutdown thread");

        assert!(
            !tokio::fs::try_exists(rollout_path.as_path())
                .await
                .expect("rollout path should be checkable")
        );
        assert_eq!(
            runtime
                .get_thread(thread_id)
                .await
                .expect("sqlite metadata read"),
            None
        );
    }

    #[tokio::test]
    async fn live_thread_memory_mode_update_before_rollout_materializes_keeps_history_mode() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone())));
        let thread_id = ThreadId::default();
        let live_thread = LiveThread::create(store.clone(), create_thread_params(thread_id))
            .await
            .expect("create live thread");

        live_thread
            .update_memory_mode(ThreadMemoryMode::Disabled, /*include_archived*/ false)
            .await
            .expect("update memory mode");

        assert_eq!(
            runtime
                .get_thread(thread_id)
                .await
                .expect("sqlite metadata read")
                .expect("sqlite metadata")
                .history_mode,
            ThreadHistoryMode::Legacy
        );
        assert_eq!(
            runtime
                .get_thread_memory_mode(thread_id)
                .await
                .expect("thread memory mode should be readable")
                .as_deref(),
            Some("disabled")
        );
    }

    #[tokio::test]
    async fn live_thread_shutdown_with_buffered_items_materializes_before_metadata_read() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone())));
        let thread_id = ThreadId::default();
        let live_thread = LiveThread::create(store.clone(), create_thread_params(thread_id))
            .await
            .expect("create live thread");
        let rollout_path = store
            .live_rollout_path(thread_id)
            .await
            .expect("live rollout path");

        live_thread
            .append_items(&[RolloutItem::EventMsg(EventMsg::TokenCount(
                codex_protocol::protocol::TokenCountEvent {
                    info: None,
                    rate_limits: None,
                },
            ))])
            .await
            .expect("append metadata-only item");
        live_thread.shutdown().await.expect("shutdown thread");

        assert!(
            tokio::fs::try_exists(rollout_path.as_path())
                .await
                .expect("rollout path should be checkable")
        );
        let metadata = runtime
            .get_thread(thread_id)
            .await
            .expect("sqlite metadata read")
            .expect("sqlite metadata");
        assert_eq!(metadata.rollout_path, rollout_path);
    }

    #[tokio::test]
    async fn live_thread_resume_loads_history_before_observing_metadata() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone())));
        let uuid = uuid::Uuid::from_u128(401);
        let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
        let rollout_path =
            write_session_file(home.path(), "2025-01-03T17-00-00", uuid).expect("session file");
        let (live_thread, _history) = LiveThread::resume(
            store,
            ThreadHistoryMode::Legacy,
            ResumeThreadParams {
                history_revision: None,
                thread_id,
                rollout_path: Some(rollout_path),
                history: None,
                include_archived: false,
                metadata: ThreadPersistenceMetadata {
                    cwd: Some(home.path().to_path_buf()),
                    model_provider: "different-provider".to_string(),
                    memory_mode: ThreadMemoryMode::Enabled,
                },
            },
            /*resume_preparation*/ None,
        )
        .await
        .expect("resume live thread");

        live_thread
            .append_items(&[user_message_item("new live append")])
            .await
            .expect("append after resume");

        let metadata = runtime
            .get_thread(thread_id)
            .await
            .expect("sqlite metadata read")
            .expect("sqlite metadata");
        assert_eq!(
            metadata.created_at.to_rfc3339(),
            "2025-01-03T17:00:00+00:00"
        );
        assert_eq!(metadata.model_provider, "test-provider");
        assert_eq!(
            metadata.first_user_message.as_deref(),
            Some("Hello from user")
        );
    }

    #[tokio::test]
    async fn live_thread_resume_loads_history_from_explicit_external_rollout_path() {
        let home = TempDir::new().expect("temp dir");
        let external_home = TempDir::new().expect("external temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone())));
        let uuid = uuid::Uuid::from_u128(402);
        let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
        let rollout_path = write_session_file(external_home.path(), "2025-01-03T17-30-00", uuid)
            .expect("external session file");
        let (live_thread, _history) = LiveThread::resume(
            store,
            ThreadHistoryMode::Legacy,
            ResumeThreadParams {
                history_revision: None,
                thread_id,
                rollout_path: Some(rollout_path),
                history: None,
                include_archived: false,
                metadata: ThreadPersistenceMetadata {
                    cwd: Some(home.path().to_path_buf()),
                    model_provider: "different-provider".to_string(),
                    memory_mode: ThreadMemoryMode::Enabled,
                },
            },
            /*resume_preparation*/ None,
        )
        .await
        .expect("resume external live thread");

        live_thread
            .append_items(&[user_message_item("new external append")])
            .await
            .expect("append after external resume");

        let metadata = runtime
            .get_thread(thread_id)
            .await
            .expect("sqlite metadata read")
            .expect("sqlite metadata");
        assert_eq!(
            metadata.created_at.to_rfc3339(),
            "2025-01-03T17:30:00+00:00"
        );
        assert_eq!(metadata.model_provider, "test-provider");
        assert_eq!(
            metadata.first_user_message.as_deref(),
            Some("Hello from user")
        );
    }

    #[tokio::test]
    async fn create_thread_rejects_missing_cwd() {
        let home = TempDir::new().expect("temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let competing_store =
            LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] {
            let thread_id = ThreadId::default();
            let mut params = create_thread_params(thread_id);
            params.history_mode = history_mode;
            params.metadata.cwd = None;

            let err = store
                .create_thread(params)
                .await
                .expect_err("local thread store should require cwd");

            assert!(matches!(
                err,
                ThreadStoreError::InvalidRequest { message }
                    if message == "local thread store requires a cwd"
            ));

            let mut valid_params = create_thread_params(thread_id);
            valid_params.history_mode = history_mode;
            competing_store
                .create_thread(valid_params)
                .await
                .expect("failed initialization should release cross-process writer ownership");
        }
    }

    #[tokio::test]
    async fn discard_thread_drops_unmaterialized_live_writer() {
        let home = TempDir::new().expect("temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] {
            let thread_id = ThreadId::default();
            let mut params = create_thread_params(thread_id);
            params.history_mode = history_mode;

            store
                .create_thread(params)
                .await
                .expect("create live thread");
            let rollout_path = store
                .live_rollout_path(thread_id)
                .await
                .expect("load rollout path");
            {
                let recorder = store
                    .live_recorders
                    .lock()
                    .await
                    .get(&thread_id)
                    .expect("live recorder")
                    .recorder
                    .clone();
                recorder
                    .record_canonical_items(&[user_message_item("deferred item to discard")])
                    .await
                    .expect("queue deferred item without materializing it");
            }
            assert!(!rollout_path.exists());

            let lock_path = home
                .path()
                .join("thread-writer-locks")
                .join(format!("{thread_id}.lock"));
            assert!(lock_path.exists());
            store
                .discard_thread(thread_id)
                .await
                .expect("discard live thread");

            assert!(!rollout_path.exists());
            assert!(!lock_path.exists());
            let err = store
                .append_items(AppendThreadItemsParams {
                    thread_id,
                    items: vec![user_message_item("write after discard")],
                })
                .await
                .expect_err("discard should remove the live thread writer");
            assert!(
                matches!(err, ThreadStoreError::ThreadNotFound { thread_id: missing } if missing == thread_id)
            );
        }
    }

    #[tokio::test]
    async fn attached_creation_and_deletion_coordinate_across_stores() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("initialize state db");
        let first = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
        let second = LocalThreadStore::new(config, Some(runtime.clone()));
        let parent_uuid = uuid::Uuid::new_v4();
        let parent = ThreadId::from_u128(parent_uuid.as_u128());
        let parent_path = write_session_file(home.path(), "2025-01-03T00-00-00", parent_uuid)
            .expect("write parent");
        let child = ThreadId::new();
        let mut params = create_thread_params(child);
        params.attached_thread = true;
        assert!(matches!(
            second.create_thread(params.clone()).await,
            Err(ThreadStoreError::InvalidRequest { .. })
        ));
        params.forked_from_id = Some(parent);
        let parent_guard = first.acquire_writer_lock(parent).expect("lock parent");
        assert!(matches!(
            second.create_thread(params.clone()).await,
            Err(ThreadStoreError::Conflict { .. })
        ));
        drop(parent_guard);
        let pool = runtime
            .sqlite()
            .open_read_write_pool(&runtime.sqlite().state_db_path())
            .await
            .expect("open state db");
        sqlx::query(
            "CREATE TRIGGER fail_attached_index BEFORE UPDATE OF memory_mode ON threads WHEN NEW.attached_thread_parent_id IS NOT NULL BEGIN SELECT RAISE(ABORT, 'index failed'); END",
        )
        .execute(&pool)
        .await
        .expect("inject index failure");
        let mut failing_params = params.clone();
        failing_params.metadata.memory_mode = ThreadMemoryMode::Disabled;
        second
            .create_thread(failing_params)
            .await
            .expect_err("indexing must fail");
        first
            .ensure_no_attached_threads(&[parent])
            .await
            .expect("failed creation must not leave an attached rollout");
        assert!(
            runtime
                .get_thread(child)
                .await
                .expect("read index")
                .is_none()
        );
        sqlx::query("DROP TRIGGER fail_attached_index")
            .execute(&pool)
            .await
            .expect("remove index failure");
        second
            .create_thread(params.clone())
            .await
            .expect("create child");
        assert_eq!(
            runtime
                .delete_thread(child)
                .await
                .expect("remove index row"),
            1
        );
        assert!(matches!(
            first
                .delete_threads(DeleteThreadsParams {
                    thread_ids: vec![parent]
                })
                .await,
            Err(ThreadStoreError::InvalidRequest { .. })
        ));
        assert!(
            parent_path.exists(),
            "refused deletion must preserve parent"
        );
        second
            .discard_thread(child)
            .await
            .expect("release child writer");
        assert!(matches!(
            first
                .delete_threads(DeleteThreadsParams {
                    thread_ids: vec![parent]
                })
                .await,
            Err(ThreadStoreError::InvalidRequest { .. })
        ));
        first
            .delete_thread(DeleteThreadParams { thread_id: child })
            .await
            .expect("delete child");
        first
            .delete_threads(DeleteThreadsParams {
                thread_ids: vec![parent],
            })
            .await
            .expect("parent can be deleted after its child");
        assert!(!parent_path.exists());
        assert!(matches!(
            second.create_thread(params).await,
            Err(ThreadStoreError::ThreadNotFound { thread_id }) if thread_id == parent
        ));
    }

    #[tokio::test]
    async fn resume_thread_reopens_live_writer_and_appends_with_stale_sqlite_path() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let runtime = codex_state::StateRuntime::init(
            config.sqlite.clone(),
            config.default_model_provider_id.clone(),
        )
        .await
        .expect("state db should initialize");
        let thread_id = ThreadId::default();

        let first_store = LocalThreadStore::new(config.clone(), Some(runtime.clone()));
        first_store
            .create_thread(create_thread_params(thread_id))
            .await
            .expect("create initial thread");
        first_store
            .append_items(AppendThreadItemsParams {
                thread_id,
                items: vec![user_message_item("before resume")],
            })
            .await
            .expect("append initial item");
        first_store
            .persist_thread(thread_id, PersistContext::Standard)
            .await
            .expect("persist initial thread");
        first_store
            .flush_thread(thread_id)
            .await
            .expect("flush initial thread");
        let rollout_path = first_store
            .live_rollout_path(thread_id)
            .await
            .expect("load rollout path");
        first_store
            .shutdown_thread(thread_id)
            .await
            .expect("shutdown initial writer");

        let mut builder = codex_state::ThreadMetadataBuilder::new(
            thread_id,
            home.path().join("missing-rollout.jsonl"),
            chrono::Utc::now(),
            SessionSource::Exec,
        );
        builder.history_mode = ThreadHistoryMode::Legacy;
        builder.cwd = home.path().to_path_buf();
        let metadata = builder.build("test-provider");
        runtime
            .upsert_thread(&metadata)
            .await
            .expect("store stale sqlite path");

        let resumed_store = LocalThreadStore::new(config, Some(runtime));
        resumed_store
            .resume_thread(ResumeThreadParams {
                history_revision: None,
                thread_id,
                rollout_path: None,
                history: None,
                include_archived: true,
                metadata: thread_metadata(),
            })
            .await
            .expect("resume live thread");
        resumed_store
            .append_items(AppendThreadItemsParams {
                thread_id,
                items: vec![user_message_item("after resume")],
            })
            .await
            .expect("append resumed item");
        resumed_store
            .flush_thread(thread_id)
            .await
            .expect("flush resumed thread");

        assert_rollout_contains_message(rollout_path.as_path(), "before resume").await;
        assert_rollout_contains_message(rollout_path.as_path(), "after resume").await;
    }

    #[tokio::test]
    async fn live_writers_reject_cross_process_create_and_resume() {
        let home = TempDir::new().expect("temp dir");
        let config = test_config(home.path());
        let primary = LocalThreadStore::new(config.clone(), /*state_db*/ None);
        let secondary = LocalThreadStore::new(config, /*state_db*/ None);

        for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] {
            let thread_id = ThreadId::default();
            let mut create_params = create_thread_params(thread_id);
            create_params.history_mode = history_mode;

            primary
                .create_thread(create_params.clone())
                .await
                .expect("create live thread");
            primary
                .persist_thread(thread_id, PersistContext::Standard)
                .await
                .expect("persist thread for resume");
            let rollout_path = primary
                .live_rollout_path(thread_id)
                .await
                .expect("load rollout path");
            let snapshot = primary
                .load_latest_model_context(LoadThreadHistoryParams {
                    thread_id,
                    include_archived: true,
                })
                .await
                .expect("load initial snapshot");
            let mut expected = snapshot.items;
            let resume_params = ResumeThreadParams {
                history_revision: snapshot.revision,
                thread_id,
                rollout_path: Some(rollout_path),
                history: Some(Arc::new(expected.clone())),
                include_archived: true,
                metadata: thread_metadata(),
            };

            let error = secondary
                .create_thread(create_params)
                .await
                .expect_err("competing create should fail");
            assert!(matches!(error, ThreadStoreError::Conflict { .. }));

            let error = secondary
                .resume_thread(resume_params.clone())
                .await
                .expect_err("competing resume should fail");
            assert!(matches!(error, ThreadStoreError::Conflict { .. }));

            primary
                .shutdown_thread(thread_id)
                .await
                .expect("release writer");
            let unchanged = secondary
                .resume_thread(resume_params.clone())
                .await
                .expect("resume unchanged snapshot");
            assert!(Arc::ptr_eq(
                &unchanged,
                resume_params.history.as_ref().unwrap()
            ));

            let late_edit = RolloutItem::ResponseItem(
                ResponseItem::Message {
                    status: None,
                    encrypted_content: None,
                    id: None,
                    role: "user".to_string(),
                    content: vec![ContentItem::InputText {
                        text: "edit after the initial snapshot".to_string(),
                    }],
                    phase: None,
                    internal_chat_message_metadata_passthrough: None,
                }
                .into(),
            );
            secondary
                .append_items(AppendThreadItemsParams {
                    thread_id,
                    items: vec![late_edit.clone()],
                })
                .await
                .expect("append before handing off writer ownership");
            expected.push(late_edit);
            secondary
                .shutdown_thread(thread_id)
                .await
                .expect("shutdown should release writer ownership");
            let resumed = primary
                .resume_thread(resume_params)
                .await
                .expect("resume after shutdown should acquire writer ownership");
            assert_eq!(serde_json::json!(*resumed), serde_json::json!(expected));
        }
    }

    #[tokio::test]
    async fn create_thread_rejects_duplicate_live_writer() {
        let home = TempDir::new().expect("temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let thread_id = ThreadId::default();

        store
            .create_thread(create_thread_params(thread_id))
            .await
            .expect("create live thread");

        let err = store
            .create_thread(create_thread_params(thread_id))
            .await
            .expect_err("duplicate live writer should fail");

        assert!(matches!(err, ThreadStoreError::InvalidRequest { .. }));
        assert!(err.to_string().contains("already has a live local writer"));
    }

    #[tokio::test]
    async fn resume_thread_rejects_duplicate_live_writer() {
        let home = TempDir::new().expect("temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let thread_id = ThreadId::default();

        store
            .create_thread(create_thread_params(thread_id))
            .await
            .expect("create live thread");
        let rollout_path = store
            .live_rollout_path(thread_id)
            .await
            .expect("live rollout path");
        let err = store
            .resume_thread(ResumeThreadParams {
                history_revision: None,
                thread_id,
                rollout_path: Some(rollout_path),
                history: None,
                include_archived: true,
                metadata: thread_metadata(),
            })
            .await
            .expect_err("duplicate live resume should fail");
        assert!(matches!(err, ThreadStoreError::InvalidRequest { .. }));
        assert!(err.to_string().contains("already has a live local writer"));
    }

    #[tokio::test]
    async fn resume_thread_rejects_missing_cwd() {
        let home = TempDir::new().expect("temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let competing_store =
            LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let uuid = uuid::Uuid::from_u128(408);
        let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
        let rollout_path = write_session_file_with_history_mode(
            home.path(),
            "2025-01-04T11-30-00",
            uuid,
            ThreadHistoryMode::Paginated,
        )
        .expect("session file");
        let err = store
            .resume_thread(ResumeThreadParams {
                history_revision: None,
                thread_id,
                rollout_path: Some(rollout_path.clone()),
                history: None,
                include_archived: true,
                metadata: ThreadPersistenceMetadata {
                    cwd: None,
                    model_provider: "test-provider".to_string(),
                    memory_mode: ThreadMemoryMode::Enabled,
                },
            })
            .await
            .expect_err("missing cwd should fail");

        assert!(matches!(err, ThreadStoreError::InvalidRequest { .. }));
        assert!(err.to_string().contains("requires a cwd"));

        competing_store
            .resume_thread(ResumeThreadParams {
                history_revision: None,
                thread_id,
                rollout_path: Some(rollout_path),
                history: None,
                include_archived: true,
                metadata: thread_metadata(),
            })
            .await
            .expect("failed initialization should release cross-process writer ownership");
    }

    #[tokio::test]
    async fn load_history_uses_live_writer_rollout_path() {
        let home = TempDir::new().expect("temp dir");
        let external_home = TempDir::new().expect("external temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let uuid = uuid::Uuid::from_u128(404);
        let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
        let rollout_path = write_session_file(external_home.path(), "2025-01-04T10-00-00", uuid)
            .expect("external session file");

        store
            .resume_thread(ResumeThreadParams {
                history_revision: None,
                thread_id,
                rollout_path: Some(rollout_path),
                history: None,
                include_archived: true,
                metadata: thread_metadata(),
            })
            .await
            .expect("resume live thread");
        store
            .append_items(AppendThreadItemsParams {
                thread_id,
                items: vec![user_message_item("external history item")],
            })
            .await
            .expect("append live item");
        store
            .flush_thread(thread_id)
            .await
            .expect("flush live thread");

        let history = store
            .load_history(LoadThreadHistoryParams {
                thread_id,
                include_archived: false,
            })
            .await
            .expect("load external live history");

        assert!(history.items.iter().any(|item| {
            matches!(
                item,
                RolloutItem::EventMsg(EventMsg::UserMessage(event)) if event.message == "external history item"
            )
        }));
    }

    #[tokio::test]
    async fn read_thread_uses_live_writer_rollout_path_for_external_resume() {
        let home = TempDir::new().expect("temp dir");
        let external_home = TempDir::new().expect("external temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let uuid = uuid::Uuid::from_u128(406);
        let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
        let rollout_path = write_session_file(external_home.path(), "2025-01-04T11-00-00", uuid)
            .expect("external session file");

        store
            .resume_thread(ResumeThreadParams {
                history_revision: None,
                thread_id,
                rollout_path: Some(rollout_path.clone()),
                history: None,
                include_archived: true,
                metadata: thread_metadata(),
            })
            .await
            .expect("resume live thread");

        let thread = store
            .read_thread(ReadThreadParams {
                thread_id,
                include_archived: false,
                include_history: true,
            })
            .await
            .expect("read external live thread");

        assert_eq!(
            thread.rollout_path,
            Some(std::fs::canonicalize(rollout_path).expect("canonical rollout path"))
        );
        assert!(thread.history.expect("history").items.iter().any(|item| {
            matches!(
                item,
                RolloutItem::EventMsg(EventMsg::UserMessage(event)) if event.message == "Hello from user"
            )
        }));

        let error = store
            .prepare_fork(PrepareForkParams {
                thread_id,
                boundary: crate::ForkBoundary::Latest,
            })
            .await
            .expect_err("external rollouts cannot be referenced by thread id");
        assert!(error.to_string().contains("must be in Codex home"));
    }

    #[tokio::test]
    async fn load_history_uses_live_writer_rollout_path_for_archived_source() {
        let home = TempDir::new().expect("temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let uuid = uuid::Uuid::from_u128(405);
        let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
        let rollout_path = write_archived_session_file(home.path(), "2025-01-04T10-30-00", uuid)
            .expect("archived session file");
        #[cfg(unix)]
        let rollout_path = {
            let alias = home.path().join("active-alias.jsonl");
            std::os::unix::fs::symlink(&rollout_path, &alias).expect("symlink archived rollout");
            alias
        };

        let mut resume_params = ResumeThreadParams {
            history_revision: None,
            thread_id,
            rollout_path: Some(rollout_path),
            history: None,
            include_archived: false,
            metadata: thread_metadata(),
        };
        let err = store
            .resume_thread(resume_params.clone())
            .await
            .expect_err("active-only resume should reject archived rollout targets");
        assert!(matches!(err, ThreadStoreError::InvalidRequest { .. }));
        assert!(err.to_string().contains("archived"));
        resume_params.include_archived = true;
        store
            .resume_thread(resume_params)
            .await
            .expect("resume live archived thread");
        store
            .append_items(AppendThreadItemsParams {
                thread_id,
                items: vec![user_message_item("archived live history item")],
            })
            .await
            .expect("append live item");
        store
            .flush_thread(thread_id)
            .await
            .expect("flush live thread");

        let err = store
            .read_thread(ReadThreadParams {
                thread_id,
                include_archived: false,
                include_history: false,
            })
            .await
            .expect_err("active-only read should reject archived live thread");
        assert!(matches!(err, ThreadStoreError::InvalidRequest { .. }));

        let err = store
            .load_history(LoadThreadHistoryParams {
                thread_id,
                include_archived: false,
            })
            .await
            .expect_err("active-only history should reject archived live thread");
        assert!(matches!(err, ThreadStoreError::InvalidRequest { .. }));
        assert!(err.to_string().contains("archived"));

        let history = store
            .load_history(LoadThreadHistoryParams {
                thread_id,
                include_archived: true,
            })
            .await
            .expect("load archived live history");

        assert!(history.items.iter().any(|item| {
            matches!(
                item,
                RolloutItem::EventMsg(EventMsg::UserMessage(event)) if event.message == "archived live history item"
            )
        }));
    }

    #[tokio::test]
    async fn read_thread_by_rollout_path_includes_history() {
        let home = TempDir::new().expect("temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let thread_id = ThreadId::default();

        store
            .create_thread(create_thread_params(thread_id))
            .await
            .expect("create thread");
        store
            .append_items(AppendThreadItemsParams {
                thread_id,
                items: vec![user_message_item("path read")],
            })
            .await
            .expect("append item");
        store.flush_thread(thread_id).await.expect("flush thread");
        let rollout_path = store
            .live_rollout_path(thread_id)
            .await
            .expect("load rollout path");

        let thread = store
            .read_thread_by_rollout_path(
                rollout_path,
                /*include_archived*/ true,
                /*include_history*/ true,
            )
            .await
            .expect("read thread by rollout path");

        assert_eq!(thread.thread_id, thread_id);
        assert_eq!(thread.history_mode, ThreadHistoryMode::Legacy);
        assert_eq!(
            thread
                .history
                .as_ref()
                .expect("history")
                .items
                .iter()
                .filter(|item| matches!(item, RolloutItem::EventMsg(EventMsg::UserMessage(_))))
                .count(),
            1
        );
    }

    #[tokio::test]
    async fn paginated_threads_allow_metadata_reads_and_resume_but_reject_legacy_history_paths() {
        let home = TempDir::new().expect("temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let uuid = uuid::Uuid::from_u128(408);
        let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
        let rollout_path = write_session_file_with_history_mode(
            home.path(),
            "2025-01-04T12-00-00",
            uuid,
            ThreadHistoryMode::Paginated,
        )
        .expect("session file");

        let thread = store
            .read_thread(ReadThreadParams {
                thread_id,
                include_archived: false,
                include_history: false,
            })
            .await
            .expect("metadata read");
        assert_eq!(thread.history_mode, ThreadHistoryMode::Paginated);
        assert!(thread.history.is_none());

        let thread = store
            .read_thread_by_rollout_path(
                rollout_path.clone(),
                /*include_archived*/ true,
                /*include_history*/ false,
            )
            .await
            .expect("metadata path read");
        assert_eq!(thread.history_mode, ThreadHistoryMode::Paginated);
        assert!(thread.history.is_none());

        assert_paginated_threads_unsupported(
            store
                .read_thread(ReadThreadParams {
                    thread_id,
                    include_archived: false,
                    include_history: true,
                })
                .await
                .expect_err("full history read should fail"),
        );
        assert_paginated_threads_unsupported(
            store
                .read_thread_by_rollout_path(
                    rollout_path.clone(),
                    /*include_archived*/ true,
                    /*include_history*/ true,
                )
                .await
                .expect_err("full history path read should fail"),
        );
        assert_paginated_threads_unsupported(
            store
                .load_history(LoadThreadHistoryParams {
                    thread_id,
                    include_archived: false,
                })
                .await
                .expect_err("history load should fail"),
        );
        store
            .resume_thread(ResumeThreadParams {
                history_revision: None,
                thread_id,
                rollout_path: Some(rollout_path),
                history: None,
                include_archived: false,
                metadata: thread_metadata(),
            })
            .await
            .expect("resume should succeed");
    }

    #[tokio::test]
    async fn paginated_live_appends_use_paginated_history_mode() {
        let home = TempDir::new().expect("temp dir");
        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
        let thread_id = ThreadId::default();
        let mut create_params = create_thread_params(thread_id);
        create_params.history_mode = ThreadHistoryMode::Paginated;
        store
            .create_thread(create_params)
            .await
            .expect("create paginated thread");
        let paginated_item = RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
            content_token_count: None,
            thread_id,
            turn_id: "turn-1".to_string(),
            item: TurnItem::UserMessage(UserMessageItem {
                id: "item-1".to_string(),
                client_id: None,
                content: Vec::new(),
            }),
            started_at_ms: Some(0),
            completed_at_ms: 1,
        }));
        store
            .append_items(AppendThreadItemsParams {
                thread_id,
                items: vec![
                    user_message_item("legacy event should not persist"),
                    paginated_item,
                ],
            })
            .await
            .expect("append paginated item");
        let rollout_path = store
            .live_rollout_path(thread_id)
            .await
            .expect("paginated rollout path");
        let (items, _, _) = RolloutRecorder::load_rollout_items(rollout_path.as_path())
            .await
            .expect("load paginated rollout");
        assert!(items.iter().any(|item| {
            matches!(
                item,
                RolloutItem::EventMsg(EventMsg::ItemCompleted(event))
                    if event.turn_id == "turn-1"
            )
        }));
        assert!(!items.iter().any(|item| {
            matches!(
                item,
                RolloutItem::EventMsg(EventMsg::UserMessage(event))
                    if event.message == "legacy event should not persist"
            )
        }));
    }

    fn create_thread_params(thread_id: ThreadId) -> CreateThreadParams {
        CreateThreadParams {
            attached_thread: false,
            creator_user_id: None,
            creator_account_id: None,
            session_id: thread_id.into(),
            thread_id,
            extra_config: None,
            forked_from_id: None,
            parent_thread_id: None,
            source: SessionSource::Exec,
            thread_source: None,
            originator: "test_originator".to_string(),
            base_instructions: BaseInstructions::default(),
            dynamic_tools: Vec::new(),
            selected_capability_roots: Vec::new(),
            multi_agent_version: None,
            history_mode: ThreadHistoryMode::Legacy,
            history_base: None,
            subagent_history_start_ordinal: None,
            initial_window_id: codex_protocol::ContextWindowId::new_uuid_v7().to_string(),
            runtime_workspace_roots: None,
            metadata: thread_metadata(),
        }
    }

    fn assert_paginated_threads_unsupported(err: ThreadStoreError) {
        assert!(matches!(
            err,
            ThreadStoreError::Unsupported {
                operation: "paginated_threads"
            }
        ));
    }

    fn thread_metadata() -> ThreadPersistenceMetadata {
        ThreadPersistenceMetadata {
            cwd: Some(std::env::current_dir().expect("cwd")),
            model_provider: "test-provider".to_string(),
            memory_mode: ThreadMemoryMode::Enabled,
        }
    }

    fn user_message_item(message: &str) -> RolloutItem {
        RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
            client_id: None,
            message: message.to_string(),
            images: None,
            local_images: Vec::new(),
            text_elements: Vec::new(),
            ..Default::default()
        }))
    }

    async fn assert_rollout_contains_message(path: &std::path::Path, expected: &str) {
        let (items, _, _) = RolloutRecorder::load_rollout_items(path)
            .await
            .expect("load rollout items");
        assert!(items.iter().any(|item| {
            matches!(
                item,
                RolloutItem::EventMsg(EventMsg::UserMessage(event)) if event.message == expected
            )
        }));
    }
}