Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 26 additions & 7 deletions codex-rs/state/src/runtime/threads.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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<bool> {
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<bool> {
// 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<Option<String>> {
Expand Down Expand Up @@ -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")
);
Expand Down
85 changes: 70 additions & 15 deletions codex-rs/thread-store/src/local/rollout_migration.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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 {
Expand All @@ -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 {
Expand Down Expand Up @@ -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) => {
Expand All @@ -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);
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
}

Expand Down Expand Up @@ -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)?;
Expand Down Expand Up @@ -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())
}
Expand Down Expand Up @@ -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,
Expand Down
135 changes: 135 additions & 0 deletions codex-rs/thread-store/src/local/rollout_migration_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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";
Expand Down Expand Up @@ -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() {
Expand Down
Loading