Add legacy rollout migration to paginated history (#37175)

## What changed

- Add `LocalThreadStore::migrate_rollouts` with dry-run and apply modes,
  optional thread selection, throughput limiting, and per-rollout outcomes.
- Canonicalize legacy JSONL records into paginated history while preserving
  model-visible conversation context, including compressed rollouts and copied
  fork history.
- Publish replacements atomically and journal migrations so interrupted SQLite
  projections can be recovered. Coordinate migration with rollout compression
  and skip rollouts with active writers.
- Make paginated history a one-way SQLite promotion so stale metadata cannot
  downgrade a migrated thread.

## Testing

- Cover legacy record normalization, dry runs, idempotency, malformed input,
  compressed rollouts, active writers, interrupted migration recovery, and a
  cold app-server resume after migration.

GitOrigin-RevId: b9991b659f28ebb52da39ce62e8b4e0bae2ba2bb

codex-rs/Cargo.lock

@@ -4208,6 +4208,7 @@ dependencies = [
  "codex-state",
  "codex-utils-absolute-path",
  "codex-utils-path",
+ "codex-utils-path-uri",
  "futures",
  "pretty_assertions",
  "pulldown-cmark",

codex-rs/app-server/tests/suite/v2/mod.rs

@@ -68,6 +68,7 @@ mod request_permissions;
 mod request_user_input;
 mod request_validation;
 mod review;
+mod rollout_migration;
 mod safety_check_downgrade;
 #[cfg(not(target_os = "windows"))]
 mod selected_capability_stack;

codex-rs/app-server/tests/suite/v2/rollout_migration.rs

@@ -0,0 +1,132 @@
+use anyhow::Result;
+use app_test_support::MockResponsesConfig;
+use app_test_support::TestAppServer;
+use codex_app_server_protocol::ThreadHistoryMode;
+use codex_app_server_protocol::ThreadResumeParams;
+use codex_app_server_protocol::ThreadResumeResponse;
+use codex_app_server_protocol::ThreadStartParams;
+use codex_app_server_protocol::ThreadStartResponse;
+use codex_app_server_protocol::TurnStartParams;
+use codex_app_server_protocol::UserInput;
+use codex_thread_store::LocalThreadStore;
+use codex_thread_store::LocalThreadStoreConfig;
+use codex_thread_store::RolloutMigrationMode;
+use codex_thread_store::RolloutMigrationOptions;
+use codex_thread_store::RolloutMigrationStatus;
+use codex_utils_absolute_path::test_support::PathExt;
+use core_test_support::responses;
+use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+use tokio::time::timeout;
+
+const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
+
+#[tokio::test]
+async fn migrated_legacy_thread_cold_resume_preserves_model_context() -> Result<()> {
+    let server = responses::start_mock_server().await;
+    let response_mock = responses::mount_sse_sequence(
+        &server,
+        vec![
+            responses::sse(vec![
+                responses::ev_response_created("resp-1"),
+                responses::ev_assistant_message("msg-1", "legacy assistant message"),
+                responses::ev_completed("resp-1"),
+            ]),
+            responses::sse(vec![
+                responses::ev_response_created("resp-2"),
+                responses::ev_assistant_message("msg-2", "resumed assistant message"),
+                responses::ev_completed("resp-2"),
+            ]),
+        ],
+    )
+    .await;
+    let codex_home = TempDir::new()?;
+    MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
+
+    let mut primary = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .build_initialized()
+        .await?;
+    let start_id = primary
+        .send_thread_start_request_with_auto_env(ThreadStartParams {
+            history_mode: Some(ThreadHistoryMode::Legacy),
+            ..Default::default()
+        })
+        .await?;
+    let ThreadStartResponse { thread, .. } =
+        timeout(DEFAULT_READ_TIMEOUT, primary.read_response(start_id)).await??;
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        primary.start_turn_and_wait_for_completion(TurnStartParams {
+            thread_id: thread.id.clone(),
+            input: vec![UserInput::Text {
+                text: "legacy user message".to_string(),
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        }),
+    )
+    .await??;
+    timeout(DEFAULT_READ_TIMEOUT, primary.shutdown_gracefully()).await??;
+
+    let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs());
+    let state_db =
+        codex_state::StateRuntime::init(sqlite.clone(), "mock_provider".to_string()).await?;
+    let store = LocalThreadStore::new(
+        LocalThreadStoreConfig {
+            codex_home: codex_home.path().to_path_buf(),
+            sqlite,
+            default_model_provider_id: "mock_provider".to_string(),
+        },
+        Some(state_db),
+    );
+    let report = store
+        .migrate_rollouts(RolloutMigrationOptions {
+            mode: RolloutMigrationMode::Apply,
+            max_mib_per_second: 1024,
+            ..RolloutMigrationOptions::default()
+        })
+        .await?;
+    assert_eq!(report.outcomes.len(), 1);
+    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
+    drop(store);
+
+    let mut secondary = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .build_initialized()
+        .await?;
+    let resume_id = secondary
+        .send_thread_resume_request(ThreadResumeParams {
+            thread_id: thread.id.clone(),
+            exclude_turns: true,
+            ..Default::default()
+        })
+        .await?;
+    let ThreadResumeResponse {
+        thread: resumed, ..
+    } = timeout(DEFAULT_READ_TIMEOUT, secondary.read_response(resume_id)).await??;
+    assert_eq!(resumed.history_mode, ThreadHistoryMode::Paginated);
+
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        secondary.start_turn_and_wait_for_completion(TurnStartParams {
+            thread_id: thread.id,
+            input: vec![UserInput::Text {
+                text: "resumed user message".to_string(),
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        }),
+    )
+    .await??;
+
+    let requests = response_mock.requests();
+    assert_eq!(requests.len(), 2);
+    let resumed_request = requests.last().expect("resumed turn request");
+    let user_messages = resumed_request.message_input_texts("user");
+    assert!(user_messages.contains(&"legacy user message".to_string()));
+    assert!(user_messages.contains(&"resumed user message".to_string()));
+    assert!(resumed_request.body_contains_text("legacy assistant message"));
+
+    Ok(())
+}

codex-rs/rollout/src/compression.rs

@@ -344,6 +344,16 @@ mod worker {
     }
 
     pub(super) async fn run(codex_home: PathBuf) -> io::Result<()> {
+        let Some(_maintenance_guard) =
+            crate::try_acquire_rollout_maintenance_lock(codex_home.as_path())?
+        else {
+            metrics::run("skipped_maintenance");
+            debug!(
+                "rollout maintenance is already running for {}",
+                codex_home.display()
+            );
+            return Ok(());
+        };
         let marker = match CompressionRunMarker::try_claim(codex_home.as_path()) {
             Ok(Some(marker)) => marker,
             Ok(None) => {

codex-rs/rollout/src/compression_tests.rs

@@ -308,6 +308,28 @@ async fn worker_compresses_old_active_and_archived_rollouts() -> anyhow::Result<
     Ok(())
 }
 
+#[tokio::test]
+async fn worker_waits_for_rollout_maintenance_before_compressing() -> anyhow::Result<()> {
+    let home = TempDir::new()?;
+    let uuid = Uuid::from_u128(26);
+    let thread_id = ThreadId::from_string(&uuid.to_string())?;
+    let path = rollout_path(home.path(), "2025-01-03T12-00-00", uuid);
+    write_rollout(&path, thread_id, "migration in progress")?;
+    set_old_mtime(&path)?;
+    let guard = crate::try_acquire_rollout_maintenance_lock(home.path())?
+        .expect("claim rollout maintenance lock");
+
+    worker::run(home.path().to_path_buf()).await?;
+    assert!(path.exists());
+    assert!(!compressed_rollout_path(&path).exists());
+
+    drop(guard);
+    worker::run(home.path().to_path_buf()).await?;
+    assert!(!path.exists());
+    assert!(compressed_rollout_path(&path).exists());
+    Ok(())
+}
+
 #[tokio::test]
 async fn worker_skips_archived_paginated_fork_pointer_chain() -> anyhow::Result<()> {
     let home = TempDir::new()?;

codex-rs/rollout/src/lib.rs

@@ -7,6 +7,7 @@ use codex_protocol::protocol::SessionSource;
 pub(crate) mod compression;
 pub(crate) mod config;
 pub(crate) mod list;
+mod maintenance;
 pub(crate) mod metadata;
 mod model_context;
 mod ordinal;
@@ -67,6 +68,8 @@ pub use list::read_head_for_summary;
 pub use list::read_session_meta_line;
 pub use list::read_thread_item_from_rollout;
 pub use list::rollout_date_parts;
+pub use maintenance::RolloutMaintenanceGuard;
+pub use maintenance::try_acquire_rollout_maintenance_lock;
 pub use metadata::builder_from_items;
 pub use model_context::ModelContextScan;
 pub use model_context::ModelContextScanProgress;

codex-rs/rollout/src/maintenance.rs

@@ -0,0 +1,41 @@
+//! Coordinates maintenance jobs that replace local rollout files.
+//!
+//! Rollout compression and legacy rollout migration both publish by renaming a replacement over an
+//! existing rollout path. They must not do that at the same time for one Codex home, so they share
+//! this process-scoped, nonblocking file lock.
+//!
+//! This is separate from per-thread writer locks, which protect live rollout appenders. It is also
+//! separate from compression's durable run marker, which throttles how often compression scans.
+
+use std::fs;
+use std::fs::File;
+use std::fs::OpenOptions;
+use std::io;
+use std::path::Path;
+
+const ROLLOUT_MAINTENANCE_LOCK: &str = "rollout-maintenance.lock";
+
+/// Holds exclusive ownership of operations that replace local rollout files.
+pub struct RolloutMaintenanceGuard {
+    _file: File,
+}
+
+/// Try to exclude rollout compression and migration for one Codex home.
+pub fn try_acquire_rollout_maintenance_lock(
+    codex_home: &Path,
+) -> io::Result<Option<RolloutMaintenanceGuard>> {
+    let directory = codex_home.join(".tmp");
+    fs::create_dir_all(&directory)?;
+    let file = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(directory.join(ROLLOUT_MAINTENANCE_LOCK))?;
+
+    match file.try_lock() {
+        Ok(()) => Ok(Some(RolloutMaintenanceGuard { _file: file })),
+        Err(std::fs::TryLockError::WouldBlock) => Ok(None),
+        Err(std::fs::TryLockError::Error(error)) => Err(error),
+    }
+}

codex-rs/state/src/runtime/threads.rs

@@ -55,6 +55,14 @@ WHERE threads.id = ?
             .transpose()
     }
 
+    /// Permanently promote a thread to paginated history without changing metadata or recency.
+    pub async fn mark_thread_paginated(&self, thread_id: ThreadId) -> anyhow::Result<bool> {
+        let result = sqlx::query("UPDATE threads SET history_mode = 'paginated' WHERE id = ?")
+            .bind(thread_id.to_string())
+            .execute(self.pool.as_ref())
+            .await?;
+        Ok(result.rows_affected() > 0)
+    }
     pub async fn get_thread_memory_mode(&self, id: ThreadId) -> anyhow::Result<Option<String>> {
         let row = sqlx::query("SELECT memory_mode FROM threads WHERE id = ?")
             .bind(id.to_string())
@@ -873,7 +881,11 @@ ON CONFLICT(id) DO UPDATE SET
     updated_at_ms = excluded.updated_at_ms,
     recency_at_ms = threads.recency_at_ms,
     source = excluded.source,
-    history_mode = excluded.history_mode,
+    -- Paginated history is a one-way promotion; stale legacy metadata must not downgrade it.
+    history_mode = CASE
+        WHEN threads.history_mode = 'paginated' THEN threads.history_mode
+        ELSE excluded.history_mode
+    END,
     thread_source = excluded.thread_source,
     agent_nickname = excluded.agent_nickname,
     agent_role = excluded.agent_role,
@@ -1512,7 +1524,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn thread_metadata_round_trips_history_mode() {
+    async fn thread_metadata_history_mode_does_not_downgrade() {
         let codex_home = unique_temp_dir();
         let runtime = StateRuntime::init(
             crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()),
@@ -1522,20 +1534,42 @@ mod tests {
         .expect("state db should initialize");
         let thread_id =
             ThreadId::from_string("00000000-0000-0000-0000-000000000124").expect("valid thread id");
-        let mut metadata = test_thread_metadata(&codex_home, thread_id, codex_home.clone());
-        metadata.history_mode = ThreadHistoryMode::Paginated;
+        let metadata = test_thread_metadata(&codex_home, thread_id, codex_home.clone());
 
         runtime
             .upsert_thread(&metadata)
             .await
             .expect("upsert should succeed");
 
+        assert!(
+            runtime
+                .mark_thread_paginated(thread_id)
+                .await
+                .expect("mark paginated history")
+        );
+
         let metadata = runtime
             .get_thread(thread_id)
             .await
             .expect("thread should load")
             .expect("thread should exist");
         assert_eq!(metadata.history_mode, ThreadHistoryMode::Paginated);
+
+        let mut stale_metadata = metadata;
+        stale_metadata.history_mode = ThreadHistoryMode::Legacy;
+        runtime
+            .upsert_thread(&stale_metadata)
+            .await
+            .expect("upsert stale legacy metadata");
+        assert_eq!(
+            runtime
+                .get_thread(thread_id)
+                .await
+                .expect("read migrated thread")
+                .expect("thread should exist")
+                .history_mode,
+            ThreadHistoryMode::Paginated
+        );
     }
 
     #[tokio::test]

codex-rs/thread-store/Cargo.toml

@@ -23,6 +23,7 @@ codex-rollout = { workspace = true }
 codex-state = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 codex-utils-path = { workspace = true }
+codex-utils-path-uri = { workspace = true }
 futures = { workspace = true }
 pulldown-cmark = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
@@ -31,6 +32,7 @@ sqlx = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+zstd = { workspace = true }
 
 [dev-dependencies]
 codex-utils-absolute-path = { workspace = true }
@@ -38,4 +40,3 @@ pretty_assertions = { workspace = true }
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
 uuid = { workspace = true }
-zstd = { workspace = true }

codex-rs/thread-store/src/lib.rs

@@ -24,6 +24,11 @@ pub use live_thread::LiveThread;
 pub use live_thread::LiveThreadInitGuard;
 pub use local::LocalThreadStore;
 pub use local::LocalThreadStoreConfig;
+pub use local::RolloutMigrationMode;
+pub use local::RolloutMigrationOptions;
+pub use local::RolloutMigrationOutcome;
+pub use local::RolloutMigrationReport;
+pub use local::RolloutMigrationStatus;
 pub use queue_store::LocalQueueStore;
 pub use queue_store::QueueStore;
 pub use store::ThreadStore;

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

@@ -8,6 +8,7 @@ mod model_context;
 mod move_thread_to_section;
 mod paginated_fork;
 mod read_thread;
+mod rollout_migration;
 // This lands before the reader PRs that consume the shared lineage resolver.
 #[allow(dead_code)]
 mod rollout_lineage;
@@ -78,6 +79,12 @@ use crate::UpdateThreadMetadataParams;
 use crate::local::writer_lock::WriterLockCoordinator;
 use crate::local::writer_lock::WriterLockGuard;
 
+pub use rollout_migration::RolloutMigrationMode;
+pub use rollout_migration::RolloutMigrationOptions;
+pub use rollout_migration::RolloutMigrationOutcome;
+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

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

@@ -0,0 +1,812 @@
+//! Orchestrates legacy rollout migration into paginated history.
+//!
+//! This is the high-level migration state machine: find rollout files, decide whether each one is
+//! eligible, take the maintenance and writer locks, canonicalize into a staged JSONL file,
+//! project that staged file into SQLite, verify the projection, then atomically publish it.
+//!
+//! The important invariant is that we always leave behind either the original legacy rollout or a
+//! recoverable paginated rollout. Once the rollout path is replaced, the durable `.pending`
+//! journal must be enough for a later migration run to finish SQLite recovery safely.
+
+use std::path::Path;
+use std::path::PathBuf;
+use std::time::Duration;
+
+use chrono::DateTime;
+use codex_app_server_protocol::project_rollout_line;
+use codex_protocol::ThreadId;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::RolloutLine;
+use codex_protocol::protocol::ThreadHistoryMode;
+use codex_protocol::protocol::ThreadSource;
+use serde::Serialize;
+use tokio::fs::File;
+use tokio::io::AsyncBufReadExt;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufReader;
+use tokio::io::BufWriter;
+use tokio::time::Instant;
+
+use super::LocalThreadStore;
+use super::thread_history;
+use super::thread_history::ProjectedRolloutLine;
+use super::thread_history::RolloutProjectionStep;
+use crate::ThreadStoreError;
+use crate::ThreadStoreResult;
+
+mod canonicalizer;
+mod legacy_event;
+mod line_parser;
+mod publish;
+
+use canonicalizer::LegacyRolloutCanonicalizer;
+use publish::compress_rollout_to_path;
+use publish::compressed_staged_rollout_path;
+use publish::decompress_rollout_to_path;
+use publish::decompressed_staged_rollout_path;
+use publish::migration_journal_path;
+use publish::pending_migration_thread_ids;
+use publish::remove_file_if_present;
+use publish::staged_rollout_path;
+use publish::sync_parent_directory;
+use publish::write_migration_journal;
+
+const PROJECTION_BATCH_BYTES: u64 = 256 * 1024;
+const MAX_ROLLOUT_LINE_BYTES: usize = 16 * 1024 * 1024;
+
+struct CanonicalizationSource<'a> {
+    thread_id: ThreadId,
+    source_path: &'a Path,
+    staged_path: &'a Path,
+    source_permissions: &'a std::fs::Permissions,
+    canonical_session_meta: &'a RolloutLine,
+}
+
+/// Controls whether eligible rollouts are reported or migrated.
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum RolloutMigrationMode {
+    /// Report eligible files without modifying local storage.
+    #[default]
+    DryRun,
+    /// Publish paginated rollouts and materialize their SQLite history.
+    Apply,
+}
+
+/// Selection and throughput limits for a local rollout migration.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct RolloutMigrationOptions {
+    pub mode: RolloutMigrationMode,
+    pub thread_ids: Vec<ThreadId>,
+    pub max_mib_per_second: u64,
+}
+
+impl Default for RolloutMigrationOptions {
+    fn default() -> Self {
+        Self {
+            mode: RolloutMigrationMode::DryRun,
+            thread_ids: Vec::new(),
+            max_mib_per_second: 8,
+        }
+    }
+}
+
+/// The observable result of inspecting one local rollout.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum RolloutMigrationStatus {
+    Eligible,
+    Migrated,
+    AlreadyPaginated,
+    SkippedSubagent,
+    SkippedBusy,
+    Failed,
+}
+
+/// The per-thread result of a rollout migration run.
+#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
+pub struct RolloutMigrationOutcome {
+    pub thread_id: Option<ThreadId>,
+    pub rollout_path: PathBuf,
+    pub status: RolloutMigrationStatus,
+    pub bytes_processed: u64,
+    pub message: Option<String>,
+}
+
+/// The complete result of scanning active rollout files.
+#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)]
+pub struct RolloutMigrationReport {
+    pub outcomes: Vec<RolloutMigrationOutcome>,
+}
+
+struct RolloutMigrationRateLimiter {
+    started_at: Instant,
+    bytes_processed: u64,
+    bytes_per_second: u64,
+    bytes_since_yield: u64,
+}
+
+struct RolloutRecord {
+    line: Option<RolloutLine>,
+    byte_count: u64,
+}
+
+impl RolloutMigrationRateLimiter {
+    fn new(max_mib_per_second: u64) -> ThreadStoreResult<Self> {
+        let bytes_per_second = max_mib_per_second
+            .checked_mul(1024 * 1024)
+            .filter(|rate| *rate > 0)
+            .ok_or_else(|| ThreadStoreError::InvalidRequest {
+                message: "--max-mib-per-second must be a positive supported integer".to_string(),
+            })?;
+        Ok(Self {
+            started_at: Instant::now(),
+            bytes_processed: 0,
+            bytes_per_second,
+            bytes_since_yield: 0,
+        })
+    }
+
+    async fn account(&mut self, bytes: u64) {
+        self.bytes_processed = self.bytes_processed.saturating_add(bytes);
+        self.bytes_since_yield = self.bytes_since_yield.saturating_add(bytes);
+        if self.bytes_since_yield < PROJECTION_BATCH_BYTES {
+            return;
+        }
+        self.bytes_since_yield = 0;
+        let expected =
+            Duration::from_secs_f64(self.bytes_processed as f64 / self.bytes_per_second as f64);
+        if let Some(remaining) = expected.checked_sub(self.started_at.elapsed()) {
+            tokio::time::sleep(remaining).await;
+        } else {
+            tokio::task::yield_now().await;
+        }
+    }
+}
+
+impl LocalThreadStore {
+    /// Inspect or migrate eligible legacy rollout files beneath active sessions.
+    pub async fn migrate_rollouts(
+        &self,
+        options: RolloutMigrationOptions,
+    ) -> ThreadStoreResult<RolloutMigrationReport> {
+        let mut limiter = RolloutMigrationRateLimiter::new(options.max_mib_per_second)?;
+        let _maintenance_guard = match options.mode {
+            RolloutMigrationMode::DryRun => None,
+            RolloutMigrationMode::Apply => Some(
+                codex_rollout::try_acquire_rollout_maintenance_lock(&self.config.codex_home)
+                    .map_err(migration_error)?
+                    .ok_or_else(|| ThreadStoreError::Conflict {
+                        message: "rollout compression or another migration is already running"
+                            .to_string(),
+                    })?,
+            ),
+        };
+        let mut paths =
+            find_rollout_paths(&self.config.codex_home.join(codex_rollout::SESSIONS_SUBDIR))
+                .await?;
+        if options.mode == RolloutMigrationMode::Apply {
+            let pending_thread_ids = pending_migration_thread_ids(&self.config.codex_home).await?;
+            paths.sort_by_key(|path| {
+                !thread_id_from_rollout_filename(path)
+                    .is_some_and(|thread_id| pending_thread_ids.contains(&thread_id))
+            });
+        }
+        let mut report = RolloutMigrationReport::default();
+
+        for path in paths {
+            let metadata = match codex_rollout::read_session_meta_line(&path).await {
+                Ok(metadata) => metadata,
+                Err(error) => {
+                    let thread_id = thread_id_from_rollout_filename(&path);
+                    if matches_selection(&options.thread_ids, thread_id) {
+                        report.outcomes.push(RolloutMigrationOutcome {
+                            thread_id,
+                            rollout_path: path,
+                            status: RolloutMigrationStatus::Failed,
+                            bytes_processed: 0,
+                            message: Some(error.to_string()),
+                        });
+                    }
+                    continue;
+                }
+            };
+            let thread_id = metadata.meta.id;
+            if !matches_selection(&options.thread_ids, Some(thread_id)) {
+                continue;
+            }
+            if metadata.meta.source.is_non_root_agent()
+                || matches!(
+                    metadata.meta.thread_source,
+                    Some(ThreadSource::Subagent | ThreadSource::MemoryConsolidation)
+                )
+            {
+                report.outcomes.push(RolloutMigrationOutcome {
+                    thread_id: Some(thread_id),
+                    rollout_path: path,
+                    status: RolloutMigrationStatus::SkippedSubagent,
+                    bytes_processed: 0,
+                    message: None,
+                });
+                continue;
+            }
+
+            let journal_path = migration_journal_path(&self.config.codex_home, thread_id);
+            if metadata.meta.history_mode == ThreadHistoryMode::Paginated {
+                let bytes_before = limiter.bytes_processed;
+                let result = if options.mode == RolloutMigrationMode::Apply
+                    && tokio::fs::try_exists(&journal_path)
+                        .await
+                        .map_err(migration_error)?
+                {
+                    match self
+                        .recover_published_migration(thread_id, &path, &journal_path, &mut limiter)
+                        .await
+                    {
+                        Ok(()) => Ok(RolloutMigrationStatus::Migrated),
+                        Err(ThreadStoreError::Conflict { message }) => {
+                            let bytes_processed =
+                                limiter.bytes_processed.saturating_sub(bytes_before);
+                            report.outcomes.push(skipped_busy_outcome(
+                                thread_id,
+                                path,
+                                message,
+                                bytes_processed,
+                            ));
+                            continue;
+                        }
+                        Err(error) => Err(error),
+                    }
+                } else {
+                    Ok(RolloutMigrationStatus::AlreadyPaginated)
+                };
+                let bytes_processed = limiter.bytes_processed.saturating_sub(bytes_before);
+                report
+                    .outcomes
+                    .push(migration_outcome(thread_id, path, result, bytes_processed));
+                continue;
+            }
+
+            if options.mode == RolloutMigrationMode::DryRun {
+                report.outcomes.push(RolloutMigrationOutcome {
+                    thread_id: Some(thread_id),
+                    rollout_path: path,
+                    status: RolloutMigrationStatus::Eligible,
+                    bytes_processed: 0,
+                    message: None,
+                });
+                continue;
+            }
+
+            let writer_guard = match self.writer_lock_coordinator.acquire(thread_id) {
+                Ok(guard) => guard,
+                Err(ThreadStoreError::Conflict { message }) => {
+                    report.outcomes.push(skipped_busy_outcome(
+                        thread_id, path, message, /*bytes_processed*/ 0,
+                    ));
+                    continue;
+                }
+                Err(error) => {
+                    report.outcomes.push(migration_outcome(
+                        thread_id,
+                        path,
+                        Err(error),
+                        /*bytes_processed*/ 0,
+                    ));
+                    continue;
+                }
+            };
+            let bytes_before = limiter.bytes_processed;
+            let result = match self
+                .migrate_one_rollout(thread_id, &path, &journal_path, &mut limiter)
+                .await
+            {
+                Ok(()) => Ok(RolloutMigrationStatus::Migrated),
+                Err(error) => {
+                    if let Err(cleanup_error) = self
+                        .cleanup_failed_unpublished_migration(thread_id, &path, &journal_path)
+                        .await
+                    {
+                        Err(migration_error(format!(
+                            "{error}; failed to clean up unpublished migration: {cleanup_error}"
+                        )))
+                    } else {
+                        Err(error)
+                    }
+                }
+            };
+            drop(writer_guard);
+            let bytes_processed = limiter.bytes_processed.saturating_sub(bytes_before);
+            let outcome = match result {
+                Err(ThreadStoreError::Conflict { message }) => {
+                    skipped_busy_outcome(thread_id, path, message, bytes_processed)
+                }
+                result => migration_outcome(thread_id, path, result, bytes_processed),
+            };
+            report.outcomes.push(outcome);
+        }
+
+        Ok(report)
+    }
+
+    async fn migrate_one_rollout(
+        &self,
+        thread_id: ThreadId,
+        rollout_path: &Path,
+        journal_path: &Path,
+        limiter: &mut RolloutMigrationRateLimiter,
+    ) -> ThreadStoreResult<()> {
+        if let Some(state_db) = &self.state_db
+            && state_db
+                .get_thread(thread_id)
+                .await
+                .map_err(migration_error)?
+                .is_none()
+        {
+            return Err(migration_error(format!(
+                "thread {thread_id} is missing its SQLite metadata"
+            )));
+        }
+
+        let compressed = rollout_path_is_compressed(rollout_path);
+        let staged_path = staged_rollout_path(rollout_path)?;
+        let decompressed_path = compressed
+            .then(|| decompressed_staged_rollout_path(rollout_path))
+            .transpose()?;
+        thread_history::delete_thread(self, thread_id).await?;
+        write_migration_journal(journal_path).await?;
+
+        let source_metadata = tokio::fs::metadata(rollout_path)
+            .await
+            .map_err(migration_error)?;
+        let source_modified = source_metadata.modified().ok();
+        let source_permissions = source_metadata.permissions();
+        let source_path = if let Some(decompressed_path) = decompressed_path.as_ref() {
+            decompress_rollout_to_path(rollout_path, decompressed_path).await?;
+            let decompressed_bytes = tokio::fs::metadata(decompressed_path)
+                .await
+                .map_err(migration_error)?
+                .len();
+            limiter
+                .account(source_metadata.len().saturating_add(decompressed_bytes))
+                .await;
+            decompressed_path.as_path()
+        } else {
+            rollout_path
+        };
+        let source_file = File::open(source_path).await.map_err(migration_error)?;
+        let mut source = BufReader::with_capacity(PROJECTION_BATCH_BYTES as usize, source_file);
+        let mut bytes = Vec::new();
+
+        // Paginated rollouts always keep their canonical SessionMeta at ordinal zero. Legacy
+        // readers tolerate a pre-header prefix, so find that metadata before replaying the source
+        // instead of buffering the prefix in memory.
+        let canonical_session_meta = loop {
+            let record = read_rollout_record(&mut source, &mut bytes)
+                .await?
+                .ok_or_else(|| migration_error("rollout contains no session metadata"))?;
+            limiter.account(record.byte_count).await;
+            let Some(line) = record.line else {
+                continue;
+            };
+            if matches!(&line.item, RolloutItem::SessionMeta(_)) {
+                break line;
+            }
+        };
+        drop(source);
+
+        let canonicalization_source = CanonicalizationSource {
+            thread_id,
+            source_path,
+            staged_path: &staged_path,
+            source_permissions: &source_permissions,
+            canonical_session_meta: &canonical_session_meta,
+        };
+        let (expected_length, expected_ordinal) =
+            Self::write_canonical_rollout(&canonicalization_source, limiter).await?;
+
+        // SQLite projection only starts after every staged-file mutation is durable.
+        let modified_at = source_modified;
+        let path = staged_path.clone();
+        tokio::task::spawn_blocking(move || {
+            let file = std::fs::OpenOptions::new().write(true).open(path)?;
+            if let Some(modified_at) = modified_at {
+                file.set_times(std::fs::FileTimes::new().set_modified(modified_at))?;
+            }
+            file.sync_all()
+        })
+        .await
+        .map_err(migration_error)?
+        .map_err(migration_error)?;
+
+        self.project_rollout_in_batches(thread_id, &staged_path, limiter)
+            .await?;
+        let projection = thread_history::projection_state(self, thread_id)
+            .await?
+            .ok_or_else(|| migration_error("completed rollout has no SQLite projection"))?;
+        if projection.next_byte_offset != expected_length
+            || projection.next_ordinal != expected_ordinal
+        {
+            return Err(migration_error(
+                "SQLite projection does not cover the complete staged rollout",
+            ));
+        }
+
+        let compressed_staged_path = if compressed {
+            let path = compressed_staged_rollout_path(rollout_path)?;
+            compress_rollout_to_path(&staged_path, &path, source_permissions, source_modified)
+                .await?;
+            Some(path)
+        } else {
+            None
+        };
+
+        // Older writers do not know about migration locks. Keep the legacy path visible if its
+        // append-only source changed while the replacement rollout was being staged.
+        let current_source_metadata = tokio::fs::metadata(rollout_path)
+            .await
+            .map_err(migration_error)?;
+        if current_source_metadata.len() != source_metadata.len()
+            || current_source_metadata.modified().ok() != source_modified
+        {
+            return Err(ThreadStoreError::Conflict {
+                message: "rollout changed while migration was staging it; close older Codex processes and retry".to_string(),
+            });
+        }
+
+        if let Some(compressed_staged_path) = compressed_staged_path {
+            tokio::fs::rename(compressed_staged_path, rollout_path)
+                .await
+                .map_err(migration_error)?;
+            remove_file_if_present(&staged_path).await?;
+            if let Some(decompressed_path) = decompressed_path.as_ref() {
+                remove_file_if_present(decompressed_path).await?;
+            }
+        } else {
+            tokio::fs::rename(&staged_path, rollout_path)
+                .await
+                .map_err(migration_error)?;
+        }
+        sync_parent_directory(rollout_path).await?;
+        self.finish_published_migration(thread_id, journal_path)
+            .await
+    }
+
+    async fn write_canonical_rollout(
+        input: &CanonicalizationSource<'_>,
+        limiter: &mut RolloutMigrationRateLimiter,
+    ) -> ThreadStoreResult<(u64, u64)> {
+        let source_file = File::open(input.source_path)
+            .await
+            .map_err(migration_error)?;
+        let staged_file = tokio::fs::OpenOptions::new()
+            .create(true)
+            .truncate(true)
+            .write(true)
+            .open(input.staged_path)
+            .await
+            .map_err(migration_error)?;
+        staged_file
+            .set_permissions(input.source_permissions.clone())
+            .await
+            .map_err(migration_error)?;
+        let mut source = BufReader::with_capacity(PROJECTION_BATCH_BYTES as usize, source_file);
+        let mut staged = BufWriter::with_capacity(PROJECTION_BATCH_BYTES as usize, staged_file);
+        let mut bytes = Vec::new();
+        let mut canonicalizer = LegacyRolloutCanonicalizer::new(input.thread_id);
+        let written = canonicalizer
+            .write_head_session_meta(input.canonical_session_meta.clone(), &mut staged)
+            .await?;
+        limiter.account(written).await;
+        let mut last_timestamp = input.canonical_session_meta.timestamp.clone();
+
+        while let Some(record) = read_rollout_record(&mut source, &mut bytes).await? {
+            limiter.account(record.byte_count).await;
+            let Some(line) = record.line else {
+                continue;
+            };
+            if matches!(
+                &line.item,
+                RolloutItem::EventMsg(codex_protocol::protocol::EventMsg::ThreadRolledBack(_))
+            ) {
+                return Err(migration_error("legacy rollout contains a rollback marker"));
+            }
+            last_timestamp = line.timestamp.clone();
+            let written = canonicalizer.process_line(line, &mut staged).await?;
+            limiter.account(written).await;
+        }
+
+        let written = canonicalizer.finish(&mut staged, &last_timestamp).await?;
+        limiter.account(written).await;
+        staged.flush().await.map_err(migration_error)?;
+        Ok((
+            canonicalizer.output_byte_offset(),
+            canonicalizer.next_ordinal(),
+        ))
+    }
+
+    async fn recover_published_migration(
+        &self,
+        thread_id: ThreadId,
+        rollout_path: &Path,
+        journal_path: &Path,
+        limiter: &mut RolloutMigrationRateLimiter,
+    ) -> ThreadStoreResult<()> {
+        let _writer_guard = self.writer_lock_coordinator.acquire(thread_id)?;
+        let decompressed_path = rollout_path_is_compressed(rollout_path)
+            .then(|| decompressed_staged_rollout_path(rollout_path))
+            .transpose()?;
+        let projection_path = if let Some(decompressed_path) = decompressed_path.as_ref() {
+            decompress_rollout_to_path(rollout_path, decompressed_path).await?;
+            let decompressed_bytes = tokio::fs::metadata(decompressed_path)
+                .await
+                .map_err(migration_error)?
+                .len();
+            limiter
+                .account(
+                    tokio::fs::metadata(rollout_path)
+                        .await
+                        .map_err(migration_error)?
+                        .len()
+                        .saturating_add(decompressed_bytes),
+                )
+                .await;
+            decompressed_path.as_path()
+        } else {
+            rollout_path
+        };
+        let expected_length = tokio::fs::metadata(projection_path)
+            .await
+            .map_err(migration_error)?
+            .len();
+        let projection = thread_history::projection_state(self, thread_id).await?;
+        if projection.is_none_or(|state| state.next_byte_offset != expected_length) {
+            thread_history::delete_thread(self, thread_id).await?;
+            self.project_rollout_in_batches(thread_id, projection_path, limiter)
+                .await?;
+        }
+        remove_file_if_present(&staged_rollout_path(rollout_path)?).await?;
+        remove_file_if_present(&compressed_staged_rollout_path(rollout_path)?).await?;
+        if let Some(decompressed_path) = decompressed_path.as_ref() {
+            remove_file_if_present(decompressed_path).await?;
+        }
+        self.finish_published_migration(thread_id, journal_path)
+            .await
+    }
+
+    async fn cleanup_failed_unpublished_migration(
+        &self,
+        thread_id: ThreadId,
+        rollout_path: &Path,
+        journal_path: &Path,
+    ) -> ThreadStoreResult<()> {
+        let Ok(metadata) = codex_rollout::read_session_meta_line(rollout_path).await else {
+            return Ok(());
+        };
+        if metadata.meta.history_mode == ThreadHistoryMode::Paginated {
+            return Ok(());
+        }
+
+        thread_history::delete_thread(self, thread_id).await?;
+        remove_file_if_present(&staged_rollout_path(rollout_path)?).await?;
+        remove_file_if_present(&compressed_staged_rollout_path(rollout_path)?).await?;
+        remove_file_if_present(&decompressed_staged_rollout_path(rollout_path)?).await?;
+        remove_file_if_present(journal_path).await?;
+        sync_parent_directory(journal_path).await
+    }
+
+    async fn finish_published_migration(
+        &self,
+        thread_id: ThreadId,
+        journal_path: &Path,
+    ) -> ThreadStoreResult<()> {
+        if let Some(state_db) = &self.state_db
+            && !state_db
+                .mark_thread_paginated(thread_id)
+                .await
+                .map_err(migration_error)?
+        {
+            return Err(migration_error(format!(
+                "thread {thread_id} is missing its SQLite metadata"
+            )));
+        }
+        tokio::fs::remove_file(journal_path)
+            .await
+            .map_err(migration_error)?;
+        sync_parent_directory(journal_path).await
+    }
+
+    async fn project_rollout_in_batches(
+        &self,
+        thread_id: ThreadId,
+        rollout_path: &Path,
+        limiter: &mut RolloutMigrationRateLimiter,
+    ) -> ThreadStoreResult<()> {
+        let file = File::open(rollout_path).await.map_err(migration_error)?;
+        let mut reader = BufReader::with_capacity(PROJECTION_BATCH_BYTES as usize, file);
+        let mut line_bytes = Vec::new();
+        let mut batch = Vec::new();
+        let mut batch_start = 0_u64;
+        let mut offset = 0_u64;
+
+        while let Some(record) = read_rollout_record(&mut reader, &mut line_bytes).await? {
+            let next_offset = offset
+                .checked_add(record.byte_count)
+                .ok_or_else(|| migration_error("staged rollout byte offset overflow"))?;
+            limiter.account(record.byte_count).await;
+            let Some(line) = record.line else {
+                offset = next_offset;
+                continue;
+            };
+            let ordinal = line
+                .ordinal
+                .ok_or_else(|| migration_error("staged rollout line is missing its ordinal"))?;
+            let fallback_created_at_ms = DateTime::parse_from_rfc3339(&line.timestamp)
+                .map_err(migration_error)?
+                .timestamp_millis();
+            batch.push(RolloutProjectionStep::Line(ProjectedRolloutLine {
+                ordinal,
+                start_byte_offset: offset,
+                end_byte_offset: next_offset,
+                fallback_created_at_ms: Some(fallback_created_at_ms),
+                changes: project_rollout_line(&line),
+            }));
+            offset = next_offset;
+
+            if offset.saturating_sub(batch_start) >= PROJECTION_BATCH_BYTES {
+                thread_history::apply_projection(
+                    self,
+                    thread_id,
+                    batch_start,
+                    offset,
+                    /*initial_ordinal*/ 0,
+                    std::mem::take(&mut batch),
+                )
+                .await?;
+                batch_start = offset;
+            }
+        }
+
+        if batch_start != offset {
+            thread_history::apply_projection(
+                self,
+                thread_id,
+                batch_start,
+                offset,
+                /*initial_ordinal*/ 0,
+                batch,
+            )
+            .await?;
+        }
+        Ok(())
+    }
+}
+
+async fn read_rollout_record(
+    reader: &mut BufReader<File>,
+    bytes: &mut Vec<u8>,
+) -> ThreadStoreResult<Option<RolloutRecord>> {
+    bytes.clear();
+    let byte_count = reader
+        .take((MAX_ROLLOUT_LINE_BYTES + 1) as u64)
+        .read_until(b'\n', bytes)
+        .await
+        .map_err(migration_error)?;
+    if byte_count == 0 {
+        return Ok(None);
+    }
+    if byte_count > MAX_ROLLOUT_LINE_BYTES {
+        return Err(migration_error(
+            "rollout contains an oversized JSONL record",
+        ));
+    }
+    let line = line_parser::parse_legacy_rollout_line(bytes).map_err(migration_error)?;
+    Ok(Some(RolloutRecord {
+        line,
+        byte_count: byte_count as u64,
+    }))
+}
+
+async fn find_rollout_paths(root: &Path) -> ThreadStoreResult<Vec<PathBuf>> {
+    let mut directories = vec![root.to_path_buf()];
+    let mut paths = Vec::new();
+
+    while let Some(directory) = directories.pop() {
+        let mut entries = match tokio::fs::read_dir(&directory).await {
+            Ok(entries) => entries,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
+            Err(error) => return Err(migration_error(error)),
+        };
+        while let Some(entry) = entries.next_entry().await.map_err(migration_error)? {
+            let kind = entry.file_type().await.map_err(migration_error)?;
+            if kind.is_dir() {
+                directories.push(entry.path());
+                continue;
+            }
+            if !kind.is_file() {
+                continue;
+            }
+            let name = entry.file_name();
+            let Some(name) = name.to_str() else {
+                continue;
+            };
+            if name.starts_with("rollout-")
+                && (name.ends_with(".jsonl") || name.ends_with(".jsonl.zst"))
+            {
+                paths.push(entry.path());
+            }
+        }
+    }
+
+    paths.sort_by(|left, right| right.cmp(left));
+    Ok(paths)
+}
+
+fn matches_selection(selected: &[ThreadId], actual: Option<ThreadId>) -> bool {
+    selected.is_empty() || actual.is_some_and(|thread_id| selected.contains(&thread_id))
+}
+
+fn thread_id_from_rollout_filename(path: &Path) -> Option<ThreadId> {
+    let name = path.file_name()?.to_str()?;
+    let stem = name
+        .strip_suffix(".jsonl.zst")
+        .or_else(|| name.strip_suffix(".jsonl"))?;
+    let start = stem.len().checked_sub(36)?;
+    ThreadId::from_string(stem.get(start..)?).ok()
+}
+
+fn rollout_path_is_compressed(path: &Path) -> bool {
+    path.file_name()
+        .and_then(|name| name.to_str())
+        .is_some_and(|name| name.ends_with(".jsonl.zst"))
+}
+
+fn migration_outcome(
+    thread_id: ThreadId,
+    rollout_path: PathBuf,
+    result: ThreadStoreResult<RolloutMigrationStatus>,
+    bytes_processed: u64,
+) -> RolloutMigrationOutcome {
+    match result {
+        Ok(status) => RolloutMigrationOutcome {
+            thread_id: Some(thread_id),
+            rollout_path,
+            status,
+            bytes_processed,
+            message: None,
+        },
+        Err(error) => RolloutMigrationOutcome {
+            thread_id: Some(thread_id),
+            rollout_path,
+            status: RolloutMigrationStatus::Failed,
+            bytes_processed,
+            message: Some(error.to_string()),
+        },
+    }
+}
+
+fn skipped_busy_outcome(
+    thread_id: ThreadId,
+    rollout_path: PathBuf,
+    message: String,
+    bytes_processed: u64,
+) -> RolloutMigrationOutcome {
+    RolloutMigrationOutcome {
+        thread_id: Some(thread_id),
+        rollout_path,
+        status: RolloutMigrationStatus::SkippedBusy,
+        bytes_processed,
+        message: Some(message),
+    }
+}
+
+fn migration_error(error: impl std::fmt::Display) -> ThreadStoreError {
+    ThreadStoreError::Internal {
+        message: format!("rollout migration failed: {error}"),
+    }
+}
+
+#[cfg(test)]
+#[path = "rollout_migration_tests.rs"]
+mod tests;

codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs

@@ -0,0 +1,505 @@
+//! Replays normalized legacy rollout records into canonical paginated JSONL.
+//!
+//! `line_parser` makes old JSON shapes parseable, `legacy_event` converts obsolete completion
+//! events into modern turn items. This module assigns stable ordinals, keeps `SessionMeta` at
+//! ordinal zero, and emits the surviving history selected by the caller.
+//!
+//! The goal is to preserve the model-visible conversation, not to preserve every legacy record
+//! byte-for-byte. Filesystem publishing and SQLite projection intentionally live outside this
+//! module.
+
+use chrono::DateTime;
+use codex_protocol::ThreadId;
+use codex_protocol::items::ReasoningItem;
+use codex_protocol::items::TurnItem;
+use codex_protocol::items::parse_hook_prompt_message;
+use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::ItemCompletedEvent;
+use codex_protocol::protocol::RolloutItem;
+use codex_protocol::protocol::RolloutLine;
+use codex_protocol::protocol::ThreadHistoryMode;
+use codex_protocol::protocol::TurnCompleteEvent;
+use codex_protocol::protocol::TurnStartedEvent;
+use std::collections::HashSet;
+use tokio::fs::File;
+use tokio::io::AsyncWriteExt;
+use tokio::io::BufWriter;
+
+use super::legacy_event;
+use super::migration_error;
+use crate::ThreadStoreResult;
+
+#[derive(Clone)]
+struct ActiveTurn {
+    id: String,
+    explicit: bool,
+    saw_user: bool,
+}
+
+enum ReasoningTextKind {
+    Summary,
+    Raw,
+}
+
+pub(super) struct LegacyRolloutCanonicalizer {
+    thread_id: ThreadId,
+    next_ordinal: u64,
+    next_item_index: u64,
+    output_byte_offset: u64,
+    bytes_written: u64,
+    source_line_index: u64,
+    active_turn: Option<ActiveTurn>,
+    known_turn_ids: HashSet<String>,
+    reasoning: Option<ReasoningItem>,
+    saw_source_session_meta: bool,
+}
+
+impl LegacyRolloutCanonicalizer {
+    pub(super) fn new(thread_id: ThreadId) -> Self {
+        Self {
+            thread_id,
+            next_ordinal: 0,
+            next_item_index: 1,
+            output_byte_offset: 0,
+            bytes_written: 0,
+            source_line_index: 0,
+            active_turn: None,
+            known_turn_ids: HashSet::new(),
+            reasoning: None,
+            saw_source_session_meta: false,
+        }
+    }
+
+    pub(super) fn output_byte_offset(&self) -> u64 {
+        self.output_byte_offset
+    }
+
+    pub(super) fn next_ordinal(&self) -> u64 {
+        self.next_ordinal
+    }
+
+    pub(super) async fn write_head_session_meta(
+        &mut self,
+        line: RolloutLine,
+        writer: &mut BufWriter<File>,
+    ) -> ThreadStoreResult<u64> {
+        let timestamp = line.timestamp;
+        let RolloutItem::SessionMeta(mut metadata) = line.item else {
+            return Err(migration_error("canonical session metadata is missing"));
+        };
+        if metadata.meta.id != self.thread_id {
+            return Err(migration_error("rollout metadata thread id changed"));
+        }
+        metadata.meta.history_mode = ThreadHistoryMode::Paginated;
+        metadata.meta.history_base = None;
+        metadata.meta.subagent_history_start_ordinal = None;
+
+        let bytes_before = self.bytes_written;
+        self.write_item(writer, &timestamp, RolloutItem::SessionMeta(metadata))
+            .await?;
+        Ok(self.bytes_written - bytes_before)
+    }
+
+    pub(super) async fn process_line(
+        &mut self,
+        line: RolloutLine,
+        writer: &mut BufWriter<File>,
+    ) -> ThreadStoreResult<u64> {
+        let source_index = self.source_line_index;
+        self.source_line_index = self
+            .source_line_index
+            .checked_add(1)
+            .ok_or_else(|| migration_error("legacy rollout line index overflow"))?;
+        let timestamp = line.timestamp;
+        let bytes_before = self.bytes_written;
+        match line.item {
+            RolloutItem::SessionMeta(metadata) => {
+                if !self.saw_source_session_meta {
+                    self.saw_source_session_meta = true;
+                    return Ok(0);
+                }
+                self.write_item(writer, &timestamp, RolloutItem::SessionMeta(metadata))
+                    .await?;
+            }
+            RolloutItem::ResponseItem(ResponseItem::Other) => {
+                return Err(migration_error(
+                    "legacy rollout contains an unsupported response item",
+                ));
+            }
+            RolloutItem::ResponseItem(response) => {
+                let hook = match &response {
+                    ResponseItem::Message {
+                        role, content, id, ..
+                    } if role == "user" => parse_hook_prompt_message(id.as_deref(), content),
+                    _ => None,
+                };
+                self.write_item(writer, &timestamp, RolloutItem::ResponseItem(response))
+                    .await?;
+                if let Some(hook) = hook {
+                    self.ensure_turn(writer, &timestamp, source_index).await?;
+                    self.reasoning = None;
+                    self.write_completed_item(writer, &timestamp, TurnItem::HookPrompt(hook))
+                        .await?;
+                }
+            }
+            RolloutItem::EventMsg(EventMsg::ThreadRolledBack(_)) => {
+                return Err(migration_error(
+                    "rollback marker reached canonical writer without a rollback plan",
+                ));
+            }
+            RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => {
+                self.finish_implicit_turn(writer, &timestamp).await?;
+                self.known_turn_ids.insert(event.turn_id.clone());
+                self.active_turn = Some(ActiveTurn {
+                    id: event.turn_id.clone(),
+                    explicit: true,
+                    saw_user: false,
+                });
+                self.reasoning = None;
+                self.write_item(
+                    writer,
+                    &timestamp,
+                    RolloutItem::EventMsg(EventMsg::TurnStarted(event)),
+                )
+                .await?;
+            }
+            RolloutItem::EventMsg(EventMsg::TurnComplete(event)) => {
+                self.reasoning = None;
+                if self
+                    .active_turn
+                    .as_ref()
+                    .is_some_and(|turn| turn.id == event.turn_id)
+                {
+                    self.active_turn = None;
+                }
+                self.write_item(
+                    writer,
+                    &timestamp,
+                    RolloutItem::EventMsg(EventMsg::TurnComplete(event)),
+                )
+                .await?;
+            }
+            RolloutItem::EventMsg(EventMsg::TurnAborted(mut event)) => {
+                if event.turn_id.is_none() {
+                    event.turn_id = self.active_turn.as_ref().map(|turn| turn.id.clone());
+                }
+                if self
+                    .active_turn
+                    .as_ref()
+                    .is_some_and(|turn| event.turn_id.as_deref() == Some(turn.id.as_str()))
+                {
+                    self.active_turn = None;
+                }
+                self.reasoning = None;
+                self.write_item(
+                    writer,
+                    &timestamp,
+                    RolloutItem::EventMsg(EventMsg::TurnAborted(event)),
+                )
+                .await?;
+            }
+            RolloutItem::EventMsg(EventMsg::UserMessage(event)) => {
+                if self
+                    .active_turn
+                    .as_ref()
+                    .is_some_and(|turn| !turn.explicit && turn.saw_user)
+                {
+                    self.finish_implicit_turn(writer, &timestamp).await?;
+                }
+                self.ensure_turn(writer, &timestamp, source_index).await?;
+                let item = legacy_event::user_message_item(event, &mut || self.next_item_id())?;
+                if let Some(turn) = self.active_turn.as_mut() {
+                    turn.saw_user = true;
+                }
+                self.reasoning = None;
+                self.write_completed_item(writer, &timestamp, item).await?;
+            }
+            RolloutItem::EventMsg(EventMsg::AgentReasoning(event)) => {
+                self.write_reasoning(
+                    writer,
+                    &timestamp,
+                    source_index,
+                    event.text,
+                    ReasoningTextKind::Summary,
+                )
+                .await?;
+            }
+            RolloutItem::EventMsg(EventMsg::AgentReasoningRawContent(event)) => {
+                self.write_reasoning(
+                    writer,
+                    &timestamp,
+                    source_index,
+                    event.text,
+                    ReasoningTextKind::Raw,
+                )
+                .await?;
+            }
+            RolloutItem::EventMsg(EventMsg::ItemCompleted(mut event)) => {
+                event.thread_id = self.thread_id;
+                self.reasoning = None;
+                self.write_item(
+                    writer,
+                    &timestamp,
+                    RolloutItem::EventMsg(EventMsg::ItemCompleted(event)),
+                )
+                .await?;
+            }
+            RolloutItem::EventMsg(event) => {
+                if let Some((item, turn_id)) =
+                    legacy_event::completed_item(&event, &mut || self.next_item_id())?
+                {
+                    match turn_id {
+                        Some(turn_id)
+                            if self
+                                .active_turn
+                                .as_ref()
+                                .is_some_and(|turn| turn.id.as_str() != turn_id.as_str()) =>
+                        {
+                            self.write_completed_item_to_turn(writer, &timestamp, turn_id, item)
+                                .await?;
+                        }
+                        Some(turn_id) => {
+                            if self.active_turn.is_none()
+                                && self.known_turn_ids.contains(turn_id.as_str())
+                            {
+                                self.reasoning = None;
+                                self.write_completed_item_to_turn(
+                                    writer, &timestamp, turn_id, item,
+                                )
+                                .await?;
+                            } else if self.active_turn.is_none() {
+                                self.start_implicit_turn(writer, &timestamp, turn_id)
+                                    .await?;
+                                self.reasoning = None;
+                                self.write_completed_item(writer, &timestamp, item).await?;
+                            } else {
+                                self.reasoning = None;
+                                self.write_completed_item(writer, &timestamp, item).await?;
+                            }
+                        }
+                        None => {
+                            self.ensure_turn(writer, &timestamp, source_index).await?;
+                            self.reasoning = None;
+                            self.write_completed_item(writer, &timestamp, item).await?;
+                        }
+                    }
+                } else {
+                    let item = RolloutItem::EventMsg(event);
+                    if codex_rollout::is_persisted_rollout_item(&item, ThreadHistoryMode::Paginated)
+                    {
+                        self.write_item(writer, &timestamp, item).await?;
+                    }
+                }
+            }
+            item @ RolloutItem::InterAgentCommunication(_) => {
+                self.write_item(writer, &timestamp, item).await?;
+            }
+            item @ RolloutItem::Compacted(_) => {
+                self.write_item(writer, &timestamp, item).await?;
+            }
+            item @ (RolloutItem::InterAgentCommunicationMetadata { .. }
+            | RolloutItem::TurnContext(_)
+            | RolloutItem::WorldState(_)) => {
+                self.write_item(writer, &timestamp, item).await?;
+            }
+        }
+
+        Ok(self.bytes_written - bytes_before)
+    }
+
+    pub(super) async fn finish(
+        &mut self,
+        writer: &mut BufWriter<File>,
+        timestamp: &str,
+    ) -> ThreadStoreResult<u64> {
+        let bytes_before = self.bytes_written;
+        self.finish_implicit_turn(writer, timestamp).await?;
+        Ok(self.bytes_written - bytes_before)
+    }
+
+    async fn ensure_turn(
+        &mut self,
+        writer: &mut BufWriter<File>,
+        timestamp: &str,
+        source_index: u64,
+    ) -> ThreadStoreResult<()> {
+        if self.active_turn.is_some() {
+            return Ok(());
+        }
+        let turn_id = format!("rollout-{source_index}");
+        self.start_implicit_turn(writer, timestamp, turn_id).await
+    }
+
+    async fn start_implicit_turn(
+        &mut self,
+        writer: &mut BufWriter<File>,
+        timestamp: &str,
+        turn_id: String,
+    ) -> ThreadStoreResult<()> {
+        self.active_turn = Some(ActiveTurn {
+            id: turn_id.clone(),
+            explicit: false,
+            saw_user: false,
+        });
+        self.known_turn_ids.insert(turn_id.clone());
+        self.reasoning = None;
+        let started_at = DateTime::parse_from_rfc3339(timestamp)
+            .map_err(migration_error)?
+            .timestamp();
+        self.write_item(
+            writer,
+            timestamp,
+            RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
+                turn_id,
+                trace_id: None,
+                started_at: Some(started_at),
+                model_context_window: None,
+                collaboration_mode_kind: Default::default(),
+            })),
+        )
+        .await
+    }
+
+    async fn finish_implicit_turn(
+        &mut self,
+        writer: &mut BufWriter<File>,
+        timestamp: &str,
+    ) -> ThreadStoreResult<()> {
+        let Some(turn) = self.active_turn.as_ref() else {
+            return Ok(());
+        };
+        if turn.explicit {
+            return Ok(());
+        }
+        let turn_id = turn.id.clone();
+        self.active_turn = None;
+        self.reasoning = None;
+        self.write_item(
+            writer,
+            timestamp,
+            RolloutItem::EventMsg(EventMsg::TurnComplete(TurnCompleteEvent {
+                turn_id,
+                last_agent_message: None,
+                error: None,
+                started_at: None,
+                completed_at: Some(
+                    DateTime::parse_from_rfc3339(timestamp)
+                        .map_err(migration_error)?
+                        .timestamp(),
+                ),
+                duration_ms: None,
+                time_to_first_token_ms: None,
+            })),
+        )
+        .await
+    }
+
+    async fn write_completed_item(
+        &mut self,
+        writer: &mut BufWriter<File>,
+        timestamp: &str,
+        item: TurnItem,
+    ) -> ThreadStoreResult<()> {
+        let turn_id = self
+            .active_turn
+            .as_ref()
+            .map(|turn| turn.id.clone())
+            .ok_or_else(|| migration_error("completed rollout item has no active turn"))?;
+        self.write_completed_item_to_turn(writer, timestamp, turn_id, item)
+            .await
+    }
+
+    async fn write_completed_item_to_turn(
+        &mut self,
+        writer: &mut BufWriter<File>,
+        timestamp: &str,
+        turn_id: String,
+        item: TurnItem,
+    ) -> ThreadStoreResult<()> {
+        let completed_at_ms = DateTime::parse_from_rfc3339(timestamp)
+            .map_err(migration_error)?
+            .timestamp_millis();
+        self.write_item(
+            writer,
+            timestamp,
+            RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent {
+                thread_id: self.thread_id,
+                turn_id,
+                item,
+                started_at_ms: None,
+                completed_at_ms,
+            })),
+        )
+        .await
+    }
+
+    async fn write_reasoning(
+        &mut self,
+        writer: &mut BufWriter<File>,
+        timestamp: &str,
+        source_index: u64,
+        text: String,
+        kind: ReasoningTextKind,
+    ) -> ThreadStoreResult<()> {
+        if text.is_empty() {
+            return Ok(());
+        }
+        self.ensure_turn(writer, timestamp, source_index).await?;
+        let mut item = match self.reasoning.take() {
+            Some(item) => item,
+            None => ReasoningItem {
+                id: self.next_item_id()?,
+                summary_text: Vec::new(),
+                raw_content: Vec::new(),
+            },
+        };
+        match kind {
+            ReasoningTextKind::Summary => item.summary_text.push(text),
+            ReasoningTextKind::Raw => item.raw_content.push(text),
+        }
+        self.reasoning = Some(item.clone());
+        self.write_completed_item(writer, timestamp, TurnItem::Reasoning(item))
+            .await
+    }
+
+    async fn write_item(
+        &mut self,
+        writer: &mut BufWriter<File>,
+        timestamp: &str,
+        item: RolloutItem,
+    ) -> ThreadStoreResult<()> {
+        let mut bytes = serde_json::to_vec(&RolloutLine {
+            timestamp: timestamp.to_string(),
+            ordinal: Some(self.next_ordinal),
+            item,
+        })
+        .map_err(migration_error)?;
+        bytes.push(b'\n');
+        writer.write_all(&bytes).await.map_err(migration_error)?;
+        let byte_count = u64::try_from(bytes.len())
+            .map_err(|_| migration_error("rollout record exceeds addressable size"))?;
+        self.output_byte_offset = self
+            .output_byte_offset
+            .checked_add(byte_count)
+            .ok_or_else(|| migration_error("paginated rollout byte offset overflow"))?;
+        self.bytes_written = self
+            .bytes_written
+            .checked_add(byte_count)
+            .ok_or_else(|| migration_error("paginated rollout byte count overflow"))?;
+        self.next_ordinal = self
+            .next_ordinal
+            .checked_add(1)
+            .ok_or_else(|| migration_error("paginated rollout ordinal overflow"))?;
+        Ok(())
+    }
+
+    fn next_item_id(&mut self) -> ThreadStoreResult<String> {
+        let item_id = format!("item-{}", self.next_item_index);
+        self.next_item_index = self
+            .next_item_index
+            .checked_add(1)
+            .ok_or_else(|| migration_error("legacy rollout item id overflow"))?;
+        Ok(item_id)
+    }
+}

