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
This commit is contained in:
Owen Lin authored and copyberry committed 2026-08-18 20:07:26 +00:00
1 parent 4a3e829c56
commit f950a1ba0b
3 files changed
+231 -22

No files matched your search

@@ -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<ThreadId, String>,
limiter: &mut RolloutMigrationRateLimiter,
) -> ThreadStoreResult<Option<RolloutMigrationOutcome>> {
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<ThreadId, String>,
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<ThreadId, String>,
limiter: &mut RolloutMigrationRateLimiter,
) -> ThreadStoreResult<PathBuf> {
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<ThreadId, String>,
) -> 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<ThreadId, String>,
) -> 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,
@@ -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() {