Add legacy rollout migration to paginated history

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,11 +32,11 @@ sqlx = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+zstd = { workspace = true }
 
 [dev-dependencies]
 codex-utils-absolute-path = { workspace = true }
 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