codex-rs/thread-store/src/local/rollout_migration/legacy_event.rs

@@ -0,0 +1,251 @@
+//! Converts legacy persisted events into modern paginated turn items.
+//!
+//! Legacy rollouts persisted events like `ExecCommandEnd`, `PatchApplyEnd`, and
+//! `McpToolCallEnd`. Paginated history instead persists `ItemCompleted(item)` records containing
+//! canonical `TurnItem`s.
+//!
+//! This module owns that old-event -> new-item mapping, including stable synthesized item IDs.
+//! It intentionally stays separate from live thread-history reduction because migration needs a
+//! frozen adapter for historical rollout payloads.
+
+use codex_protocol::items::AgentMessageContent;
+use codex_protocol::items::AgentMessageItem;
+use codex_protocol::items::CommandExecutionItem;
+use codex_protocol::items::ContextCompactionItem;
+use codex_protocol::items::DynamicToolCallItem;
+use codex_protocol::items::DynamicToolCallStatus;
+use codex_protocol::items::EnteredReviewModeItem;
+use codex_protocol::items::ExitedReviewModeItem;
+use codex_protocol::items::FileChangeItem;
+use codex_protocol::items::ImageGenerationItem;
+use codex_protocol::items::McpToolCallError;
+use codex_protocol::items::McpToolCallItem;
+use codex_protocol::items::McpToolCallStatus;
+use codex_protocol::items::SubAgentActivityItem;
+use codex_protocol::items::TurnItem;
+use codex_protocol::items::UserMessageItem;
+use codex_protocol::items::WebSearchItem;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::UserMessageEvent;
+use codex_protocol::user_input::UserInput;
+
+use crate::ThreadStoreResult;
+
+pub(super) fn user_message_item(
+    event: UserMessageEvent,
+    next_item_id: &mut impl FnMut() -> ThreadStoreResult<String>,
+) -> ThreadStoreResult<TurnItem> {
+    let mut content = Vec::new();
+    if !event.message.trim().is_empty() {
+        content.push(UserInput::Text {
+            text: event.message,
+            text_elements: event.text_elements,
+        });
+    }
+    if let Some(images) = event.images {
+        for (index, image_url) in images.into_iter().enumerate() {
+            content.push(UserInput::Image {
+                image_url,
+                detail: event.image_details.get(index).copied().flatten(),
+            });
+        }
+    }
+    for (index, path) in event.local_images.into_iter().enumerate() {
+        content.push(UserInput::LocalImage {
+            path,
+            detail: event.local_image_details.get(index).copied().flatten(),
+        });
+    }
+    if let Some(audio) = event.audio {
+        content.extend(
+            audio
+                .into_iter()
+                .map(|audio_url| UserInput::Audio { audio_url }),
+        );
+    }
+    content.extend(
+        event
+            .local_audio
+            .into_iter()
+            .map(|path| UserInput::LocalAudio { path }),
+    );
+    Ok(TurnItem::UserMessage(UserMessageItem {
+        id: next_item_id()?,
+        client_id: event.client_id,
+        content,
+    }))
+}
+
+/// Convert historical completion events into the canonical completed-item payloads
+/// that paginated rollouts persist. This deliberately remains a frozen migration
+/// adapter rather than sharing the live thread-history reducer: migration must
+/// preserve historical payloads while synthesizing stable item and turn IDs.
+pub(super) fn completed_item(
+    event: &EventMsg,
+    next_item_id: &mut impl FnMut() -> ThreadStoreResult<String>,
+) -> ThreadStoreResult<Option<(TurnItem, Option<String>)>> {
+    let result = match event {
+        EventMsg::AgentMessage(event) if !event.message.is_empty() => Some((
+            TurnItem::AgentMessage(AgentMessageItem {
+                id: next_item_id()?,
+                content: vec![AgentMessageContent::Text {
+                    text: event.message.clone(),
+                }],
+                phase: event.phase.clone(),
+                memory_citation: event.memory_citation.clone(),
+            }),
+            None,
+        )),
+        EventMsg::PatchApplyEnd(event) => Some((
+            TurnItem::FileChange(FileChangeItem {
+                id: event.call_id.clone(),
+                changes: event.changes.clone(),
+                status: Some(event.status.clone()),
+                auto_approved: None,
+                stdout: (!event.stdout.is_empty()).then(|| event.stdout.clone()),
+                stderr: (!event.stderr.is_empty()).then(|| event.stderr.clone()),
+            }),
+            (!event.turn_id.is_empty()).then(|| event.turn_id.clone()),
+        )),
+        EventMsg::McpToolCallEnd(event) => {
+            let (result, error) = match &event.result {
+                Ok(result) => (Some(result.clone()), None),
+                Err(message) => (
+                    None,
+                    Some(McpToolCallError {
+                        message: message.clone(),
+                    }),
+                ),
+            };
+            Some((
+                TurnItem::McpToolCall(McpToolCallItem {
+                    id: event.call_id.clone(),
+                    server: event.invocation.server.clone(),
+                    tool: event.invocation.tool.clone(),
+                    arguments: event
+                        .invocation
+                        .arguments
+                        .clone()
+                        .unwrap_or(serde_json::Value::Null),
+                    connector_id: event.connector_id.clone(),
+                    mcp_app_resource_uri: event.mcp_app_resource_uri.clone(),
+                    link_id: event.link_id.clone(),
+                    app_name: event.app_name.clone(),
+                    action_name: event.action_name.clone(),
+                    plugin_id: event.plugin_id.clone(),
+                    read_only_hint: event.read_only_hint,
+                    status: if event.is_success() {
+                        McpToolCallStatus::Completed
+                    } else {
+                        McpToolCallStatus::Failed
+                    },
+                    result,
+                    error,
+                    duration: Some(event.duration),
+                }),
+                None,
+            ))
+        }
+        EventMsg::WebSearchEnd(event) => Some((
+            TurnItem::WebSearch(WebSearchItem {
+                id: event.call_id.clone(),
+                query: event.query.clone(),
+                action: event.action.clone(),
+                results: event.results.clone(),
+            }),
+            None,
+        )),
+        EventMsg::ImageGenerationEnd(event) => Some((
+            TurnItem::ImageGeneration(ImageGenerationItem {
+                id: event.call_id.clone(),
+                status: event.status.clone(),
+                revised_prompt: event.revised_prompt.clone(),
+                result: event.result.clone(),
+                saved_path: event.saved_path.clone(),
+            }),
+            None,
+        )),
+        EventMsg::ContextCompacted(_) => Some((
+            TurnItem::ContextCompaction(ContextCompactionItem {
+                id: next_item_id()?,
+            }),
+            None,
+        )),
+        EventMsg::EnteredReviewMode(event) => Some((
+            TurnItem::EnteredReviewMode(EnteredReviewModeItem {
+                id: match event.item_id.clone() {
+                    Some(id) => id,
+                    None => next_item_id()?,
+                },
+                target: event.target.clone(),
+                user_facing_hint: event
+                    .user_facing_hint
+                    .clone()
+                    .unwrap_or_else(|| "Review requested.".to_string()),
+            }),
+            event.turn_id.clone(),
+        )),
+        EventMsg::ExitedReviewMode(event) => Some((
+            TurnItem::ExitedReviewMode(ExitedReviewModeItem {
+                id: match event.item_id.clone() {
+                    Some(id) => id,
+                    None => next_item_id()?,
+                },
+                review_output: event.review_output.clone(),
+            }),
+            event.turn_id.clone(),
+        )),
+        EventMsg::SubAgentActivity(event) => Some((
+            TurnItem::SubAgentActivity(SubAgentActivityItem {
+                id: event.event_id.clone(),
+                kind: event.kind,
+                agent_thread_id: event.agent_thread_id,
+                agent_path: event.agent_path.clone(),
+            }),
+            None,
+        )),
+        EventMsg::ExecCommandEnd(event) => Some((
+            TurnItem::CommandExecution(CommandExecutionItem {
+                id: event.call_id.clone(),
+                plugin_id: event.plugin_id.clone(),
+                script_path: event.script_path.clone(),
+                process_id: event.process_id.clone(),
+                command: event.command.clone(),
+                cwd: event.cwd.clone(),
+                parsed_cmd: event.parsed_cmd.clone(),
+                source: event.source,
+                interaction_input: event.interaction_input.clone(),
+                status: event.status.clone().into(),
+                stdout: (!event.stdout.is_empty()).then(|| event.stdout.clone()),
+                stderr: (!event.stderr.is_empty()).then(|| event.stderr.clone()),
+                aggregated_output: (!event.aggregated_output.is_empty())
+                    .then(|| event.aggregated_output.clone()),
+                exit_code: Some(event.exit_code),
+                duration: Some(event.duration),
+                formatted_output: (!event.formatted_output.is_empty())
+                    .then(|| event.formatted_output.clone()),
+            }),
+            Some(event.turn_id.clone()),
+        )),
+        EventMsg::DynamicToolCallResponse(event) => Some((
+            TurnItem::DynamicToolCall(DynamicToolCallItem {
+                id: event.call_id.clone(),
+                namespace: event.namespace.clone(),
+                tool: event.tool.clone(),
+                arguments: event.arguments.clone(),
+                status: if event.success {
+                    DynamicToolCallStatus::Completed
+                } else {
+                    DynamicToolCallStatus::Failed
+                },
+                content_items: Some(event.content_items.clone()),
+                success: Some(event.success),
+                error: event.error.clone(),
+                duration: Some(event.duration),
+            }),
+            Some(event.turn_id.clone()),
+        )),
+        _ => None,
+    };
+    Ok(result)
+}

