From f950a1ba0b12884605d98a48fbedaeccff29dd0e Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Tue, 18 Aug 2026 19:57:47 +0000 Subject: [PATCH] Preserve thread names during rollout migration (#39273) ## Why Legacy threads and paginated threads read their displayed names from different metadata. Moving a rollout to paginated history must carry over its legacy-visible name so renamed threads do not lose it. ## What changed - Load legacy indexed names once for an apply run. - When promoting a thread, preserve an existing canonical name or copy its distinct legacy title, falling back to the legacy name index. - Repair missing names when migration encounters an already-paginated rollout. ## Testing Added migration tests covering preservation from both legacy name sources and repair on a rerun. GitOrigin-RevId: 5fc526570b1cf358a159ef174c159ac091ccc40f --- codex-rs/state/src/runtime/threads.rs | 33 ++++- .../src/local/rollout_migration.rs | 85 +++++++++-- .../src/local/rollout_migration_tests.rs | 135 ++++++++++++++++++ 3 files changed, 231 insertions(+), 22 deletions(-) diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index f7eac50676..035f76ba7c 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -61,12 +61,31 @@ 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?; + /// Permanently promote a thread, preserving its canonical name or a legacy-visible fallback. + pub async fn mark_thread_paginated( + &self, + thread_id: ThreadId, + legacy_name: Option<&str>, + ) -> anyhow::Result { + // Legacy threads display `title`, then fall back to the name index. Paginated threads + // display `name`; `title` remains derived metadata used for search. Preserve an existing + // `name`, otherwise carry over the legacy display name. + let result = sqlx::query( + r#" +UPDATE threads +SET + history_mode = 'paginated', + name = CASE + WHEN name IS NULL OR trim(name) = '' THEN ? + ELSE name + END +WHERE id = ? + "#, + ) + .bind(legacy_name) + .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> { @@ -1594,7 +1613,7 @@ mod tests { assert!( runtime - .mark_thread_paginated(thread_id) + .mark_thread_paginated(thread_id, /*legacy_name*/ None) .await .expect("mark paginated history") ); diff --git a/codex-rs/thread-store/src/local/rollout_migration.rs b/codex-rs/thread-store/src/local/rollout_migration.rs index aa70a0d6d8..666e626f77 100644 --- a/codex-rs/thread-store/src/local/rollout_migration.rs +++ b/codex-rs/thread-store/src/local/rollout_migration.rs @@ -8,6 +8,7 @@ //! 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::collections::HashMap; use std::path::Path; use std::path::PathBuf; use std::time::Duration; @@ -32,6 +33,7 @@ use tokio::io::BufWriter; use tokio::time::Instant; use super::LocalThreadStore; +use super::helpers::distinct_thread_metadata_title; use super::thread_history; use super::thread_history::ProjectedRolloutLine; use super::thread_history::RolloutProjectionStep; @@ -297,12 +299,24 @@ impl LocalThreadStore { .is_some_and(|thread_id| pending_thread_ids.contains(&thread_id)) }); } + // The name index is global, so read it once before either migrating or repairing names. + let legacy_names = if options.mode == RolloutMigrationMode::Apply { + let thread_ids = paths + .iter() + .filter_map(|path| thread_id_from_rollout_filename(path)) + .collect(); + codex_rollout::find_thread_names_by_ids(&self.config.codex_home, &thread_ids) + .await + .unwrap_or_default() + } else { + HashMap::new() + }; let total_paths = paths.len(); let mut report = RolloutMigrationReport::default(); for (index, path) in paths.into_iter().enumerate() { let outcome = self - .migrate_rollout_path(path, &options, &mut limiter) + .migrate_rollout_path(path, &options, &legacy_names, &mut limiter) .await?; let outcome_status = outcome.as_ref().map(|outcome| outcome.status); if let Some(outcome) = outcome { @@ -322,6 +336,7 @@ impl LocalThreadStore { &self, mut path: PathBuf, options: &RolloutMigrationOptions, + legacy_names: &HashMap, limiter: &mut RolloutMigrationRateLimiter, ) -> ThreadStoreResult> { let metadata = match codex_rollout::read_session_meta_line(&path).await { @@ -380,7 +395,13 @@ impl LocalThreadStore { let result = if pending_published_migration { let _live_writer_guard = self.live_writer_locks.lock(thread_id).await; match self - .recover_published_migration(thread_id, &path, &journal_path, limiter) + .recover_published_migration( + thread_id, + &path, + &journal_path, + legacy_names, + limiter, + ) .await { Ok(recovered_path) => { @@ -399,6 +420,9 @@ impl LocalThreadStore { Err(error) => Err(error), } } else { + if options.mode == RolloutMigrationMode::Apply { + self.promote_legacy_name(thread_id, legacy_names).await?; + } Ok(RolloutMigrationStatus::AlreadyPaginated) }; let bytes_processed = limiter.bytes_processed.saturating_sub(bytes_before); @@ -438,7 +462,7 @@ impl LocalThreadStore { }; let bytes_before = limiter.bytes_processed; let result = match self - .migrate_one_rollout(thread_id, &path, &journal_path, kind, limiter) + .migrate_one_rollout(thread_id, &path, &journal_path, kind, legacy_names, limiter) .await { Ok(()) => Ok(RolloutMigrationStatus::Migrated), @@ -470,6 +494,7 @@ impl LocalThreadStore { rollout_path: &Path, journal_path: &Path, kind: RolloutMigrationKind, + legacy_names: &HashMap, limiter: &mut RolloutMigrationRateLimiter, ) -> ThreadStoreResult<()> { if let Some(state_db) = &self.state_db @@ -627,7 +652,7 @@ impl LocalThreadStore { .map_err(migration_error)?; } sync_parent_directory(rollout_path).await?; - self.finish_published_migration(thread_id, journal_path) + self.finish_published_migration(thread_id, journal_path, legacy_names) .await } @@ -798,6 +823,7 @@ impl LocalThreadStore { thread_id: ThreadId, rollout_path: &Path, journal_path: &Path, + legacy_names: &HashMap, limiter: &mut RolloutMigrationRateLimiter, ) -> ThreadStoreResult { let _writer_guard = self.writer_lock_coordinator.acquire(thread_id)?; @@ -838,7 +864,7 @@ impl LocalThreadStore { if let Some(decompressed_path) = decompressed_path.as_ref() { remove_file_if_present(decompressed_path).await?; } - self.finish_published_migration(thread_id, journal_path) + self.finish_published_migration(thread_id, journal_path, legacy_names) .await?; Ok(rollout_path.to_path_buf()) } @@ -870,23 +896,52 @@ impl LocalThreadStore { &self, thread_id: ThreadId, journal_path: &Path, + legacy_names: &HashMap, ) -> 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" - ))); - } + self.promote_legacy_name(thread_id, legacy_names).await?; tokio::fs::remove_file(journal_path) .await .map_err(migration_error)?; sync_parent_directory(journal_path).await } + async fn promote_legacy_name( + &self, + thread_id: ThreadId, + legacy_names: &HashMap, + ) -> ThreadStoreResult<()> { + if let Some(state_db) = &self.state_db { + let metadata = state_db + .get_thread(thread_id) + .await + .map_err(migration_error)? + .ok_or_else(|| { + migration_error(format!("thread {thread_id} is missing its SQLite metadata")) + })?; + if metadata.history_mode == ThreadHistoryMode::Paginated + && metadata + .name + .as_deref() + .is_some_and(|name| !name.trim().is_empty()) + { + return Ok(()); + } + let legacy_name = distinct_thread_metadata_title(&metadata) + .or_else(|| legacy_names.get(&thread_id).cloned()) + .filter(|name| !name.trim().is_empty()); + if !state_db + .mark_thread_paginated(thread_id, legacy_name.as_deref()) + .await + .map_err(migration_error)? + { + return Err(migration_error(format!( + "thread {thread_id} is missing its SQLite metadata" + ))); + } + } + Ok(()) + } + async fn project_rollout_in_batches( &self, thread_id: ThreadId, diff --git a/codex-rs/thread-store/src/local/rollout_migration_tests.rs b/codex-rs/thread-store/src/local/rollout_migration_tests.rs index b188b35247..2c7710a7b5 100644 --- a/codex-rs/thread-store/src/local/rollout_migration_tests.rs +++ b/codex-rs/thread-store/src/local/rollout_migration_tests.rs @@ -56,13 +56,17 @@ use super::thread_history; use super::write_migration_journal; use crate::ItemSortKey; use crate::ListItemsParams; +use crate::ListThreadsParams; use crate::ListTurnsParams; use crate::LoadThreadHistoryParams; use crate::SortDirection; use crate::StoredTurnItemsView; +use crate::ThreadMetadataPatch; +use crate::ThreadSortKey; use crate::ThreadStore; use crate::ThreadStoreError; use crate::TurnPage; +use crate::UpdateThreadMetadataParams; use crate::local::test_support::test_config; const TIMESTAMP: &str = "2025-01-03T12:00:00Z"; @@ -1766,6 +1770,137 @@ async fn migration_migrates_archived_rollouts_without_unarchiving_them() { assert_eq!(turns.turns[0].items.len(), 2); } +#[tokio::test] +async fn migration_preserves_legacy_displayed_thread_names() { + let home = TempDir::new().expect("create Codex home"); + let title_thread_id = ThreadId::new(); + write_rollout( + home.path(), + title_thread_id, + SessionSource::Cli, + vec![user_message("title question")], + ); + let index_thread_id = ThreadId::new(); + write_rollout( + home.path(), + index_thread_id, + SessionSource::Cli, + vec![user_message("index question")], + ); + let store = indexed_store(home.path()).await; + store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id: title_thread_id, + patch: ThreadMetadataPatch { + name: Some(Some("renamed title".to_string())), + ..Default::default() + }, + include_archived: false, + }) + .await + .expect("rename legacy thread"); + codex_rollout::append_thread_name(home.path(), title_thread_id, "stale index title") + .await + .expect("write stale legacy index name"); + codex_rollout::append_thread_name(home.path(), index_thread_id, "indexed title") + .await + .expect("write legacy index name"); + + store + .migrate_rollouts(apply_options()) + .await + .expect("migrate named rollouts"); + + let page = store + .list_threads(ListThreadsParams { + page_size: 10, + cursor: None, + sort_key: ThreadSortKey::CreatedAt, + sort_direction: SortDirection::Desc, + allowed_sources: Vec::new(), + model_providers: None, + cwd_filters: None, + section: None, + project_id: None, + archived: false, + search_term: None, + relation_filter: None, + use_state_db_only: true, + }) + .await + .expect("list migrated threads"); + let title_thread = page + .items + .iter() + .find(|thread| thread.thread_id == title_thread_id) + .expect("renamed title thread"); + let index_thread = page + .items + .iter() + .find(|thread| thread.thread_id == index_thread_id) + .expect("indexed title thread"); + + assert_eq!(title_thread.name.as_deref(), Some("renamed title")); + assert_eq!(index_thread.name.as_deref(), Some("indexed title")); +} + +#[tokio::test] +async fn migration_repairs_a_missing_paginated_name_when_rerun() { + let home = TempDir::new().expect("create Codex home"); + let thread_id = ThreadId::new(); + write_rollout( + home.path(), + thread_id, + SessionSource::Cli, + vec![user_message("question")], + ); + let store = indexed_store(home.path()).await; + store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + name: Some(Some("renamed title".to_string())), + ..Default::default() + }, + include_archived: false, + }) + .await + .expect("rename legacy thread"); + store + .migrate_rollouts(apply_options()) + .await + .expect("migrate named rollout"); + let state_db = store.state_db().await.expect("state runtime"); + state_db + .update_thread_title(thread_id, "question") + .await + .expect("restore derived title"); + state_db + .update_thread_name(thread_id, /*name*/ None) + .await + .expect("clear migrated name"); + + let report = store + .migrate_rollouts(apply_options()) + .await + .expect("repair migrated name"); + + assert_eq!( + report.outcomes[0].status, + RolloutMigrationStatus::AlreadyPaginated + ); + assert_eq!( + state_db + .get_thread(thread_id) + .await + .expect("read repaired metadata") + .expect("repaired thread") + .name + .as_deref(), + Some("renamed title") + ); +} + #[cfg(unix)] #[tokio::test] async fn decompression_temporaries_are_owner_only() {