From 6bb6e9045f728f172ee1c68e6dc56ce6a53cbdb8 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Wed, 5 Aug 2026 22:52:25 +0000 Subject: [PATCH] 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 | 1 + codex-rs/app-server/tests/suite/v2/mod.rs | 1 + .../tests/suite/v2/rollout_migration.rs | 132 +++ codex-rs/rollout/src/compression.rs | 10 + codex-rs/rollout/src/compression_tests.rs | 22 + codex-rs/rollout/src/lib.rs | 3 + codex-rs/rollout/src/maintenance.rs | 41 + codex-rs/state/src/runtime/threads.rs | 42 +- codex-rs/thread-store/Cargo.toml | 3 +- codex-rs/thread-store/src/lib.rs | 5 + codex-rs/thread-store/src/local/mod.rs | 7 + .../src/local/rollout_migration.rs | 812 ++++++++++++++++++ .../local/rollout_migration/canonicalizer.rs | 505 +++++++++++ .../local/rollout_migration/legacy_event.rs | 251 ++++++ .../local/rollout_migration/line_parser.rs | 197 +++++ .../rollout_migration/line_parser_tests.rs | 228 +++++ .../src/local/rollout_migration/publish.rs | 193 +++++ .../src/local/rollout_migration_tests.rs | 667 ++++++++++++++ 18 files changed, 3115 insertions(+), 5 deletions(-) create mode 100644 codex-rs/app-server/tests/suite/v2/rollout_migration.rs create mode 100644 codex-rs/rollout/src/maintenance.rs create mode 100644 codex-rs/thread-store/src/local/rollout_migration.rs create mode 100644 codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs create mode 100644 codex-rs/thread-store/src/local/rollout_migration/legacy_event.rs create mode 100644 codex-rs/thread-store/src/local/rollout_migration/line_parser.rs create mode 100644 codex-rs/thread-store/src/local/rollout_migration/line_parser_tests.rs create mode 100644 codex-rs/thread-store/src/local/rollout_migration/publish.rs create mode 100644 codex-rs/thread-store/src/local/rollout_migration_tests.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index fe65fb41b6a8..70880f31a721 100644 --- a/codex-rs/Cargo.lock +++ b/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", diff --git a/codex-rs/app-server/tests/suite/v2/mod.rs b/codex-rs/app-server/tests/suite/v2/mod.rs index 4ae553728784..b34d3dd5cad7 100644 --- a/codex-rs/app-server/tests/suite/v2/mod.rs +++ b/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; diff --git a/codex-rs/app-server/tests/suite/v2/rollout_migration.rs b/codex-rs/app-server/tests/suite/v2/rollout_migration.rs new file mode 100644 index 000000000000..59d182ffded4 --- /dev/null +++ b/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(()) +} diff --git a/codex-rs/rollout/src/compression.rs b/codex-rs/rollout/src/compression.rs index ef3589da2270..8b6e0de3eda5 100644 --- a/codex-rs/rollout/src/compression.rs +++ b/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) => { diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index 4a9f879949aa..be0aa7895c9a 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/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()?; diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 89a81e4c1954..3618ecf8ac1d 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/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; diff --git a/codex-rs/rollout/src/maintenance.rs b/codex-rs/rollout/src/maintenance.rs new file mode 100644 index 000000000000..35b792874096 --- /dev/null +++ b/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> { + 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), + } +} diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 100c2f8ff181..1537a5d991a3 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/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 { + 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> { 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] diff --git a/codex-rs/thread-store/Cargo.toml b/codex-rs/thread-store/Cargo.toml index ae4626852d69..a30cecdc309d 100644 --- a/codex-rs/thread-store/Cargo.toml +++ b/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 } diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 817523bce873..3d5eecb62f78 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/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; diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 23f00a205bdd..cb8e8d969985 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/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 diff --git a/codex-rs/thread-store/src/local/rollout_migration.rs b/codex-rs/thread-store/src/local/rollout_migration.rs new file mode 100644 index 000000000000..e08c2461df80 --- /dev/null +++ b/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, + 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, + pub rollout_path: PathBuf, + pub status: RolloutMigrationStatus, + pub bytes_processed: u64, + pub message: Option, +} + +/// The complete result of scanning active rollout files. +#[derive(Clone, Debug, Default, Eq, PartialEq, Serialize)] +pub struct RolloutMigrationReport { + pub outcomes: Vec, +} + +struct RolloutMigrationRateLimiter { + started_at: Instant, + bytes_processed: u64, + bytes_per_second: u64, + bytes_since_yield: u64, +} + +struct RolloutRecord { + line: Option, + byte_count: u64, +} + +impl RolloutMigrationRateLimiter { + fn new(max_mib_per_second: u64) -> ThreadStoreResult { + 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 { + 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, + bytes: &mut Vec, +) -> ThreadStoreResult> { + 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> { + 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) -> bool { + selected.is_empty() || actual.is_some_and(|thread_id| selected.contains(&thread_id)) +} + +fn thread_id_from_rollout_filename(path: &Path) -> Option { + 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, + 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; diff --git a/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs b/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs new file mode 100644 index 000000000000..d1519fe697f5 --- /dev/null +++ b/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, + known_turn_ids: HashSet, + reasoning: Option, + 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, + ) -> ThreadStoreResult { + 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, ×tamp, RolloutItem::SessionMeta(metadata)) + .await?; + Ok(self.bytes_written - bytes_before) + } + + pub(super) async fn process_line( + &mut self, + line: RolloutLine, + writer: &mut BufWriter, + ) -> ThreadStoreResult { + 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, ×tamp, 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, ×tamp, RolloutItem::ResponseItem(response)) + .await?; + if let Some(hook) = hook { + self.ensure_turn(writer, ×tamp, source_index).await?; + self.reasoning = None; + self.write_completed_item(writer, ×tamp, 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, ×tamp).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, + ×tamp, + 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, + ×tamp, + 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, + ×tamp, + 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, ×tamp).await?; + } + self.ensure_turn(writer, ×tamp, 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, ×tamp, item).await?; + } + RolloutItem::EventMsg(EventMsg::AgentReasoning(event)) => { + self.write_reasoning( + writer, + ×tamp, + source_index, + event.text, + ReasoningTextKind::Summary, + ) + .await?; + } + RolloutItem::EventMsg(EventMsg::AgentReasoningRawContent(event)) => { + self.write_reasoning( + writer, + ×tamp, + 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, + ×tamp, + 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, ×tamp, 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, ×tamp, turn_id, item, + ) + .await?; + } else if self.active_turn.is_none() { + self.start_implicit_turn(writer, ×tamp, turn_id) + .await?; + self.reasoning = None; + self.write_completed_item(writer, ×tamp, item).await?; + } else { + self.reasoning = None; + self.write_completed_item(writer, ×tamp, item).await?; + } + } + None => { + self.ensure_turn(writer, ×tamp, source_index).await?; + self.reasoning = None; + self.write_completed_item(writer, ×tamp, item).await?; + } + } + } else { + let item = RolloutItem::EventMsg(event); + if codex_rollout::is_persisted_rollout_item(&item, ThreadHistoryMode::Paginated) + { + self.write_item(writer, ×tamp, item).await?; + } + } + } + item @ RolloutItem::InterAgentCommunication(_) => { + self.write_item(writer, ×tamp, item).await?; + } + item @ RolloutItem::Compacted(_) => { + self.write_item(writer, ×tamp, item).await?; + } + item @ (RolloutItem::InterAgentCommunicationMetadata { .. } + | RolloutItem::TurnContext(_) + | RolloutItem::WorldState(_)) => { + self.write_item(writer, ×tamp, item).await?; + } + } + + Ok(self.bytes_written - bytes_before) + } + + pub(super) async fn finish( + &mut self, + writer: &mut BufWriter, + timestamp: &str, + ) -> ThreadStoreResult { + 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, + 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, + 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, + 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, + 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, + 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, + 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, + 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 { + 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) + } +} diff --git a/codex-rs/thread-store/src/local/rollout_migration/legacy_event.rs b/codex-rs/thread-store/src/local/rollout_migration/legacy_event.rs new file mode 100644 index 000000000000..561f679df429 --- /dev/null +++ b/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, +) -> ThreadStoreResult { + 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, +) -> ThreadStoreResult)>> { + 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) +} diff --git a/codex-rs/thread-store/src/local/rollout_migration/line_parser.rs b/codex-rs/thread-store/src/local/rollout_migration/line_parser.rs new file mode 100644 index 000000000000..872e2a3e7eac --- /dev/null +++ b/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, 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::(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> { + value.get_mut("payload").and_then(Value::as_object_mut) +} + +#[cfg(test)] +#[path = "line_parser_tests.rs"] +mod tests; diff --git a/codex-rs/thread-store/src/local/rollout_migration/line_parser_tests.rs b/codex-rs/thread-store/src/local/rollout_migration/line_parser_tests.rs new file mode 100644 index 000000000000..ea1d31cbba13 --- /dev/null +++ b/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 { + 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() + ); +} diff --git a/codex-rs/thread-store/src/local/rollout_migration/publish.rs b/codex-rs/thread-store/src/local/rollout_migration/publish.rs new file mode 100644 index 000000000000..21f142e3c290 --- /dev/null +++ b/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> { + 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 { + staged_path(rollout_path, "paginated") +} + +pub(super) fn decompressed_staged_rollout_path(rollout_path: &Path) -> ThreadStoreResult { + staged_path(rollout_path, "decompressed") +} + +pub(super) fn compressed_staged_rollout_path(rollout_path: &Path) -> ThreadStoreResult { + staged_path(rollout_path, "paginated.zst") +} + +fn staged_path(rollout_path: &Path, suffix: &str) -> ThreadStoreResult { + 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, +) -> 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(()) +} diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs new file mode 100644 index 000000000000..dcdae5750ed0 --- /dev/null +++ b/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, +) -> 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, + items: Vec, +) -> 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 { + 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::>(), + (0..lines.len() as u64).map(Some).collect::>() + ); + 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::>(); + 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::>(), + 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::>(), + 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![ + (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()); +}