codex-rs/thread-store/src/local/rollout_migration/line_parser.rs

@@ -0,0 +1,197 @@
+//! Makes historical rollout JSON parse like today's protocol types.
+//!
+//! Legacy rollouts span a bunch of old wire shapes: renamed fields, old sandbox policy layouts,
+//! string timestamps, legacy command cwd paths, retired events, and payloads that only deserialize
+//! correctly after going through `serde_json::Value`.
+//!
+//! This module only applies narrowly scoped compatibility rewrites for shapes we know existed in
+//! persisted rollouts. It does not decide turn boundaries, rollback behavior, or what gets written
+//! into paginated history.
+
+use chrono::DateTime;
+use codex_protocol::protocol::RolloutLine;
+use codex_utils_path_uri::LegacyAppPathString;
+use codex_utils_path_uri::PathUri;
+use serde_json::Map;
+use serde_json::Value;
+
+/// Parse one legacy rollout line after applying narrowly scoped compatibility
+/// rewrites for historical wire shapes that current protocol types no longer
+/// accept.
+pub(super) fn parse_legacy_rollout_line(bytes: &[u8]) -> Result<Option<RolloutLine>, String> {
+    if bytes.iter().all(u8::is_ascii_whitespace) {
+        return Ok(None);
+    }
+
+    // Deserializing through Value is intentional. Some historical numeric
+    // payloads fail serde_json's streaming enum path but deserialize correctly
+    // once their tagged object shape has been materialized.
+    let mut value = serde_json::from_slice::<Value>(bytes).map_err(|error| error.to_string())?;
+    if should_skip_retired_record(&value) {
+        return Ok(None);
+    }
+    normalize_legacy_turn_context(&mut value);
+    normalize_legacy_sandbox_policy(&mut value);
+    normalize_legacy_rate_limit_resets(&mut value);
+    normalize_legacy_review_entry(&mut value);
+    normalize_legacy_command_cwd(&mut value)?;
+    serde_json::from_value(value)
+        .map(Some)
+        .map_err(|error| error.to_string())
+}
+
+fn should_skip_retired_record(value: &Value) -> bool {
+    matches!(
+        event_type(value),
+        Some("guardian_assessment" | "thread_name_updated" | "undo_completed")
+    ) || (rollout_type(value) == Some("response_item")
+        && value
+            .get("payload")
+            .and_then(Value::as_object)
+            .and_then(|payload| payload.get("type"))
+            .and_then(Value::as_str)
+            == Some("ghost_snapshot"))
+}
+
+fn normalize_legacy_turn_context(value: &mut Value) {
+    if rollout_type(value) != Some("turn_context") {
+        return;
+    }
+    let Some(payload) = payload_object_mut(value) else {
+        return;
+    };
+    let Some(collaboration_mode) = payload
+        .get_mut("collaboration_mode")
+        .and_then(Value::as_object_mut)
+    else {
+        return;
+    };
+    if collaboration_mode.contains_key("settings") {
+        return;
+    }
+
+    let mut settings = Map::new();
+    for key in ["model", "reasoning_effort", "developer_instructions"] {
+        if let Some(value) = collaboration_mode.get(key) {
+            settings.insert(key.to_string(), value.clone());
+        }
+    }
+    if !settings.is_empty() {
+        collaboration_mode.insert("settings".to_string(), Value::Object(settings));
+    }
+}
+
+fn normalize_legacy_review_entry(value: &mut Value) {
+    if event_type(value) != Some("entered_review_mode") {
+        return;
+    }
+    let Some(payload) = payload_object_mut(value) else {
+        return;
+    };
+    if payload.contains_key("target") {
+        return;
+    }
+    let Some(prompt) = payload.get("prompt").and_then(Value::as_str) else {
+        return;
+    };
+    payload.insert(
+        "target".to_string(),
+        serde_json::json!({
+            "type": "custom",
+            "instructions": prompt,
+        }),
+    );
+}
+
+fn normalize_legacy_rate_limit_resets(value: &mut Value) {
+    if event_type(value) != Some("token_count") {
+        return;
+    }
+    let Some(rate_limits) = payload_object_mut(value)
+        .and_then(|payload| payload.get_mut("rate_limits"))
+        .and_then(Value::as_object_mut)
+    else {
+        return;
+    };
+    for window_name in ["primary", "secondary"] {
+        let Some(window) = rate_limits
+            .get_mut(window_name)
+            .and_then(Value::as_object_mut)
+        else {
+            continue;
+        };
+        let Some(resets_at) = window.get("resets_at").and_then(Value::as_str) else {
+            continue;
+        };
+        let Ok(resets_at) = DateTime::parse_from_rfc3339(resets_at) else {
+            continue;
+        };
+        window.insert("resets_at".to_string(), Value::from(resets_at.timestamp()));
+    }
+}
+
+fn normalize_legacy_sandbox_policy(value: &mut Value) {
+    if rollout_type(value) != Some("turn_context") {
+        return;
+    }
+    let Some(sandbox_policy) = payload_object_mut(value)
+        .and_then(|payload| payload.get_mut("sandbox_policy"))
+        .and_then(Value::as_object_mut)
+    else {
+        return;
+    };
+    if sandbox_policy.contains_key("type") {
+        return;
+    }
+    if let Some(mode) = sandbox_policy.get("mode").cloned() {
+        sandbox_policy.insert("type".to_string(), mode);
+    }
+}
+
+fn normalize_legacy_command_cwd(value: &mut Value) -> Result<(), String> {
+    if !matches!(
+        event_type(value),
+        Some("exec_command_begin" | "exec_command_end")
+    ) {
+        return Ok(());
+    }
+    let Some(payload) = payload_object_mut(value) else {
+        return Ok(());
+    };
+    let Some(cwd) = payload.get("cwd").and_then(Value::as_str) else {
+        return Ok(());
+    };
+    if cwd.starts_with("file:") {
+        return Ok(());
+    }
+    let cwd = PathUri::try_from(LegacyAppPathString::from_string(cwd))
+        .map_err(|error| format!("invalid legacy command cwd: {error}"))?;
+    payload.insert(
+        "cwd".to_string(),
+        serde_json::to_value(cwd).map_err(|error| error.to_string())?,
+    );
+    Ok(())
+}
+
+fn rollout_type(value: &Value) -> Option<&str> {
+    value.get("type").and_then(Value::as_str)
+}
+
+fn event_type(value: &Value) -> Option<&str> {
+    if rollout_type(value) != Some("event_msg") {
+        return None;
+    }
+    value
+        .get("payload")
+        .and_then(Value::as_object)
+        .and_then(|payload| payload.get("type"))
+        .and_then(Value::as_str)
+}
+
+fn payload_object_mut(value: &mut Value) -> Option<&mut Map<String, Value>> {
+    value.get_mut("payload").and_then(Value::as_object_mut)
+}
+
+#[cfg(test)]
+#[path = "line_parser_tests.rs"]
+mod tests;

