diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 80675f30a00e..509ff342270b 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -3959,9 +3959,12 @@ impl ThreadRequestProcessor { .await && let Some(title) = stored_thread.name.as_deref().map(str::trim) && !title.is_empty() - && stored_thread.preview.trim() != title { - set_thread_name_from_title(thread, title.to_string()); + if stored_thread.history_mode == ThreadHistoryMode::Paginated { + thread.name = Some(title.to_string()); + } else { + set_thread_name_from_title(thread, title.to_string()); + } } } diff --git a/codex-rs/app-server/tests/suite/v2/thread_read.rs b/codex-rs/app-server/tests/suite/v2/thread_read.rs index 78de837e7612..14efbd72f0c3 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -1220,18 +1220,16 @@ async fn thread_read_loaded_thread_returns_precomputed_path_before_materializati } #[tokio::test] -async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> { +async fn paginated_thread_name_set_is_reflected_in_read_list_and_metadata_resume() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; - let preview = "Saved user message"; - let conversation_id = create_fake_rollout_with_text_elements( + let conversation_id = create_fake_paginated_rollout( codex_home.path(), "2025-01-05T12-00-00", "2025-01-05T12:00:00Z", - preview, - vec![], + "Saved user message", Some("mock_provider"), /*git_info*/ None, )?; @@ -1244,7 +1242,7 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> { timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; // Set a user-facing thread title. - let new_name = "My renamed thread"; + let new_name = "Saved user message"; let set_id = mcp .send_thread_set_name_request(ThreadSetNameParams { thread_id: conversation_id.clone(), @@ -1283,6 +1281,7 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> { let ThreadReadResponse { thread, .. } = to_response::(read_resp)?; assert_eq!(thread.id, conversation_id); assert_eq!(thread.name.as_deref(), Some(new_name)); + assert_eq!(thread.history_mode, ThreadHistoryMode::Paginated); let thread_json = read_result .get("thread") .and_then(Value::as_object) @@ -1309,7 +1308,7 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> { source_kinds: None, archived: None, cwd: None, - use_state_db_only: false, + use_state_db_only: true, search_term: None, parent_thread_id: None, ancestor_thread_id: None, @@ -1350,6 +1349,7 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> { let resume_id = mcp .send_thread_resume_request(ThreadResumeParams { thread_id: conversation_id.clone(), + exclude_turns: true, ..Default::default() }) .await?; diff --git a/codex-rs/state/migrations/0041_threads_name.sql b/codex-rs/state/migrations/0041_threads_name.sql new file mode 100644 index 000000000000..72ba802e8d38 --- /dev/null +++ b/codex-rs/state/migrations/0041_threads_name.sql @@ -0,0 +1 @@ +ALTER TABLE threads ADD COLUMN name TEXT; diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index 796852280862..f2d2b3c09a91 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -666,6 +666,7 @@ mod tests { cwd: PathBuf::from("/tmp"), cli_version: "0.0.0".to_string(), title: String::new(), + name: None, preview: None, sandbox_policy: "read-only".to_string(), approval_mode: "on-request".to_string(), diff --git a/codex-rs/state/src/model/thread_metadata.rs b/codex-rs/state/src/model/thread_metadata.rs index 0024c9866d39..60a8f3692a5d 100644 --- a/codex-rs/state/src/model/thread_metadata.rs +++ b/codex-rs/state/src/model/thread_metadata.rs @@ -73,7 +73,7 @@ pub struct ExtractionOutcome { pub parse_errors: usize, } -/// Canonical thread metadata derived from rollout files. +/// Canonical persisted thread metadata. #[derive(Debug, Clone, PartialEq, Eq)] pub struct ThreadMetadata { /// The thread identifier. @@ -110,6 +110,8 @@ pub struct ThreadMetadata { pub cli_version: String, /// A best-effort thread title. pub title: String, + /// Explicit user-facing thread name, if one was set. + pub name: Option, /// Best available user-facing preview for discovery and list display. pub preview: Option, /// The sandbox policy (stringified enum). @@ -245,6 +247,7 @@ impl ThreadMetadataBuilder { cwd: self.cwd.clone(), cli_version: self.cli_version.clone().unwrap_or_default(), title: String::new(), + name: None, preview: None, sandbox_policy, approval_mode, @@ -332,6 +335,9 @@ impl ThreadMetadata { if self.title != other.title { diffs.push("title"); } + if self.name != other.name { + diffs.push("name"); + } if self.preview != other.preview { diffs.push("preview"); } @@ -386,6 +392,7 @@ pub(crate) struct ThreadRow { cwd: String, cli_version: String, title: String, + name: Option, preview: String, sandbox_policy: String, approval_mode: String, @@ -417,6 +424,7 @@ impl ThreadRow { cwd: row.try_get("cwd")?, cli_version: row.try_get("cli_version")?, title: row.try_get("title")?, + name: row.try_get("name")?, preview: row.try_get("preview")?, sandbox_policy: row.try_get("sandbox_policy")?, approval_mode: row.try_get("approval_mode")?, @@ -452,6 +460,7 @@ impl TryFrom for ThreadMetadata { cwd, cli_version, title, + name, preview, sandbox_policy, approval_mode, @@ -486,6 +495,7 @@ impl TryFrom for ThreadMetadata { cwd: PathBuf::from(cwd), cli_version, title, + name, preview: (!preview.is_empty()).then_some(preview), sandbox_policy, approval_mode, @@ -583,6 +593,7 @@ mod tests { cwd: "/tmp/workspace".to_string(), cli_version: "0.0.0".to_string(), title: String::new(), + name: None, preview: String::new(), sandbox_policy: "read-only".to_string(), approval_mode: "on-request".to_string(), @@ -615,6 +626,7 @@ mod tests { cwd: PathBuf::from("/tmp/workspace"), cli_version: "0.0.0".to_string(), title: String::new(), + name: None, preview: None, sandbox_policy: "read-only".to_string(), approval_mode: "on-request".to_string(), diff --git a/codex-rs/state/src/runtime/memories.rs b/codex-rs/state/src/runtime/memories.rs index ff25fed982a1..fb82a82390b4 100644 --- a/codex-rs/state/src/runtime/memories.rs +++ b/codex-rs/state/src/runtime/memories.rs @@ -189,6 +189,7 @@ SELECT threads.cwd, threads.cli_version, threads.title, + threads.name, threads.preview, threads.sandbox_policy, threads.approval_mode, @@ -562,6 +563,7 @@ SELECT threads.cwd, threads.cli_version, threads.title, + threads.name, threads.preview, threads.sandbox_policy, threads.approval_mode, diff --git a/codex-rs/state/src/runtime/test_support.rs b/codex-rs/state/src/runtime/test_support.rs index d5dbe7e0af64..f4b23bfaa5cd 100644 --- a/codex-rs/state/src/runtime/test_support.rs +++ b/codex-rs/state/src/runtime/test_support.rs @@ -62,6 +62,7 @@ pub(super) fn test_thread_metadata( cwd, cli_version: "0.0.0".to_string(), title: String::new(), + name: None, preview: Some("hello".to_string()), sandbox_policy: crate::extract::enum_to_string(&SandboxPolicy::new_read_only_policy()), approval_mode: crate::extract::enum_to_string(&AskForApproval::OnRequest), diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index bdefa53c0433..4f2b8b5fed06 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -26,6 +26,7 @@ SELECT threads.cwd, threads.cli_version, threads.title, + threads.name, threads.preview, threads.sandbox_policy, threads.approval_mode, @@ -556,6 +557,7 @@ INSERT INTO threads ( cwd, cli_version, title, + name, preview, sandbox_policy, approval_mode, @@ -567,7 +569,7 @@ INSERT INTO threads ( git_branch, git_origin_url, memory_mode -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO NOTHING "#, ) @@ -601,6 +603,7 @@ ON CONFLICT(id) DO NOTHING .bind(metadata.cwd.display().to_string()) .bind(metadata.cli_version.as_str()) .bind(metadata.title.as_str()) + .bind(metadata.name.as_deref()) .bind(preview) .bind(metadata.sandbox_policy.as_str()) .bind(metadata.approval_mode.as_str()) @@ -645,6 +648,19 @@ ON CONFLICT(id) DO NOTHING Ok(result.rows_affected() > 0) } + pub async fn update_thread_name( + &self, + thread_id: ThreadId, + name: Option<&str>, + ) -> anyhow::Result { + let result = sqlx::query("UPDATE threads SET name = ? WHERE id = ?") + .bind(name) + .bind(thread_id.to_string()) + .execute(self.pool.as_ref()) + .await?; + Ok(result.rows_affected() > 0) + } + pub async fn touch_thread_updated_at( &self, thread_id: ThreadId, @@ -810,6 +826,7 @@ INSERT INTO threads ( cwd, cli_version, title, + name, preview, sandbox_policy, approval_mode, @@ -821,7 +838,7 @@ INSERT INTO threads ( git_branch, git_origin_url, memory_mode -) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET rollout_path = excluded.rollout_path, created_at = excluded.created_at, @@ -884,6 +901,7 @@ ON CONFLICT(id) DO UPDATE SET .bind(metadata.cwd.display().to_string()) .bind(metadata.cli_version.as_str()) .bind(metadata.title.as_str()) + .bind(metadata.name.as_deref()) .bind(preview) .bind(metadata.sandbox_policy.as_str()) .bind(metadata.approval_mode.as_str()) @@ -1223,6 +1241,7 @@ SELECT threads.cwd, threads.cli_version, threads.title, + threads.name, threads.preview, threads.sandbox_policy, threads.approval_mode, @@ -1322,7 +1341,9 @@ pub(super) fn push_thread_filters<'a>( None => {} } if let Some(search_term) = search_term { - builder.push(" AND (instr(threads.title, "); + builder.push(" AND (instr(COALESCE(threads.name, ''), "); + builder.push_bind(search_term); + builder.push(") > 0 OR instr(threads.title, "); builder.push_bind(search_term); builder.push(") > 0 OR instr(threads.preview, "); builder.push_bind(search_term); diff --git a/codex-rs/thread-store/src/local/helpers.rs b/codex-rs/thread-store/src/local/helpers.rs index d883bd7874e1..b8ac90507d58 100644 --- a/codex-rs/thread-store/src/local/helpers.rs +++ b/codex-rs/thread-store/src/local/helpers.rs @@ -1,3 +1,5 @@ +use std::collections::HashMap; +use std::collections::HashSet; use std::ffi::OsStr; use std::fs::FileTimes; use std::fs::OpenOptions; @@ -15,10 +17,13 @@ use codex_protocol::protocol::GitInfo; use codex_protocol::protocol::NetworkAccess; use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadHistoryMode; use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; use codex_rollout::ThreadItem; +use codex_rollout::find_thread_names_by_ids; use codex_state::ThreadMetadata; +use super::LocalThreadStore; use crate::StoredThread; use crate::ThreadStoreError; use crate::ThreadStoreResult; @@ -177,6 +182,52 @@ pub(super) fn permission_profile_to_metadata_value( } } +pub(super) fn sqlite_thread_name(metadata: &ThreadMetadata) -> Option { + metadata + .name + .as_deref() + .map(str::trim) + .filter(|name| !name.is_empty()) + .map(str::to_string) +} + +pub(super) async fn resolve_thread_names( + store: &LocalThreadStore, + thread_history_modes: &HashMap, +) -> HashMap { + let mut names = HashMap::::with_capacity(thread_history_modes.len()); + let legacy_thread_ids = thread_history_modes + .iter() + .filter_map(|(&thread_id, &history_mode)| { + (history_mode == ThreadHistoryMode::Legacy).then_some(thread_id) + }) + .collect::>(); + if let Some(state_db_ctx) = store.state_db().await { + for (&thread_id, &history_mode) in thread_history_modes { + let Ok(Some(metadata)) = state_db_ctx.get_thread(thread_id).await else { + continue; + }; + let name = match history_mode { + ThreadHistoryMode::Legacy => distinct_thread_metadata_title(&metadata), + ThreadHistoryMode::Paginated => sqlite_thread_name(&metadata), + }; + if let Some(name) = name { + names.insert(thread_id, name); + } + } + } + if let Ok(legacy_names) = + find_thread_names_by_ids(store.config.codex_home.as_path(), &legacy_thread_ids).await + { + // Legacy titles remain authoritative when present; the index only fills + // names for threads whose SQLite title is still derived from the preview. + for (thread_id, name) in legacy_names { + names.entry(thread_id).or_insert(name); + } + } + names +} + pub(super) fn distinct_thread_metadata_title(metadata: &ThreadMetadata) -> Option { let title = metadata.title.trim(); if title.is_empty() || metadata.first_user_message.as_deref().map(str::trim) == Some(title) { @@ -186,11 +237,10 @@ pub(super) fn distinct_thread_metadata_title(metadata: &ThreadMetadata) -> Optio } } -pub(super) fn set_thread_name_from_title(thread: &mut StoredThread, title: String) { - if title.trim().is_empty() || thread.preview.trim() == title.trim() { - return; +pub(super) fn set_thread_name(thread: &mut StoredThread, name: String) { + if thread.history_mode == ThreadHistoryMode::Paginated || thread.preview.trim() != name.trim() { + thread.name = Some(name); } - thread.name = Some(title); } fn parse_rfc3339(value: Option<&str>) -> Option> { diff --git a/codex-rs/thread-store/src/local/list_threads.rs b/codex-rs/thread-store/src/local/list_threads.rs index f514231333ba..e63d3e28bcf0 100644 --- a/codex-rs/thread-store/src/local/list_threads.rs +++ b/codex-rs/thread-store/src/local/list_threads.rs @@ -1,15 +1,12 @@ use std::collections::HashMap; -use std::collections::HashSet; -use codex_protocol::ThreadId; use codex_rollout::RolloutConfig; use codex_rollout::RolloutRecorder; -use codex_rollout::find_thread_names_by_ids; use codex_rollout::parse_cursor; use super::LocalThreadStore; -use super::helpers::distinct_thread_metadata_title; -use super::helpers::set_thread_name_from_title; +use super::helpers::resolve_thread_names; +use super::helpers::set_thread_name; use super::helpers::stored_thread_from_rollout_item; use crate::ListThreadsParams; use crate::SortDirection; @@ -77,32 +74,14 @@ pub(super) async fn list_threads( }) .collect::>(); - let thread_ids = items + let thread_history_modes = items .iter() - .map(|thread| thread.thread_id) - .collect::>(); - let mut names = HashMap::::with_capacity(thread_ids.len()); - if let Some(state_db_ctx) = store.state_db().await { - for &thread_id in &thread_ids { - let Ok(Some(metadata)) = state_db_ctx.get_thread(thread_id).await else { - continue; - }; - if let Some(title) = distinct_thread_metadata_title(&metadata) { - names.insert(thread_id, title); - } - } - } - if names.len() < thread_ids.len() - && let Ok(legacy_names) = - find_thread_names_by_ids(store.config.codex_home.as_path(), &thread_ids).await - { - for (thread_id, title) in legacy_names { - names.entry(thread_id).or_insert(title); - } - } + .map(|thread| (thread.thread_id, thread.history_mode)) + .collect::>(); + let names = resolve_thread_names(store, &thread_history_modes).await; for thread in &mut items { - if let Some(title) = names.get(&thread.thread_id).cloned() { - set_thread_name_from_title(thread, title); + if let Some(name) = names.get(&thread.thread_id).cloned() { + set_thread_name(thread, name); } } @@ -337,6 +316,74 @@ mod tests { ); } + #[tokio::test] + async fn list_paginated_threads_uses_sqlite_name_over_legacy_compatibility() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let uuid = Uuid::from_u128(104); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let rollout_path = home.path().join("rollout-paginated-name-search.jsonl"); + fs::write(&rollout_path, "").expect("placeholder rollout file"); + + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config.clone(), Some(runtime.clone())); + runtime + .mark_backfill_complete(/*last_watermark*/ None) + .await + .expect("backfill should be complete"); + let mut builder = codex_state::ThreadMetadataBuilder::new( + thread_id, + rollout_path, + Utc::now(), + SessionSource::Cli, + ); + builder.history_mode = ThreadHistoryMode::Paginated; + builder.model_provider = Some(config.default_model_provider_id.clone()); + builder.cwd = home.path().to_path_buf(); + builder.cli_version = Some("test_version".to_string()); + let mut metadata = builder.build(config.default_model_provider_id.as_str()); + metadata.name = Some("canonical paginated name".to_string()); + metadata.title = "stale title name".to_string(); + metadata.first_user_message = Some("plain preview".to_string()); + metadata.preview = metadata.first_user_message.clone(); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + codex_rollout::append_thread_name(home.path(), thread_id, "stale index name") + .await + .expect("append legacy thread name"); + + 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, + archived: false, + search_term: Some("canonical".to_string()), + relation_filter: None, + use_state_db_only: true, + }) + .await + .expect("thread listing"); + + assert_eq!(page.items.len(), 1); + assert_eq!(page.items[0].thread_id, thread_id); + assert_eq!( + page.items[0].name.as_deref(), + Some("canonical paginated name") + ); + } + #[tokio::test] async fn list_threads_selects_active_or_archived_collection() { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index 035774b04e0f..5be86d882597 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -4,6 +4,7 @@ use codex_protocol::models::PermissionProfile; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadHistoryMode; use codex_rollout::RolloutRecorder; use codex_rollout::find_archived_thread_path_by_id_str; use codex_rollout::find_thread_name_by_id; @@ -17,7 +18,8 @@ use super::helpers::distinct_thread_metadata_title; use super::helpers::git_info_from_parts; use super::helpers::permission_profile_from_metadata_value; use super::helpers::rollout_path_is_archived; -use super::helpers::set_thread_name_from_title; +use super::helpers::set_thread_name; +use super::helpers::sqlite_thread_name; use super::helpers::stored_thread_from_rollout_item; use super::live_writer; use crate::ReadThreadParams; @@ -119,6 +121,9 @@ pub(super) async fn read_thread_by_rollout_path( }); } if let Some(metadata) = read_sqlite_metadata(store, thread.thread_id).await { + if thread.history_mode == ThreadHistoryMode::Paginated { + thread.name = sqlite_thread_name(&metadata); + } thread.recency_at = metadata.recency_at; let existing_git_info = thread.git_info.take(); let (fallback_sha, fallback_branch, fallback_origin_url) = match existing_git_info { @@ -283,10 +288,12 @@ async fn read_thread_from_rollout_path( { thread.model_provider = model_provider; } - if let Ok(Some(title)) = - find_thread_name_by_id(store.config.codex_home.as_path(), &thread.thread_id).await + if thread.history_mode == ThreadHistoryMode::Legacy + && let Ok(Some(name)) = + find_thread_name_by_id(store.config.codex_home.as_path(), &thread.thread_id).await + && !name.trim().is_empty() { - set_thread_name_from_title(&mut thread, title); + set_thread_name(&mut thread, name); } Ok(thread) } @@ -314,14 +321,6 @@ async fn stored_thread_from_sqlite_metadata( store: &LocalThreadStore, metadata: ThreadMetadata, ) -> ThreadStoreResult { - let name = match distinct_thread_metadata_title(&metadata) { - Some(title) => Some(title), - None => find_thread_name_by_id(store.config.codex_home.as_path(), &metadata.id) - .await - .ok() - .flatten() - .filter(|title| !title.trim().is_empty()), - }; let session_meta = match read_required_session_meta_line(metadata.rollout_path.as_path()).await { Ok(meta_line) => Some(meta_line.meta), @@ -348,6 +347,7 @@ async fn stored_thread_from_sqlite_metadata( .as_ref() .map(|meta| meta.history_mode) .unwrap_or(metadata.history_mode); + let name = thread_name_from_metadata(store, &metadata, history_mode).await; let preview = metadata .preview .clone() @@ -395,6 +395,27 @@ async fn stored_thread_from_sqlite_metadata( }) } +async fn thread_name_from_metadata( + store: &LocalThreadStore, + metadata: &ThreadMetadata, + history_mode: ThreadHistoryMode, +) -> Option { + match history_mode { + ThreadHistoryMode::Paginated => sqlite_thread_name(metadata), + ThreadHistoryMode::Legacy => { + if let Some(title) = distinct_thread_metadata_title(metadata) { + Some(title) + } else { + find_thread_name_by_id(store.config.codex_home.as_path(), &metadata.id) + .await + .ok() + .flatten() + .filter(|name| !name.trim().is_empty()) + } + } + } +} + async fn stored_thread_from_session_meta( store: &LocalThreadStore, path: std::path::PathBuf, @@ -508,6 +529,7 @@ mod tests { use crate::local::test_support::write_archived_session_file; use crate::local::test_support::write_session_file; use crate::local::test_support::write_session_file_with_fork; + use crate::local::test_support::write_session_file_with_history_mode; #[tokio::test] async fn read_thread_returns_active_rollout_summary() { @@ -568,7 +590,7 @@ mod tests { } #[tokio::test] - async fn read_thread_by_rollout_path_prefers_sqlite_git_info() { + async fn read_thread_by_rollout_path_preserves_legacy_name_and_sqlite_git_info() { let home = TempDir::new().expect("temp dir"); let config = test_config(home.path()); let uuid = Uuid::from_u128(223); @@ -594,10 +616,15 @@ mod tests { .expect("timestamp should parse") .with_timezone(&Utc); builder.recency_at = Some(recency_at); + let mut metadata = builder.build(config.default_model_provider_id.as_str()); + metadata.title = "Stale SQLite name".to_string(); runtime - .upsert_thread(&builder.build(config.default_model_provider_id.as_str())) + .upsert_thread(&metadata) .await .expect("state db upsert should succeed"); + codex_rollout::append_thread_name(home.path(), thread_id, "Latest index name") + .await + .expect("append thread name"); let thread = store .read_thread_by_rollout_path( @@ -609,6 +636,7 @@ mod tests { .expect("read thread by rollout path"); let git_info = thread.git_info.expect("git info should be present"); + assert_eq!(thread.name.as_deref(), Some("Latest index name")); assert_eq!(thread.recency_at, recency_at); assert_eq!(git_info.branch.as_deref(), Some("sqlite-branch")); assert_eq!( @@ -760,6 +788,54 @@ mod tests { assert_eq!(thread.name, Some("Saved title".to_string())); } + #[tokio::test] + async fn read_paginated_thread_uses_sqlite_name_over_legacy_compatibility() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let uuid = Uuid::from_u128(228); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let rollout_path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T12-00-00", + uuid, + ThreadHistoryMode::Paginated, + ) + .expect("session file"); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config.clone(), Some(runtime.clone())); + let mut builder = + ThreadMetadataBuilder::new(thread_id, rollout_path, Utc::now(), SessionSource::Cli); + builder.history_mode = ThreadHistoryMode::Paginated; + builder.model_provider = Some(config.default_model_provider_id.clone()); + builder.cwd = home.path().to_path_buf(); + let mut metadata = builder.build(config.default_model_provider_id.as_str()); + metadata.name = Some("Canonical SQLite name".to_string()); + metadata.title = "Stale title name".to_string(); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + codex_rollout::append_thread_name(home.path(), thread_id, "Stale index name") + .await + .expect("append legacy thread name"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect("read thread"); + + assert_eq!(thread.name.as_deref(), Some("Canonical SQLite name")); + } + #[tokio::test] async fn read_thread_returns_permission_profile_from_sqlite_metadata() { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/thread-store/src/local/search_threads.rs b/codex-rs/thread-store/src/local/search_threads.rs index e881dabd595b..84ecfea96384 100644 --- a/codex-rs/thread-store/src/local/search_threads.rs +++ b/codex-rs/thread-store/src/local/search_threads.rs @@ -1,17 +1,14 @@ use std::collections::HashMap; -use std::collections::HashSet; use codex_install_context::InstallContext; -use codex_protocol::ThreadId; use codex_rollout::RolloutConfig; -use codex_rollout::find_thread_names_by_ids; use codex_rollout::first_rollout_content_match_snippet; use codex_rollout::parse_cursor; use codex_rollout::search_rollout_matches; use super::LocalThreadStore; -use super::helpers::distinct_thread_metadata_title; -use super::helpers::set_thread_name_from_title; +use super::helpers::resolve_thread_names; +use super::helpers::set_thread_name; use super::helpers::stored_thread_from_rollout_item; use super::list_threads::list_rollout_threads; use crate::ListThreadsParams; @@ -201,32 +198,14 @@ async fn set_thread_search_result_names( store: &LocalThreadStore, items: &mut [StoredThreadSearchResult], ) { - let thread_ids = items + let thread_history_modes = items .iter() - .map(|item| item.thread.thread_id) - .collect::>(); - let mut names = HashMap::::with_capacity(thread_ids.len()); - if let Some(state_db_ctx) = store.state_db().await { - for &thread_id in &thread_ids { - let Ok(Some(metadata)) = state_db_ctx.get_thread(thread_id).await else { - continue; - }; - if let Some(title) = distinct_thread_metadata_title(&metadata) { - names.insert(thread_id, title); - } - } - } - if names.len() < thread_ids.len() - && let Ok(legacy_names) = - find_thread_names_by_ids(store.config.codex_home.as_path(), &thread_ids).await - { - for (thread_id, title) in legacy_names { - names.entry(thread_id).or_insert(title); - } - } + .map(|item| (item.thread.thread_id, item.thread.history_mode)) + .collect::>(); + let names = resolve_thread_names(store, &thread_history_modes).await; for item in items { - if let Some(title) = names.get(&item.thread.thread_id).cloned() { - set_thread_name_from_title(&mut item.thread, title); + if let Some(name) = names.get(&item.thread.thread_id).cloned() { + set_thread_name(&mut item.thread, name); } } } diff --git a/codex-rs/thread-store/src/local/update_thread_metadata.rs b/codex-rs/thread-store/src/local/update_thread_metadata.rs index 800e1799c091..58b684a0332a 100644 --- a/codex-rs/thread-store/src/local/update_thread_metadata.rs +++ b/codex-rs/thread-store/src/local/update_thread_metadata.rs @@ -54,30 +54,55 @@ pub(super) async fn update_thread_metadata( .await; } - let needs_rollout_compat = needs_rollout_compatibility_update(&patch); - if needs_rollout_compat { - // These explicit patches still write legacy rollout/name-index state after the - // SQLite update. Paginated threads must fail before either side is mutated. - let thread = read_thread::read_thread( - store, - ReadThreadParams { - thread_id, - include_archived: params.include_archived, - include_history: false, - }, + let requires_rollout_compat = requires_rollout_compatibility_update(&patch); + let history_mode = if patch.name.is_some() || requires_rollout_compat { + Some( + read_thread::read_thread( + store, + ReadThreadParams { + thread_id, + include_archived: params.include_archived, + include_history: false, + }, + ) + .await? + .history_mode, ) - .await?; - reject_paginated_history_mode(thread.history_mode)?; + } else { + None + }; + if requires_rollout_compat { + // Explicit patches that require legacy rollout state must fail before either + // persistence layer is mutated. + if let Some(history_mode) = history_mode { + reject_paginated_history_mode(history_mode)?; + } } - let require_sqlite_write = sqlite_write_failure_should_block(&patch); + let paginated_name = + patch.name.is_some() && matches!(history_mode, Some(ThreadHistoryMode::Paginated)); + let needs_rollout_compat = requires_rollout_compat || patch.name.is_some() && !paginated_name; + let require_sqlite_write = sqlite_write_failure_should_block(&patch) || paginated_name; let updated = apply_metadata_update( store, thread_id, patch.clone(), params.include_archived, require_sqlite_write, + history_mode, ) .await?; + if paginated_name && let Some(name) = patch.name.as_ref() { + if let Err(err) = append_thread_name( + store.config.codex_home.as_path(), + thread_id, + name.as_deref().unwrap_or_default(), + ) + .await + { + warn!("failed to index paginated thread name for {thread_id}: {err}"); + } + return Ok(updated); + } if !needs_rollout_compat { return Ok(updated); } @@ -108,7 +133,15 @@ pub(super) async fn update_thread_metadata( .await; if let Some(name) = name { - apply_thread_name(store, thread_id, name.unwrap_or_default()).await?; + append_thread_name( + store.config.codex_home.as_path(), + thread_id, + &name.unwrap_or_default(), + ) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to index thread name: {err}"), + })?; } let resolved_git_info = match git_info { @@ -203,6 +236,7 @@ async fn apply_metadata_update( patch: ThreadMetadataPatch, include_archived: bool, require_sqlite_write: bool, + history_mode: Option, ) -> ThreadStoreResult { let live_rollout_path = live_writer::rollout_path(store, thread_id).await.ok(); let mut rollout_path = patch.rollout_path.clone().or(live_rollout_path); @@ -253,9 +287,6 @@ async fn apply_metadata_update( if let Some(preview) = patch.preview { metadata.preview = Some(preview); } - if let Some(name) = patch.name { - metadata.title = name.unwrap_or_default(); - } if let Some(title) = patch.title { metadata.title = title; } @@ -329,6 +360,35 @@ async fn apply_metadata_update( .map_err(|err| ThreadStoreError::Internal { message: format!("failed to update thread metadata for {thread_id}: {err}"), })?; + if let Some(name) = patch.name.as_ref() { + let history_mode = history_mode.ok_or_else(|| ThreadStoreError::Internal { + message: format!( + "thread history mode unavailable before name update: {thread_id}" + ), + })?; + let updated = match history_mode { + ThreadHistoryMode::Legacy => { + state_db + .update_thread_title(thread_id, name.as_deref().unwrap_or_default()) + .await + } + ThreadHistoryMode::Paginated => { + state_db + .update_thread_name(thread_id, name.as_deref()) + .await + } + } + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to set thread name: {err}"), + })?; + if !updated { + return Err(ThreadStoreError::Internal { + message: format!( + "thread metadata unavailable before name update: {thread_id}" + ), + }); + } + } if existing.is_some() && let Some(recency_at) = advance_recency_at { @@ -352,22 +412,19 @@ async fn apply_metadata_update( Ok(()) } .await + } else if require_sqlite_write { + Err(ThreadStoreError::Internal { + message: format!("sqlite state db unavailable for thread {thread_id}"), + }) } else { Ok(()) }; - match (state_db.is_some(), sqlite_write_result) { - (true, Ok(())) => {} - (true, Err(err)) if require_sqlite_write || !sqlite_write_error_is_best_effort(&err) => { + match sqlite_write_result { + Ok(()) => {} + Err(err) if require_sqlite_write || !sqlite_write_error_is_best_effort(&err) => { return Err(err); } - (true, Err(err)) => { - warn!("state db update_thread_metadata failed for {thread_id}: {err}"); - } - (false, Ok(())) => {} - (false, Err(err)) if require_sqlite_write || !sqlite_write_error_is_best_effort(&err) => { - return Err(err); - } - (false, Err(err)) => { + Err(err) => { warn!("state db update_thread_metadata failed for {thread_id}: {err}"); } } @@ -455,10 +512,7 @@ async fn canonical_history_mode( Ok(session_meta.meta.history_mode) } -fn needs_rollout_compatibility_update(patch: &ThreadMetadataPatch) -> bool { - if patch.name.is_some() { - return true; - } +fn requires_rollout_compatibility_update(patch: &ThreadMetadataPatch) -> bool { if patch.memory_mode.is_none() && patch.git_info.is_none() { return false; } @@ -597,32 +651,6 @@ async fn apply_thread_git_info_to_rollout( }) } -async fn apply_thread_name( - store: &LocalThreadStore, - thread_id: ThreadId, - name: String, -) -> ThreadStoreResult<()> { - if let Some(state_db) = store.state_db().await { - let updated = state_db - .update_thread_title(thread_id, &name) - .await - .map_err(|err| ThreadStoreError::Internal { - message: format!("failed to set thread name: {err}"), - })?; - if !updated { - return Err(ThreadStoreError::Internal { - message: format!("thread metadata unavailable before name update: {thread_id}"), - }); - } - } - - append_thread_name(store.config.codex_home.as_path(), thread_id, &name) - .await - .map_err(|err| ThreadStoreError::Internal { - message: format!("failed to index thread name: {err}"), - }) -} - async fn apply_thread_memory_mode( rollout_path: &Path, thread_id: ThreadId, @@ -767,6 +795,104 @@ mod tests { assert_eq!(latest_name.as_deref(), Some("A sharper name")); } + #[tokio::test] + async fn paginated_name_updates_use_sqlite_without_rollout_writes() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let uuid = Uuid::from_u128(318); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T14-15-00", + uuid, + ThreadHistoryMode::Paginated, + ) + .expect("session file"); + let original_rollout = std::fs::read_to_string(&path).expect("read rollout"); + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config, Some(runtime.clone())); + let thread = store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + name: Some(Some("Canonical paginated name".to_string())), + ..Default::default() + }, + include_archived: false, + }) + .await + .expect("set paginated thread name"); + + assert_eq!(thread.name.as_deref(), Some("Canonical paginated name")); + let metadata = runtime + .get_thread(thread_id) + .await + .expect("read metadata") + .expect("thread metadata"); + assert_eq!(metadata.name.as_deref(), Some("Canonical paginated name")); + assert!(metadata.title.is_empty()); + assert_eq!( + codex_rollout::find_thread_name_by_id(home.path(), &thread_id) + .await + .expect("find thread name") + .as_deref(), + Some("Canonical paginated name") + ); + + let thread = store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + title: Some("Derived first message".to_string()), + preview: Some("Derived first message".to_string()), + ..Default::default() + }, + include_archived: false, + }) + .await + .expect("apply derived paginated metadata"); + assert_eq!(thread.name.as_deref(), Some("Canonical paginated name")); + + let session_index_path = home.path().join("session_index.jsonl"); + std::fs::remove_file(&session_index_path).expect("remove session index"); + std::fs::create_dir(&session_index_path).expect("block session index writes"); + let thread = store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + name: Some(Some("Updated SQLite name".to_string())), + ..Default::default() + }, + include_archived: false, + }) + .await + .expect("set paginated thread name with unavailable index"); + assert_eq!(thread.name.as_deref(), Some("Updated SQLite name")); + + let err = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None) + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + patch: ThreadMetadataPatch { + name: Some(Some("Unpersistable name".to_string())), + ..Default::default() + }, + include_archived: false, + }) + .await + .expect_err("paginated name update without sqlite should fail"); + + assert!(matches!(err, ThreadStoreError::Internal { .. })); + assert_eq!( + std::fs::read_to_string(&path).expect("read rollout"), + original_rollout + ); + } + #[tokio::test] async fn update_thread_metadata_sets_memory_mode_on_active_rollout() { let home = TempDir::new().expect("temp dir"); @@ -834,6 +960,7 @@ mod tests { .update_thread_metadata(UpdateThreadMetadataParams { thread_id, patch: ThreadMetadataPatch { + name: Some(Some("Must not persist".to_string())), memory_mode: Some(ThreadMemoryMode::Disabled), ..Default::default() }, @@ -858,6 +985,21 @@ mod tests { .as_deref(), Some("enabled") ); + assert_eq!( + runtime + .get_thread(thread_id) + .await + .expect("thread metadata should be readable") + .expect("thread metadata") + .name, + None + ); + assert_eq!( + codex_rollout::find_thread_name_by_id(home.path(), &thread_id) + .await + .expect("find thread name"), + None + ); } #[tokio::test]