codex-rs/thread-store/src/local/rollout_migration/line_parser_tests.rs

@@ -0,0 +1,228 @@
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::RolloutItem;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+
+use super::parse_legacy_rollout_line;
+
+fn line(payload_type: &str, payload: serde_json::Value) -> Vec<u8> {
+    serde_json::to_vec(&json!({
+        "timestamp": "2025-01-03T12:00:00Z",
+        "type": payload_type,
+        "payload": payload,
+    }))
+    .expect("serialize fixture")
+}
+
+#[test]
+fn parses_legacy_numeric_event_payloads_through_value() {
+    let bytes = line(
+        "event_msg",
+        json!({
+            "type": "token_count",
+            "info": null,
+            "rate_limits": {
+                "primary": {
+                    "used_percent": 2,
+                    "window_minutes": 300,
+                    "resets_at": 1_770_414_841,
+                },
+                "secondary": {
+                    "used_percent": 12,
+                    "window_minutes": 10_080,
+                    "resets_at": 1_770_698_702,
+                },
+                "credits": {
+                    "has_credits": false,
+                    "unlimited": false,
+                    "balance": null,
+                },
+                "plan_type": null,
+            },
+        }),
+    );
+
+    let parsed = parse_legacy_rollout_line(&bytes)
+        .expect("parse legacy token count")
+        .expect("keep legacy token count");
+    assert!(matches!(
+        parsed.item,
+        RolloutItem::EventMsg(EventMsg::TokenCount(_))
+    ));
+}
+
+#[test]
+fn normalizes_legacy_rate_limit_reset_timestamps() {
+    let bytes = line(
+        "event_msg",
+        json!({
+            "type": "token_count",
+            "info": null,
+            "rate_limits": {
+                "primary": {
+                    "used_percent": 2,
+                    "window_minutes": 300,
+                    "resets_at": "2025-10-19T08:51:37.876641+00:00",
+                },
+                "secondary": null,
+                "credits": null,
+                "plan_type": null,
+            },
+        }),
+    );
+
+    let parsed = parse_legacy_rollout_line(&bytes)
+        .expect("parse legacy reset timestamp")
+        .expect("keep legacy token count");
+    assert!(matches!(
+        parsed.item,
+        RolloutItem::EventMsg(EventMsg::TokenCount(_))
+    ));
+}
+
+#[test]
+fn normalizes_legacy_turn_context_collaboration_mode() {
+    let cwd = std::env::temp_dir().to_string_lossy().into_owned();
+    let bytes = line(
+        "turn_context",
+        json!({
+            "cwd": cwd,
+            "approval_policy": "never",
+            "sandbox_policy": {"type": "danger-full-access"},
+            "model": "gpt-test",
+            "personality": null,
+            "collaboration_mode": {
+                "mode": "plan",
+                "model": "gpt-test",
+                "reasoning_effort": null,
+                "developer_instructions": null,
+            },
+            "effort": null,
+            "summary": "auto",
+        }),
+    );
+
+    let parsed = parse_legacy_rollout_line(&bytes)
+        .expect("parse legacy turn context")
+        .expect("keep legacy turn context");
+    let RolloutItem::TurnContext(context) = parsed.item else {
+        panic!("expected turn context");
+    };
+    assert_eq!(
+        context
+            .collaboration_mode
+            .expect("collaboration mode")
+            .model(),
+        "gpt-test"
+    );
+}
+
+#[test]
+fn normalizes_legacy_turn_context_sandbox_policy() {
+    let cwd = std::env::temp_dir().to_string_lossy().into_owned();
+    let bytes = line(
+        "turn_context",
+        json!({
+            "cwd": cwd,
+            "approval_policy": "never",
+            "sandbox_policy": {"mode": "danger-full-access"},
+            "model": "gpt-test",
+            "personality": null,
+            "effort": null,
+            "summary": "auto",
+        }),
+    );
+
+    let parsed = parse_legacy_rollout_line(&bytes)
+        .expect("parse legacy sandbox policy")
+        .expect("keep legacy turn context");
+    assert!(matches!(parsed.item, RolloutItem::TurnContext(_)));
+}
+
+#[test]
+fn normalizes_legacy_review_entry_prompt() {
+    let bytes = line(
+        "event_msg",
+        json!({
+            "type": "entered_review_mode",
+            "prompt": "review these changes",
+            "user_facing_hint": "Review requested.",
+        }),
+    );
+
+    let parsed = parse_legacy_rollout_line(&bytes)
+        .expect("parse legacy review entry")
+        .expect("keep legacy review entry");
+    assert!(matches!(
+        parsed.item,
+        RolloutItem::EventMsg(EventMsg::EnteredReviewMode(_))
+    ));
+}
+
+#[test]
+fn normalizes_legacy_plain_command_cwd() {
+    let cwd = std::env::temp_dir().to_string_lossy().into_owned();
+    let bytes = line(
+        "event_msg",
+        json!({
+            "type": "exec_command_end",
+            "call_id": "call-1",
+            "turn_id": "turn-1",
+            "command": ["echo", "ok"],
+            "cwd": cwd,
+            "parsed_cmd": [],
+            "source": "agent",
+            "stdout": "",
+            "stderr": "",
+            "aggregated_output": "",
+            "exit_code": 0,
+            "duration": {"secs": 0, "nanos": 0},
+            "formatted_output": "",
+            "status": "completed",
+        }),
+    );
+
+    let parsed = parse_legacy_rollout_line(&bytes)
+        .expect("parse legacy command")
+        .expect("keep legacy command");
+    assert!(matches!(
+        parsed.item,
+        RolloutItem::EventMsg(EventMsg::ExecCommandEnd(_))
+    ));
+}
+
+#[test]
+fn skips_only_known_retired_events() {
+    for event_type in [
+        "guardian_assessment",
+        "thread_name_updated",
+        "undo_completed",
+    ] {
+        let bytes = line("event_msg", json!({"type": event_type}));
+        assert!(
+            parse_legacy_rollout_line(&bytes)
+                .expect("inspect retired event")
+                .is_none()
+        );
+    }
+
+    let unknown = line("event_msg", json!({"type": "unknown_legacy_event"}));
+    assert!(parse_legacy_rollout_line(&unknown).is_err());
+}
+
+#[test]
+fn skips_legacy_ghost_snapshots() {
+    let ghost_snapshot = line(
+        "response_item",
+        json!({
+            "type": "ghost_snapshot",
+            "ghost_commit": {"id": "legacy"},
+        }),
+    );
+
+    assert!(
+        parse_legacy_rollout_line(&ghost_snapshot)
+            .expect("inspect ghost snapshot")
+            .is_none()
+    );
+}

codex-rs/thread-store/src/local/rollout_migration/publish.rs

@@ -0,0 +1,193 @@
+//! Filesystem helpers for safely publishing migrated rollouts.
+//!
+//! This module owns temporary rollout paths, durable `.pending` journals, compressed rollout
+//! staging, cleanup, and parent-directory syncs. The migration orchestrator decides when to call
+//! these helpers; this module keeps the low-level file operations in one place.
+//!
+//! A migration can crash between publishing JSONL and finishing SQLite metadata. The journal is
+//! the durable handoff between those steps, so cleanup must only remove it once the paginated
+//! rollout and SQLite state are both complete.
+
+use std::collections::HashSet;
+use std::fs::Permissions;
+use std::io;
+use std::io::Write;
+use std::path::Path;
+use std::path::PathBuf;
+use std::time::SystemTime;
+
+#[cfg(unix)]
+use std::os::unix::fs::OpenOptionsExt;
+#[cfg(unix)]
+use std::os::unix::fs::PermissionsExt;
+
+use codex_protocol::ThreadId;
+
+use super::migration_error;
+use crate::ThreadStoreResult;
+
+const MIGRATION_JOURNAL_DIRECTORY: &str = "rollout-migrations";
+
+pub(super) fn migration_journal_path(codex_home: &Path, thread_id: ThreadId) -> PathBuf {
+    codex_home
+        .join(MIGRATION_JOURNAL_DIRECTORY)
+        .join(format!("{thread_id}.pending"))
+}
+
+pub(super) async fn pending_migration_thread_ids(
+    codex_home: &Path,
+) -> ThreadStoreResult<HashSet<ThreadId>> {
+    let mut entries = match tokio::fs::read_dir(codex_home.join(MIGRATION_JOURNAL_DIRECTORY)).await
+    {
+        Ok(entries) => entries,
+        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(HashSet::new()),
+        Err(error) => return Err(migration_error(error)),
+    };
+    let mut thread_ids = HashSet::new();
+    while let Some(entry) = entries.next_entry().await.map_err(migration_error)? {
+        if !entry.file_type().await.map_err(migration_error)?.is_file() {
+            continue;
+        }
+        let filename = entry.file_name();
+        let Some(thread_id) = filename
+            .to_str()
+            .and_then(|filename| filename.strip_suffix(".pending"))
+            .and_then(|thread_id| ThreadId::from_string(thread_id).ok())
+        else {
+            continue;
+        };
+        thread_ids.insert(thread_id);
+    }
+    Ok(thread_ids)
+}
+
+pub(super) fn staged_rollout_path(rollout_path: &Path) -> ThreadStoreResult<PathBuf> {
+    staged_path(rollout_path, "paginated")
+}
+
+pub(super) fn decompressed_staged_rollout_path(rollout_path: &Path) -> ThreadStoreResult<PathBuf> {
+    staged_path(rollout_path, "decompressed")
+}
+
+pub(super) fn compressed_staged_rollout_path(rollout_path: &Path) -> ThreadStoreResult<PathBuf> {
+    staged_path(rollout_path, "paginated.zst")
+}
+
+fn staged_path(rollout_path: &Path, suffix: &str) -> ThreadStoreResult<PathBuf> {
+    let filename = rollout_path
+        .file_name()
+        .and_then(|filename| filename.to_str())
+        .ok_or_else(|| migration_error("rollout path has no valid filename"))?;
+    Ok(rollout_path.with_file_name(format!(".{filename}.{suffix}.tmp")))
+}
+
+pub(super) async fn decompress_rollout_to_path(
+    compressed_path: &Path,
+    plain_path: &Path,
+) -> ThreadStoreResult<()> {
+    let compressed_path = compressed_path.to_path_buf();
+    let plain_path = plain_path.to_path_buf();
+    tokio::task::spawn_blocking(move || -> io::Result<()> {
+        let input = std::fs::File::open(compressed_path)?;
+        let mut decoder = zstd::stream::read::Decoder::new(input)?;
+        let mut options = std::fs::OpenOptions::new();
+        options.create(true).truncate(true).write(true);
+        #[cfg(unix)]
+        options.mode(0o600);
+        let mut output = options.open(plain_path)?;
+        #[cfg(unix)]
+        output.set_permissions(Permissions::from_mode(0o600))?;
+        io::copy(&mut decoder, &mut output)?;
+        output.flush()
+    })
+    .await
+    .map_err(migration_error)?
+    .map_err(migration_error)
+}
+
+pub(super) async fn compress_rollout_to_path(
+    plain_path: &Path,
+    compressed_path: &Path,
+    permissions: Permissions,
+    modified_at: Option<SystemTime>,
+) -> ThreadStoreResult<()> {
+    let plain_path = plain_path.to_path_buf();
+    let compressed_path = compressed_path.to_path_buf();
+    tokio::task::spawn_blocking(move || -> io::Result<()> {
+        let mut input = std::fs::File::open(plain_path)?;
+        let output = std::fs::OpenOptions::new()
+            .create(true)
+            .truncate(true)
+            .write(true)
+            .open(&compressed_path)?;
+        let mut encoder = zstd::stream::write::Encoder::new(output, 3)?;
+        io::copy(&mut input, &mut encoder)?;
+        let output = encoder.finish()?;
+        output.set_permissions(permissions)?;
+        if let Some(modified_at) = modified_at {
+            output.set_times(std::fs::FileTimes::new().set_modified(modified_at))?;
+        }
+        output.sync_all()?;
+
+        let input = std::fs::File::open(compressed_path)?;
+        let mut decoder = zstd::stream::read::Decoder::new(input)?;
+        io::copy(&mut decoder, &mut io::sink())?;
+        Ok(())
+    })
+    .await
+    .map_err(migration_error)?
+    .map_err(migration_error)
+}
+
+pub(super) async fn remove_file_if_present(path: &Path) -> ThreadStoreResult<()> {
+    if tokio::fs::try_exists(path).await.map_err(migration_error)? {
+        tokio::fs::remove_file(path)
+            .await
+            .map_err(migration_error)?;
+    }
+    Ok(())
+}
+
+pub(super) async fn write_migration_journal(path: &Path) -> ThreadStoreResult<()> {
+    let parent = path
+        .parent()
+        .ok_or_else(|| migration_error("rollout migration journal has no parent directory"))?;
+    let parent_already_exists = tokio::fs::try_exists(parent)
+        .await
+        .map_err(migration_error)?;
+    tokio::fs::create_dir_all(parent)
+        .await
+        .map_err(migration_error)?;
+    if !parent_already_exists {
+        sync_parent_directory(parent).await?;
+    }
+    let file = tokio::fs::OpenOptions::new()
+        .create(true)
+        .truncate(true)
+        .write(true)
+        .open(path)
+        .await
+        .map_err(migration_error)?;
+    file.sync_all().await.map_err(migration_error)?;
+    drop(file);
+    sync_parent_directory(path).await
+}
+
+pub(super) async fn sync_parent_directory(path: &Path) -> ThreadStoreResult<()> {
+    #[cfg(unix)]
+    {
+        let parent = path
+            .parent()
+            .ok_or_else(|| migration_error("rollout path has no parent directory"))?
+            .to_path_buf();
+        tokio::task::spawn_blocking(move || std::fs::File::open(parent)?.sync_all())
+            .await
+            .map_err(migration_error)?
+            .map_err(migration_error)?;
+    }
+    #[cfg(not(unix))]
+    {
+        let _ = path;
+    }
+    Ok(())
+}

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

@@ -0,0 +1,667 @@
+use std::fs;
+use std::io::Write;
+use std::path::Path;
+use std::path::PathBuf;
+
+#[cfg(unix)]
+use std::os::unix::fs::PermissionsExt;
+
+use codex_protocol::ThreadId;
+use codex_protocol::models::ContentItem;
+use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::AgentMessageEvent;
+use codex_protocol::protocol::EventMsg;
+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::SubAgentSource;
+use codex_protocol::protocol::ThreadHistoryMode;
+use codex_protocol::protocol::UserMessageEvent;
+use codex_rollout::RolloutConfig;
+use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+
+use super::LocalThreadStore;
+use super::RolloutMigrationMode;
+use super::RolloutMigrationOptions;
+use super::RolloutMigrationStatus;
+#[cfg(unix)]
+use super::decompress_rollout_to_path;
+use super::migration_journal_path;
+use super::thread_history;
+use super::write_migration_journal;
+use crate::ListTurnsParams;
+use crate::SortDirection;
+use crate::StoredTurnItemsView;
+use crate::local::test_support::test_config;
+
+const TIMESTAMP: &str = "2025-01-03T12:00:00Z";
+
+fn write_rollout(
+    home: &Path,
+    thread_id: ThreadId,
+    source: SessionSource,
+    items: Vec<RolloutItem>,
+) -> PathBuf {
+    write_rollout_with_fork(home, thread_id, source, /*forked_from_id*/ None, items)
+}
+
+fn write_rollout_with_fork(
+    home: &Path,
+    thread_id: ThreadId,
+    source: SessionSource,
+    forked_from_id: Option<ThreadId>,
+    items: Vec<RolloutItem>,
+) -> PathBuf {
+    let directory = home.join("sessions/2025/01/03");
+    fs::create_dir_all(&directory).expect("create rollout directory");
+    let path = directory.join(format!("rollout-2025-01-03T12-00-00-{thread_id}.jsonl"));
+    let mut file = fs::File::create(&path).expect("create legacy rollout");
+    let metadata = SessionMeta {
+        session_id: thread_id.into(),
+        id: thread_id,
+        forked_from_id,
+        timestamp: TIMESTAMP.to_string(),
+        cwd: home.to_path_buf(),
+        originator: "test-originator".to_string(),
+        cli_version: "0.0.0".to_string(),
+        source,
+        model_provider: Some("test-provider".to_string()),
+        ..SessionMeta::default()
+    };
+    let items = std::iter::once(RolloutItem::SessionMeta(SessionMetaLine {
+        meta: metadata,
+        git: None,
+    }))
+    .chain(items);
+    for item in items {
+        let line = RolloutLine {
+            timestamp: TIMESTAMP.to_string(),
+            ordinal: None,
+            item,
+        };
+        writeln!(
+            file,
+            "{}",
+            serde_json::to_string(&line).expect("serialize legacy record")
+        )
+        .expect("write legacy record");
+    }
+    path
+}
+
+fn user_message(text: &str) -> RolloutItem {
+    RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
+        message: text.to_string(),
+        ..UserMessageEvent::default()
+    }))
+}
+
+fn agent_message(text: &str) -> RolloutItem {
+    RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent {
+        message: text.to_string(),
+        phase: None,
+        memory_citation: None,
+    }))
+}
+
+fn read_rollout(path: &Path) -> Vec<RolloutLine> {
+    fs::read_to_string(path)
+        .expect("read migrated rollout")
+        .lines()
+        .map(|line| serde_json::from_str(line).expect("parse migrated rollout"))
+        .collect()
+}
+
+fn compress_rollout(path: PathBuf) -> PathBuf {
+    let compressed_path = path.with_extension("jsonl.zst");
+    let input = fs::File::open(&path).expect("open rollout for compression");
+    let compressed = zstd::stream::encode_all(input, /*level*/ 3).expect("compress rollout");
+    fs::write(&compressed_path, compressed).expect("write compressed rollout");
+    fs::remove_file(path).expect("remove plain rollout");
+    compressed_path
+}
+
+fn apply_options() -> RolloutMigrationOptions {
+    RolloutMigrationOptions {
+        mode: RolloutMigrationMode::Apply,
+        max_mib_per_second: 1024,
+        ..RolloutMigrationOptions::default()
+    }
+}
+
+async fn indexed_store(home: &Path) -> LocalThreadStore {
+    let config = test_config(home);
+    let rollout_config = RolloutConfig {
+        codex_home: config.codex_home.clone(),
+        sqlite: config.sqlite.clone(),
+        cwd: home.to_path_buf(),
+        model_provider_id: config.default_model_provider_id.clone(),
+        generate_memories: false,
+    };
+    let state_db = codex_rollout::state_db::try_init(&rollout_config)
+        .await
+        .expect("backfill legacy thread metadata");
+    LocalThreadStore::new(config, Some(state_db))
+}
+
+#[tokio::test]
+async fn migration_publishes_canonical_projected_history_and_is_idempotent() {
+    let home = TempDir::new().expect("create Codex home");
+    let thread_id = ThreadId::new();
+    let path = write_rollout(
+        home.path(),
+        thread_id,
+        SessionSource::Cli,
+        vec![
+            user_message("first question"),
+            agent_message("first answer"),
+        ],
+    );
+    let store = indexed_store(home.path()).await;
+
+    let report = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("migrate legacy rollout");
+    assert_eq!(report.outcomes.len(), 1);
+    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
+
+    let lines = read_rollout(&path);
+    assert_eq!(
+        lines.iter().map(|line| line.ordinal).collect::<Vec<_>>(),
+        (0..lines.len() as u64).map(Some).collect::<Vec<_>>()
+    );
+    assert!(matches!(
+        &lines[0].item,
+        RolloutItem::SessionMeta(metadata)
+            if metadata.meta.history_mode == ThreadHistoryMode::Paginated
+                && metadata.meta.id == thread_id
+                && metadata.meta.history_base.is_none()
+    ));
+    assert_eq!(
+        lines
+            .iter()
+            .filter(|line| matches!(line.item, RolloutItem::EventMsg(EventMsg::ItemCompleted(_))))
+            .count(),
+        2
+    );
+
+    let turns = store
+        .list_turns(ListTurnsParams {
+            thread_id,
+            include_archived: false,
+            cursor: None,
+            page_size: 10,
+            sort_direction: SortDirection::Asc,
+            items_view: StoredTurnItemsView::Summary,
+        })
+        .await
+        .expect("read projected turns");
+    assert_eq!(turns.turns.len(), 1);
+    assert_eq!(turns.turns[0].items.len(), 2);
+
+    let bytes = fs::read(&path).expect("read first migration");
+    let second = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("rerun migration");
+    assert_eq!(
+        second.outcomes[0].status,
+        RolloutMigrationStatus::AlreadyPaginated
+    );
+    assert_eq!(fs::read(&path).expect("read idempotent rollout"), bytes);
+}
+
+#[tokio::test]
+async fn migration_preserves_valid_final_record_without_newline() {
+    let home = TempDir::new().expect("create Codex home");
+    let thread_id = ThreadId::new();
+    let path = write_rollout(
+        home.path(),
+        thread_id,
+        SessionSource::Cli,
+        vec![user_message("question")],
+    );
+    let final_line = RolloutLine {
+        timestamp: TIMESTAMP.to_string(),
+        ordinal: None,
+        item: agent_message("answer"),
+    };
+    fs::OpenOptions::new()
+        .append(true)
+        .open(&path)
+        .expect("open legacy rollout")
+        .write_all(
+            serde_json::to_string(&final_line)
+                .expect("serialize final record")
+                .as_bytes(),
+        )
+        .expect("append final record");
+    let store = indexed_store(home.path()).await;
+
+    store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("migrate legacy rollout");
+
+    assert_eq!(
+        read_rollout(&path)
+            .iter()
+            .filter(|line| matches!(line.item, RolloutItem::EventMsg(EventMsg::ItemCompleted(_))))
+            .count(),
+        2
+    );
+}
+
+#[tokio::test]
+async fn migration_preserves_copied_user_fork_history_without_creating_a_history_base() {
+    let home = TempDir::new().expect("create Codex home");
+    let thread_id = ThreadId::new();
+    let parent_id = ThreadId::new();
+    let copied_metadata = SessionMeta {
+        session_id: parent_id.into(),
+        id: parent_id,
+        timestamp: TIMESTAMP.to_string(),
+        cwd: home.path().to_path_buf(),
+        source: SessionSource::Cli,
+        ..SessionMeta::default()
+    };
+    let copied_response = RolloutItem::ResponseItem(ResponseItem::Message {
+        id: None,
+        role: "user".to_string(),
+        content: vec![ContentItem::InputText {
+            text: "copied parent history".to_string(),
+        }],
+        phase: None,
+        internal_chat_message_metadata_passthrough: None,
+    });
+    let path = write_rollout_with_fork(
+        home.path(),
+        thread_id,
+        SessionSource::Cli,
+        Some(parent_id),
+        vec![
+            RolloutItem::SessionMeta(SessionMetaLine {
+                meta: copied_metadata,
+                git: None,
+            }),
+            copied_response,
+            user_message("child question"),
+            agent_message("child answer"),
+        ],
+    );
+    let expected_responses = read_rollout(&path)
+        .into_iter()
+        .filter_map(|line| match line.item {
+            RolloutItem::ResponseItem(item) => {
+                Some(serde_json::to_value(item).expect("serialize copied response"))
+            }
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+    let store = indexed_store(home.path()).await;
+
+    let report = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("migrate copied user fork");
+
+    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
+    let lines = read_rollout(&path);
+    assert!(matches!(
+        &lines[0].item,
+        RolloutItem::SessionMeta(metadata)
+            if metadata.meta.id == thread_id
+                && metadata.meta.forked_from_id == Some(parent_id)
+                && metadata.meta.history_mode == ThreadHistoryMode::Paginated
+                && metadata.meta.history_base.is_none()
+    ));
+    assert!(matches!(
+        &lines[1].item,
+        RolloutItem::SessionMeta(metadata)
+            if metadata.meta.id == parent_id
+                && metadata.meta.history_mode == ThreadHistoryMode::Legacy
+    ));
+    assert_eq!(
+        lines
+            .into_iter()
+            .filter_map(|line| match line.item {
+                RolloutItem::ResponseItem(item) => {
+                    Some(serde_json::to_value(item).expect("serialize migrated response"))
+                }
+                _ => None,
+            })
+            .collect::<Vec<_>>(),
+        expected_responses
+    );
+}
+
+#[tokio::test]
+async fn dry_run_reports_newest_first_and_skips_subagents() {
+    let home = TempDir::new().expect("create Codex home");
+    let root_id = ThreadId::new();
+    let root = write_rollout(
+        home.path(),
+        root_id,
+        SessionSource::Cli,
+        vec![user_message("root question")],
+    );
+    let subagent_id = ThreadId::new();
+    let subagent = write_rollout(
+        home.path(),
+        subagent_id,
+        SessionSource::SubAgent(SubAgentSource::Other("test".to_string())),
+        vec![user_message("subagent question")],
+    );
+    let compressed_id = ThreadId::new();
+    let compressed = compress_rollout(write_rollout(
+        home.path(),
+        compressed_id,
+        SessionSource::Cli,
+        vec![user_message("compressed question")],
+    ));
+    let original = fs::read(&root).expect("read original root rollout");
+    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+
+    let report = store
+        .migrate_rollouts(RolloutMigrationOptions::default())
+        .await
+        .expect("inspect legacy rollouts");
+
+    let mut expected = vec![
+        (root.clone(), root_id, RolloutMigrationStatus::Eligible),
+        (
+            subagent,
+            subagent_id,
+            RolloutMigrationStatus::SkippedSubagent,
+        ),
+        (compressed, compressed_id, RolloutMigrationStatus::Eligible),
+    ];
+    expected.sort_by(|(left, ..), (right, ..)| right.cmp(left));
+    assert_eq!(
+        report
+            .outcomes
+            .iter()
+            .map(|outcome| (
+                outcome.rollout_path.clone(),
+                outcome.thread_id.expect("rollout thread ID"),
+                outcome.status,
+            ))
+            .collect::<Vec<_>>(),
+        expected,
+    );
+    assert_eq!(fs::read(&root).expect("read inspected rollout"), original);
+}
+
+#[tokio::test]
+async fn migration_preserves_compressed_rollouts_during_publish_and_recovery() {
+    let home = TempDir::new().expect("create Codex home");
+    let thread_id = ThreadId::new();
+    let compressed_path = compress_rollout(write_rollout(
+        home.path(),
+        thread_id,
+        SessionSource::Cli,
+        vec![user_message("compressed question")],
+    ));
+    let plain_path = compressed_path.with_extension("");
+    let store = indexed_store(home.path()).await;
+
+    let report = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("migrate compressed rollout");
+    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
+    assert!(!plain_path.exists());
+    assert_eq!(
+        codex_rollout::read_session_meta_line(&compressed_path)
+            .await
+            .expect("read compressed metadata")
+            .meta
+            .history_mode,
+        ThreadHistoryMode::Paginated
+    );
+
+    thread_history::delete_thread(&store, thread_id)
+        .await
+        .expect("simulate missing projection");
+    let journal_path = migration_journal_path(home.path(), thread_id);
+    write_migration_journal(&journal_path)
+        .await
+        .expect("simulate pending migration journal");
+    let recovered = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("recover compressed rollout");
+
+    assert_eq!(
+        recovered.outcomes[0].status,
+        RolloutMigrationStatus::Migrated
+    );
+    assert!(recovered.outcomes[0].bytes_processed > 0);
+    assert!(compressed_path.exists());
+    assert!(!plain_path.exists());
+    assert!(!journal_path.exists());
+}
+
+#[cfg(unix)]
+#[tokio::test]
+async fn decompression_temporaries_are_owner_only() {
+    let home = TempDir::new().expect("create Codex home");
+    let thread_id = ThreadId::new();
+    let compressed_path = compress_rollout(write_rollout(
+        home.path(),
+        thread_id,
+        SessionSource::Cli,
+        vec![user_message("compressed question")],
+    ));
+    let plain_path = home.path().join("decompressed.tmp");
+
+    decompress_rollout_to_path(&compressed_path, &plain_path)
+        .await
+        .expect("decompress rollout");
+
+    assert_eq!(
+        fs::metadata(&plain_path)
+            .expect("read temporary metadata")
+            .permissions()
+            .mode()
+            & 0o777,
+        0o600
+    );
+}
+
+#[tokio::test]
+async fn migration_skips_threads_with_an_active_writer() {
+    let home = TempDir::new().expect("create Codex home");
+    let thread_id = ThreadId::new();
+    let path = write_rollout(
+        home.path(),
+        thread_id,
+        SessionSource::Cli,
+        vec![user_message("active question")],
+    );
+    let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+    let _writer = store
+        .writer_lock_coordinator
+        .acquire(thread_id)
+        .expect("acquire live writer lock");
+    let original = fs::read(&path).expect("read active rollout");
+
+    let report = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("inspect active writer");
+
+    assert_eq!(
+        report.outcomes[0].status,
+        RolloutMigrationStatus::SkippedBusy
+    );
+    assert_eq!(fs::read(&path).expect("read unmodified rollout"), original);
+}
+
+#[tokio::test]
+async fn migration_recovers_a_published_rollout_with_missing_projection() {
+    let home = TempDir::new().expect("create Codex home");
+    let thread_id = ThreadId::new();
+    let path = write_rollout(
+        home.path(),
+        thread_id,
+        SessionSource::Cli,
+        vec![
+            user_message("recover question"),
+            agent_message("recover answer"),
+        ],
+    );
+    let store = indexed_store(home.path()).await;
+    store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("publish canonical rollout");
+    thread_history::delete_thread(&store, thread_id)
+        .await
+        .expect("simulate interrupted projection");
+    let journal_path = migration_journal_path(home.path(), thread_id);
+    write_migration_journal(&journal_path)
+        .await
+        .expect("simulate pending migration journal");
+
+    let writer = store
+        .writer_lock_coordinator
+        .acquire(thread_id)
+        .expect("acquire live writer lock");
+    let busy = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("inspect busy published recovery");
+    assert_eq!(busy.outcomes[0].status, RolloutMigrationStatus::SkippedBusy);
+    assert!(journal_path.exists());
+    drop(writer);
+
+    let report = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("recover published rollout");
+
+    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Migrated);
+    assert!(!journal_path.exists());
+    let projection = thread_history::projection_state(&store, thread_id)
+        .await
+        .expect("read repaired projection")
+        .expect("projection was rebuilt");
+    assert_eq!(
+        projection.next_byte_offset,
+        fs::metadata(&path).expect("read rollout metadata").len()
+    );
+}
+
+#[tokio::test]
+async fn migration_recovers_pending_rollouts_before_new_work() {
+    let home = TempDir::new().expect("create Codex home");
+    let pending_thread_id = ThreadId::new();
+    write_rollout(
+        home.path(),
+        pending_thread_id,
+        SessionSource::Cli,
+        vec![user_message("pending question")],
+    );
+    let new_thread_id = ThreadId::new();
+    let new_path = write_rollout(
+        home.path(),
+        new_thread_id,
+        SessionSource::Cli,
+        vec![user_message("new question")],
+    );
+    let newer_directory = home.path().join("sessions/2025/01/04");
+    fs::create_dir_all(&newer_directory).expect("create newer rollout directory");
+    fs::rename(
+        &new_path,
+        newer_directory.join(new_path.file_name().expect("rollout filename")),
+    )
+    .expect("move newer rollout");
+    let store = indexed_store(home.path()).await;
+
+    store
+        .migrate_rollouts(RolloutMigrationOptions {
+            thread_ids: vec![pending_thread_id],
+            ..apply_options()
+        })
+        .await
+        .expect("publish pending rollout");
+    thread_history::delete_thread(&store, pending_thread_id)
+        .await
+        .expect("simulate missing projection");
+    write_migration_journal(&migration_journal_path(home.path(), pending_thread_id))
+        .await
+        .expect("simulate pending migration journal");
+
+    let report = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("recover pending rollout before new work");
+
+    assert_eq!(
+        report
+            .outcomes
+            .iter()
+            .map(|outcome| (
+                outcome.thread_id.expect("rollout thread ID"),
+                outcome.status
+            ))
+            .collect::<Vec<_>>(),
+        vec![
+            (pending_thread_id, RolloutMigrationStatus::Migrated),
+            (new_thread_id, RolloutMigrationStatus::Migrated),
+        ]
+    );
+}
+
+#[tokio::test]
+async fn failed_migration_preserves_the_legacy_rollout_and_can_be_retried() {
+    let home = TempDir::new().expect("create Codex home");
+    let thread_id = ThreadId::new();
+    let path = write_rollout(
+        home.path(),
+        thread_id,
+        SessionSource::Cli,
+        vec![user_message("recoverable question")],
+    );
+    let valid_length = fs::metadata(&path)
+        .expect("read valid rollout length")
+        .len();
+    let mut file = fs::OpenOptions::new()
+        .append(true)
+        .open(&path)
+        .expect("open legacy rollout");
+    writeln!(file, "{{not valid rollout json").expect("append malformed record");
+    drop(file);
+    let original = fs::read(&path).expect("read malformed legacy rollout");
+    let store = indexed_store(home.path()).await;
+
+    let report = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("report malformed legacy rollout");
+
+    assert_eq!(report.outcomes[0].status, RolloutMigrationStatus::Failed);
+    assert_eq!(
+        fs::read(&path).expect("read preserved legacy rollout"),
+        original
+    );
+
+    fs::OpenOptions::new()
+        .write(true)
+        .open(&path)
+        .expect("open repaired legacy rollout")
+        .set_len(valid_length)
+        .expect("remove malformed tail");
+    let retry = store
+        .migrate_rollouts(apply_options())
+        .await
+        .expect("retry repaired legacy rollout");
+
+    assert_eq!(retry.outcomes[0].status, RolloutMigrationStatus::Migrated);
+    assert!(!migration_journal_path(home.path(), thread_id).exists());
+}