diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 4e3bc70cb..aae240e3f 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -199,6 +199,7 @@ use codex_core::SteerInputError; use codex_core::ThreadConfigSnapshot; use codex_core::ThreadManager; use codex_core::ThreadSortKey as CoreThreadSortKey; +use codex_core::append_thread_name; use codex_core::config::Config; use codex_core::config::ConfigOverrides; use codex_core::config::NetworkProxyAuditMetadata; @@ -288,11 +289,13 @@ use codex_protocol::protocol::ReviewTarget as CoreReviewTarget; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionMetaLine; +use codex_protocol::protocol::ThreadNameUpdatedEvent; use codex_protocol::protocol::USER_MESSAGE_BEGIN; use codex_protocol::protocol::W3cTraceContext; use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS; use codex_protocol::user_input::UserInput as CoreInputItem; use codex_rmcp_client::perform_oauth_login_return_url; +use codex_rollout::append_rollout_item_to_path; use codex_rollout::state_db::StateDbHandle; use codex_rollout::state_db::get_state_db; use codex_rollout::state_db::reconcile_rollout; @@ -2674,11 +2677,11 @@ impl CodexMessageProcessor { return; } - let thread_exists = + let rollout_path = match find_thread_path_by_id_str(&self.config.codex_home, &thread_id.to_string()).await { - Ok(Some(_)) => true, - Ok(None) => false, + Ok(Some(path)) => Some(path), + Ok(None) => None, Err(err) => { self.send_invalid_request_error( request_id, @@ -2689,19 +2692,39 @@ impl CodexMessageProcessor { } }; - if !thread_exists { + let Some(rollout_path) = rollout_path else { self.send_invalid_request_error(request_id, format!("thread not found: {thread_id}")) .await; return; - } + }; - if let Err(err) = - codex_core::append_thread_name(&self.config.codex_home, thread_id, &name).await - { + let msg = EventMsg::ThreadNameUpdated(ThreadNameUpdatedEvent { + thread_id, + thread_name: Some(name.clone()), + }); + let item = RolloutItem::EventMsg(msg); + if let Err(err) = append_rollout_item_to_path(rollout_path.as_path(), &item).await { self.send_internal_error(request_id, format!("failed to set thread name: {err}")) .await; return; } + if let Err(err) = append_thread_name(&self.config.codex_home, thread_id, &name).await { + self.send_internal_error(request_id, format!("failed to index thread name: {err}")) + .await; + return; + } + + let state_db_ctx = open_state_db_for_direct_thread_lookup(&self.config).await; + reconcile_rollout( + state_db_ctx.as_deref(), + rollout_path.as_path(), + self.config.model_provider_id.as_str(), + /*builder*/ None, + &[], + /*archived_only*/ None, + /*new_thread_memory_mode*/ None, + ) + .await; self.outgoing .send_response(request_id, ThreadSetNameResponse {}) @@ -3445,13 +3468,7 @@ impl CodexMessageProcessor { threads.push((conversation_id, thread)); } - let names = match find_thread_names_by_ids(&self.config.codex_home, &thread_ids).await { - Ok(names) => names, - Err(err) => { - warn!("Failed to read thread names: {err}"); - HashMap::new() - } - }; + let names = thread_titles_by_ids(&self.config, &thread_ids).await; let statuses = self .thread_watch_manager @@ -3461,7 +3478,9 @@ impl CodexMessageProcessor { let data = threads .into_iter() .map(|(conversation_id, mut thread)| { - thread.name = names.get(&conversation_id).cloned(); + if let Some(title) = names.get(&conversation_id).cloned() { + set_thread_name_from_title(&mut thread, title); + } if let Some(status) = statuses.get(&thread.id) { thread.status = status.clone(); } @@ -4265,13 +4284,8 @@ impl CodexMessageProcessor { } async fn attach_thread_name(&self, thread_id: ThreadId, thread: &mut Thread) { - match find_thread_name_by_id(&self.config.codex_home, &thread_id).await { - Ok(name) => { - thread.name = name; - } - Err(err) => { - warn!("Failed to read thread name for {thread_id}: {err}"); - } + if let Some(title) = title_from_state_db(&self.config, thread_id).await { + set_thread_name_from_title(thread, title); } } @@ -8029,7 +8043,7 @@ async fn handle_thread_listener_command( async fn handle_pending_thread_resume_request( conversation_id: ThreadId, conversation: &Arc, - codex_home: &Path, + _codex_home: &Path, thread_state_manager: &ThreadStateManager, thread_state: &Arc>, thread_watch_manager: &ThreadWatchManager, @@ -8087,11 +8101,6 @@ async fn handle_pending_thread_resume_request( has_live_in_progress_turn, ); - match find_thread_name_by_id(codex_home, &conversation_id).await { - Ok(thread_name) => thread.name = thread_name, - Err(err) => warn!("Failed to read thread name for {conversation_id}: {err}"), - } - let ThreadConfigSnapshot { model, model_provider_id, @@ -8653,7 +8662,7 @@ async fn read_summary_from_state_db_by_thread_id( config: &Config, thread_id: ThreadId, ) -> Option { - let state_db_ctx = get_state_db(config).await; + let state_db_ctx = open_state_db_for_direct_thread_lookup(config).await; read_summary_from_state_db_context_by_thread_id(state_db_ctx.as_ref(), thread_id).await } @@ -8670,6 +8679,71 @@ async fn read_summary_from_state_db_context_by_thread_id( Some(summary_from_thread_metadata(&metadata)) } +async fn title_from_state_db(config: &Config, thread_id: ThreadId) -> Option { + if let Some(state_db_ctx) = open_state_db_for_direct_thread_lookup(config).await + && let Some(metadata) = state_db_ctx.get_thread(thread_id).await.ok().flatten() + && let Some(title) = distinct_title(&metadata) + { + return Some(title); + } + find_thread_name_by_id(&config.codex_home, &thread_id) + .await + .ok() + .flatten() +} + +async fn thread_titles_by_ids( + config: &Config, + thread_ids: &HashSet, +) -> HashMap { + let mut names = HashMap::with_capacity(thread_ids.len()); + if let Some(state_db_ctx) = open_state_db_for_direct_thread_lookup(config).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_title(&metadata) { + names.insert(thread_id, title); + } + } + } + if names.len() < thread_ids.len() + && let Ok(legacy_names) = find_thread_names_by_ids(&config.codex_home, thread_ids).await + { + for (thread_id, title) in legacy_names { + names.entry(thread_id).or_insert(title); + } + } + names +} + +async fn open_state_db_for_direct_thread_lookup(config: &Config) -> Option { + StateRuntime::init(config.sqlite_home.clone(), config.model_provider_id.clone()) + .await + .ok() +} + +fn non_empty_title(metadata: &ThreadMetadata) -> Option { + let title = metadata.title.trim(); + (!title.is_empty()).then(|| title.to_string()) +} + +fn distinct_title(metadata: &ThreadMetadata) -> Option { + let title = non_empty_title(metadata)?; + if metadata.first_user_message.as_deref().map(str::trim) == Some(title.as_str()) { + None + } else { + Some(title) + } +} + +fn set_thread_name_from_title(thread: &mut Thread, title: String) { + if title.trim().is_empty() || thread.preview.trim() == title.trim() { + return; + } + thread.name = Some(title); +} + async fn summary_from_thread_list_item( it: codex_core::ThreadItem, fallback_provider: &str, @@ -8968,6 +9042,14 @@ async fn load_thread_summary_for_rollout( } else if let Some(summary) = read_summary_from_state_db_by_thread_id(config, thread_id).await { merge_mutable_thread_metadata(&mut thread, summary_to_thread(summary)); } + let title = if let Some(metadata) = persisted_metadata { + non_empty_title(metadata) + } else { + title_from_state_db(config, thread_id).await + }; + if let Some(title) = title { + set_thread_name_from_title(&mut thread, title); + } Ok(thread) } diff --git a/codex-rs/app-server/tests/suite/v2/thread_name_websocket.rs b/codex-rs/app-server/tests/suite/v2/thread_name_websocket.rs index 4ec7344d6..b41e2f1d1 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_name_websocket.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_name_websocket.rs @@ -21,7 +21,14 @@ use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadSetNameParams; use codex_app_server_protocol::ThreadSetNameResponse; +use codex_core::find_thread_name_by_id; +use codex_core::find_thread_path_by_id_str; +use codex_protocol::ThreadId; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::RolloutLine; use pretty_assertions::assert_eq; +use std::path::Path; use tempfile::TempDir; use tokio::time::Duration; use tokio::time::timeout; @@ -77,6 +84,11 @@ async fn thread_name_updated_broadcasts_for_loaded_threads() -> Result<()> { let ws2_notification = read_notification_for_method(&mut ws2, "thread/name/updated").await?; assert_thread_name_updated(ws2_notification, &conversation_id, renamed)?; + assert_legacy_thread_name(codex_home.path(), &conversation_id, renamed).await?; + assert_eq!( + thread_name_update_rollout_count(codex_home.path(), &conversation_id).await?, + 1 + ); assert_no_message(&mut ws1, Duration::from_millis(250)).await?; assert_no_message(&mut ws2, Duration::from_millis(250)).await?; @@ -128,6 +140,11 @@ async fn thread_name_updated_broadcasts_for_not_loaded_threads() -> Result<()> { let ws2_notification = read_notification_for_method(&mut ws2, "thread/name/updated").await?; assert_thread_name_updated(ws2_notification, &conversation_id, renamed)?; + assert_legacy_thread_name(codex_home.path(), &conversation_id, renamed).await?; + assert_eq!( + thread_name_update_rollout_count(codex_home.path(), &conversation_id).await?, + 1 + ); assert_no_message(&mut ws1, Duration::from_millis(250)).await?; assert_no_message(&mut ws2, Duration::from_millis(250)).await?; @@ -174,3 +191,38 @@ fn assert_thread_name_updated( assert_eq!(notification.thread_name.as_deref(), Some(thread_name)); Ok(()) } + +async fn assert_legacy_thread_name( + codex_home: &Path, + conversation_id: &str, + expected_name: &str, +) -> Result<()> { + let thread_id = ThreadId::from_string(conversation_id)?; + assert_eq!( + find_thread_name_by_id(codex_home, &thread_id) + .await? + .as_deref(), + Some(expected_name) + ); + Ok(()) +} + +async fn thread_name_update_rollout_count( + codex_home: &Path, + conversation_id: &str, +) -> Result { + let rollout_path = find_thread_path_by_id_str(codex_home, conversation_id) + .await? + .context("rollout path")?; + let contents = tokio::fs::read_to_string(rollout_path).await?; + Ok(contents + .lines() + .filter_map(|line| serde_json::from_str::(line).ok()) + .filter(|line| { + matches!( + line.item, + RolloutItem::EventMsg(EventMsg::ThreadNameUpdated(_)) + ) + }) + .count()) +} diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 54b9d524b..6375f2a83 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -33,7 +33,7 @@ use crate::realtime_conversation::handle_close as handle_realtime_conversation_c use crate::realtime_conversation::handle_start as handle_realtime_conversation_start; use crate::realtime_conversation::handle_text as handle_realtime_conversation_text; use crate::render_skills_section; -use crate::rollout::session_index; +use crate::rollout::find_thread_name_by_id; use crate::session_prefix::format_subagent_notification_message; use crate::skills_load_input_from_config; use crate::stream_events_utils::HandleOutputCtx; @@ -1096,6 +1096,26 @@ fn local_time_context() -> (String, String) { } } +async fn thread_title_from_state_db( + state_db: Option<&state_db::StateDbHandle>, + codex_home: &Path, + conversation_id: ThreadId, +) -> Option { + if let Some(metadata) = state_db + && let Some(metadata) = metadata.get_thread(conversation_id).await.ok().flatten() + { + let title = metadata.title.trim(); + if !title.is_empty() && metadata.first_user_message.as_deref().map(str::trim) != Some(title) + { + return Some(title.to_string()); + } + } + find_thread_name_by_id(codex_home, &conversation_id) + .await + .ok() + .flatten() +} + #[derive(Clone)] pub(crate) struct SessionConfiguration { /// Provider identifier ("openai", "openrouter", ...). @@ -1882,19 +1902,12 @@ impl Session { tx }; let thread_name = - match session_index::find_thread_name_by_id(&config.codex_home, &conversation_id) + thread_title_from_state_db(state_db_ctx.as_ref(), &config.codex_home, conversation_id) .instrument(info_span!( "session_init.thread_name_lookup", otel.name = "session_init.thread_name_lookup", )) - .await - { - Ok(name) => name, - Err(err) => { - warn!("Failed to read session index for thread name: {err}"); - None - } - }; + .await; session_configuration.thread_name = thread_name.clone(); let state = SessionState::new(session_configuration.clone()); let managed_network_requirements_enabled = config.managed_network_requirements_enabled(); @@ -4848,7 +4861,6 @@ mod handlers { use crate::review_prompts::resolve_review_request; use crate::rollout::RolloutRecorder; - use crate::rollout::session_index; use crate::tasks::CompactTask; use crate::tasks::UndoTask; use crate::tasks::UserShellCommandMode; @@ -5528,13 +5540,25 @@ mod handlers { .await; } - /// Persists the thread name in the session index, updates in-memory state, and emits - /// a `ThreadNameUpdated` event on success. - /// - /// This appends the name to `CODEX_HOME/sessions_index.jsonl` via `session_index::append_thread_name` for the - /// current `thread_id`, then updates `SessionConfiguration::thread_name`. - /// - /// Returns an error event if the name is empty or session persistence is disabled. + async fn persist_thread_name_update( + sess: &Arc, + event: ThreadNameUpdatedEvent, + ) -> anyhow::Result { + let msg = EventMsg::ThreadNameUpdated(event); + let item = RolloutItem::EventMsg(msg.clone()); + let recorder = { + let guard = sess.services.rollout.lock().await; + guard.clone() + } + .ok_or_else(|| anyhow::anyhow!("Session persistence is disabled; cannot rename thread."))?; + recorder.persist().await?; + recorder.record_items(std::slice::from_ref(&item)).await?; + recorder.flush().await?; + Ok(msg) + } + + /// Persists the thread name in the rollout and state database, updates in-memory state, and + /// emits a `ThreadNameUpdated` event on success. pub async fn set_thread_name(sess: &Arc, sub_id: String, name: String) { let Some(name) = crate::util::normalize_thread_name(&name) else { let event = Event { @@ -5548,47 +5572,33 @@ mod handlers { return; }; - let persistence_enabled = { - let rollout = sess.services.rollout.lock().await; - rollout.is_some() - }; - if !persistence_enabled { - let event = Event { - id: sub_id, - msg: EventMsg::Error(ErrorEvent { - message: "Session persistence is disabled; cannot rename thread.".to_string(), - codex_error_info: Some(CodexErrorInfo::Other), - }), - }; - sess.send_event_raw(event).await; - return; + let updated = ThreadNameUpdatedEvent { + thread_id: sess.conversation_id, + thread_name: Some(name.clone()), }; - if let Err(e) = sess.try_ensure_rollout_materialized().await { - let event = Event { - id: sub_id, - msg: EventMsg::Error(ErrorEvent { - message: format!("Failed to set thread name: {e}"), - codex_error_info: Some(CodexErrorInfo::Other), - }), - }; - sess.send_event_raw(event).await; - return; - } + let msg = match persist_thread_name_update(sess, updated).await { + Ok(msg) => msg, + Err(err) => { + warn!("Failed to persist thread name update to rollout: {err}"); + let event = Event { + id: sub_id, + msg: EventMsg::Error(ErrorEvent { + message: err.to_string(), + codex_error_info: Some(CodexErrorInfo::Other), + }), + }; + sess.send_event_raw(event).await; + return; + } + }; - let codex_home = sess.codex_home().await; - if let Err(e) = - session_index::append_thread_name(&codex_home, sess.conversation_id, &name).await + if let Some(state_db) = sess.services.state_db.as_deref() + && let Err(err) = state_db + .update_thread_title(sess.conversation_id, &name) + .await { - let event = Event { - id: sub_id, - msg: EventMsg::Error(ErrorEvent { - message: format!("Failed to set thread name: {e}"), - codex_error_info: Some(CodexErrorInfo::Other), - }), - }; - sess.send_event_raw(event).await; - return; + warn!("Failed to update thread title in state db: {err}"); } { @@ -5596,14 +5606,14 @@ mod handlers { state.session_configuration.thread_name = Some(name.clone()); } - sess.send_event_raw(Event { - id: sub_id, - msg: EventMsg::ThreadNameUpdated(ThreadNameUpdatedEvent { - thread_id: sess.conversation_id, - thread_name: Some(name), - }), - }) - .await; + let codex_home = sess.codex_home().await; + if let Err(err) = + crate::rollout::append_thread_name(&codex_home, sess.conversation_id, &name).await + { + warn!("Failed to update legacy thread name index: {err}"); + } + + sess.deliver_event_raw(Event { id: sub_id, msg }).await; } pub async fn shutdown(sess: &Arc, sub_id: String) -> bool { diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index efae8ae91..db04a0985 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -165,10 +165,10 @@ pub use rollout::append_thread_name; pub use rollout::find_archived_thread_path_by_id_str; #[deprecated(note = "use find_thread_path_by_id_str")] pub use rollout::find_conversation_path_by_id_str; +pub use rollout::find_thread_meta_by_name_str; pub use rollout::find_thread_name_by_id; pub use rollout::find_thread_names_by_ids; pub use rollout::find_thread_path_by_id_str; -pub use rollout::find_thread_path_by_name_str; pub use rollout::parse_cursor; pub use rollout::read_head_for_summary; pub use rollout::read_session_meta_line; diff --git a/codex-rs/core/src/rollout.rs b/codex-rs/core/src/rollout.rs index 492b0bc05..26e3a84d5 100644 --- a/codex-rs/core/src/rollout.rs +++ b/codex-rs/core/src/rollout.rs @@ -14,10 +14,10 @@ pub use codex_rollout::append_thread_name; pub use codex_rollout::find_archived_thread_path_by_id_str; #[deprecated(note = "use find_thread_path_by_id_str")] pub use codex_rollout::find_conversation_path_by_id_str; +pub use codex_rollout::find_thread_meta_by_name_str; pub use codex_rollout::find_thread_name_by_id; pub use codex_rollout::find_thread_names_by_ids; pub use codex_rollout::find_thread_path_by_id_str; -pub use codex_rollout::find_thread_path_by_name_str; pub use codex_rollout::parse_cursor; pub use codex_rollout::read_head_for_summary; pub use codex_rollout::read_session_meta_line; @@ -66,11 +66,6 @@ pub(crate) mod recorder { pub use codex_rollout::RolloutRecorder; } -pub(crate) mod session_index { - pub use codex_rollout::append_thread_name; - pub use codex_rollout::find_thread_name_by_id; -} - pub(crate) use crate::session_rollout_init_error::map_session_init_error; pub(crate) mod truncation { diff --git a/codex-rs/core/tests/suite/rollout_list_find.rs b/codex-rs/core/tests/suite/rollout_list_find.rs index 7445a4ba0..eef0d0f5f 100644 --- a/codex-rs/core/tests/suite/rollout_list_find.rs +++ b/codex-rs/core/tests/suite/rollout_list_find.rs @@ -9,8 +9,8 @@ use codex_core::RolloutRecorder; use codex_core::RolloutRecorderParams; use codex_core::config::ConfigBuilder; use codex_core::find_archived_thread_path_by_id_str; +use codex_core::find_thread_meta_by_name_str; use codex_core::find_thread_path_by_id_str; -use codex_core::find_thread_path_by_name_str; use codex_protocol::ThreadId; use codex_protocol::models::BaseInstructions; use codex_protocol::protocol::SessionSource; @@ -197,9 +197,10 @@ async fn find_locates_rollout_file_written_by_recorder() -> std::io::Result<()> ), )?; - let found = find_thread_path_by_name_str(home.path(), thread_name).await?; + let found = find_thread_meta_by_name_str(home.path(), thread_name).await?; - let path = found.expect("expected rollout path to be found"); + let (path, session_meta) = found.expect("expected rollout path to be found"); + assert_eq!(session_meta.meta.id, thread_id); assert!(path.exists()); let contents = std::fs::read_to_string(&path)?; assert!(contents.contains(&thread_id.to_string())); diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 76634e778..10a5fa1da 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -60,6 +60,7 @@ use codex_core::config::resolve_oss_provider; use codex_core::config_loader::ConfigLoadError; use codex_core::config_loader::LoaderOverrides; use codex_core::config_loader::format_config_error_with_source; +use codex_core::find_thread_meta_by_name_str; use codex_core::format_exec_policy_error_with_source; use codex_core::path_utils; use codex_feedback::CodexFeedback; @@ -1256,6 +1257,27 @@ async fn resolve_resume_thread_id( if Uuid::parse_str(session_id).is_ok() { return Ok(Some(session_id.to_string())); } + if let Some(state_db) = codex_core::get_state_db(config).await { + let cwd = (!args.all).then_some(config.cwd.as_path()); + let resolved = state_db + .find_thread_by_exact_title( + session_id, + &[], + /*model_providers*/ None, + /*archived_only*/ false, + cwd, + ) + .await?; + if let Some(thread) = resolved { + return Ok(Some(thread.id.to_string())); + } + if let Some((_, session_meta)) = + find_thread_meta_by_name_str(&config.codex_home, session_id).await? + && (args.all || cwds_match(config.cwd.as_path(), &session_meta.meta.cwd)) + { + return Ok(Some(session_meta.meta.id.to_string())); + } + } let mut cursor = None; loop { @@ -1271,10 +1293,7 @@ async fn resolve_resume_thread_id( source_kinds: Some(all_thread_source_kinds()), archived: Some(false), cwd: None, - // Thread names are attached separately from rollout titles, so name - // resolution must scan the filtered list client-side instead of relying - // on the backend `search_term` filter. - search_term: None, + search_term: Some(session_id.to_string()), }, }, "thread/list", diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 6acf2c3be..31fcf4724 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -54,10 +54,11 @@ pub use policy::EventPersistenceMode; pub use policy::should_persist_response_item_for_memories; pub use recorder::RolloutRecorder; pub use recorder::RolloutRecorderParams; +pub use recorder::append_rollout_item_to_path; pub use session_index::append_thread_name; +pub use session_index::find_thread_meta_by_name_str; pub use session_index::find_thread_name_by_id; pub use session_index::find_thread_names_by_ids; -pub use session_index::find_thread_path_by_name_str; pub use state_db::StateDbHandle; #[cfg(test)] diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index a4c8913a4..4b50781e7 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -97,6 +97,7 @@ fn event_msg_persistence_mode(ev: &EventMsg) -> Option { | EventMsg::AgentReasoning(_) | EventMsg::AgentReasoningRawContent(_) | EventMsg::TokenCount(_) + | EventMsg::ThreadNameUpdated(_) | EventMsg::ContextCompacted(_) | EventMsg::EnteredReviewMode(_) | EventMsg::ExitedReviewMode(_) @@ -142,7 +143,6 @@ fn event_msg_persistence_mode(ev: &EventMsg) -> Option { | EventMsg::AgentReasoningSectionBreak(_) | EventMsg::RawResponseItem(_) | EventMsg::SessionConfigured(_) - | EventMsg::ThreadNameUpdated(_) | EventMsg::McpToolCallBegin(_) | EventMsg::WebSearchBegin(_) | EventMsg::ExecCommandBegin(_) @@ -188,8 +188,10 @@ fn event_msg_persistence_mode(ev: &EventMsg) -> Option { mod tests { use super::EventPersistenceMode; use super::should_persist_event_msg; + use codex_protocol::ThreadId; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ImageGenerationEndEvent; + use codex_protocol::protocol::ThreadNameUpdatedEvent; #[test] fn persists_image_generation_end_events_in_limited_mode() { @@ -206,4 +208,17 @@ mod tests { EventPersistenceMode::Limited )); } + + #[test] + fn persists_thread_name_updates_in_limited_mode() { + let event = EventMsg::ThreadNameUpdated(ThreadNameUpdatedEvent { + thread_id: ThreadId::new(), + thread_name: Some("saved-session".to_string()), + }); + + assert!(should_persist_event_msg( + &event, + EventPersistenceMode::Limited + )); + } } diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 3554cb379..6098b805e 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -225,6 +225,26 @@ impl RolloutRecorder { search_term: Option<&str>, ) -> std::io::Result { let codex_home = config.codex_home(); + let state_db_ctx = state_db::get_state_db(config).await; + + if search_term.is_some() + && let Some(db_page) = state_db::list_threads_db( + state_db_ctx.as_deref(), + codex_home, + page_size, + cursor, + sort_key, + allowed_sources, + model_providers, + archived, + search_term, + ) + .await + && (!db_page.items.is_empty() || cursor.is_some()) + { + return Ok(db_page.into()); + } + // Filesystem-first listing intentionally overfetches so we can repair stale/missing // SQLite rollout paths before the final DB-backed page is returned. let fs_page_size = page_size.saturating_mul(2).max(page_size); @@ -256,7 +276,6 @@ impl RolloutRecorder { .await? }; - let state_db_ctx = state_db::get_state_db(config).await; if state_db_ctx.is_none() { // Keep legacy behavior when SQLite is unavailable: return filesystem results // at the requested page size. @@ -951,6 +970,23 @@ async fn sync_thread_state_after_write( .await; } +/// Append one already-filtered rollout item to an existing rollout JSONL file. +/// +/// This is for metadata updates to unloaded threads. Live sessions should use +/// `RolloutRecorder::record_items` so rollout and SQLite updates remain ordered +/// with the rest of the session stream. +pub async fn append_rollout_item_to_path( + rollout_path: &Path, + item: &RolloutItem, +) -> std::io::Result<()> { + let file = tokio::fs::OpenOptions::new() + .append(true) + .open(rollout_path) + .await?; + let mut writer = JsonlWriter { file }; + writer.write_rollout_item(item).await +} + struct JsonlWriter { file: tokio::fs::File, } diff --git a/codex-rs/rollout/src/session_index.rs b/codex-rs/rollout/src/session_index.rs index f1acdd812..039ce27f2 100644 --- a/codex-rs/rollout/src/session_index.rs +++ b/codex-rs/rollout/src/session_index.rs @@ -8,6 +8,7 @@ use std::path::Path; use std::path::PathBuf; use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionMetaLine; use serde::Deserialize; use serde::Serialize; use tokio::io::AsyncBufReadExt; @@ -111,12 +112,12 @@ pub async fn find_thread_names_by_ids( Ok(names) } -/// Locate a recorded thread rollout file by thread name using newest-first ordering. -/// Returns `Ok(Some(path))` if found, `Ok(None)` if not present. -pub async fn find_thread_path_by_name_str( +/// Locate a recorded thread rollout and read its session metadata by thread name. +/// Returns the newest indexed name that still has a readable rollout header. +pub async fn find_thread_meta_by_name_str( codex_home: &Path, name: &str, -) -> std::io::Result> { +) -> std::io::Result> { if name.trim().is_empty() { return Ok(None); } @@ -136,11 +137,11 @@ pub async fn find_thread_path_by_name_str( // rename cannot shadow an older persisted session with the same name. if let Some(path) = super::list::find_thread_path_by_id_str(codex_home, &thread_id.to_string()).await? - && super::list::read_session_meta_line(&path).await.is_ok() + && let Ok(session_meta) = super::list::read_session_meta_line(&path).await { drop(rx); scan.await.map_err(std::io::Error::other)??; - return Ok(Some(path)); + return Ok(Some((path, session_meta))); } } scan.await.map_err(std::io::Error::other)??; diff --git a/codex-rs/rollout/src/session_index_tests.rs b/codex-rs/rollout/src/session_index_tests.rs index 9deba70b5..c6a539fb2 100644 --- a/codex-rs/rollout/src/session_index_tests.rs +++ b/codex-rs/rollout/src/session_index_tests.rs @@ -73,7 +73,7 @@ fn find_thread_id_by_name_prefers_latest_entry() -> std::io::Result<()> { } #[tokio::test] -async fn find_thread_path_by_name_str_skips_newest_entry_without_rollout() -> std::io::Result<()> { +async fn find_thread_meta_by_name_str_skips_newest_entry_without_rollout() -> std::io::Result<()> { // A newer unsaved name entry should not shadow an older persisted rollout with the same name. let temp = TempDir::new()?; let path = session_index_path(temp.path()); @@ -99,14 +99,17 @@ async fn find_thread_path_by_name_str_skips_newest_entry_without_rollout() -> st ]; write_index(&path, &lines)?; - let found = find_thread_path_by_name_str(temp.path(), "same").await?; + let found = find_thread_meta_by_name_str(temp.path(), "same").await?; - assert_eq!(found, Some(saved_rollout_path)); + assert_eq!( + found.map(|(path, session_meta)| (path, session_meta.meta.id)), + Some((saved_rollout_path, saved_id)) + ); Ok(()) } #[tokio::test] -async fn find_thread_path_by_name_str_skips_partial_rollout() -> std::io::Result<()> { +async fn find_thread_meta_by_name_str_skips_partial_rollout() -> std::io::Result<()> { let temp = TempDir::new()?; let path = session_index_path(temp.path()); let saved_id = ThreadId::new(); @@ -133,14 +136,14 @@ async fn find_thread_path_by_name_str_skips_partial_rollout() -> std::io::Result ]; write_index(&path, &lines)?; - let found = find_thread_path_by_name_str(temp.path(), "same").await?; + let found = find_thread_meta_by_name_str(temp.path(), "same").await?; - assert_eq!(found, Some(saved_rollout_path)); + assert_eq!(found.map(|(path, _)| path), Some(saved_rollout_path)); Ok(()) } #[tokio::test] -async fn find_thread_path_by_name_str_ignores_historical_name_after_rename() -> std::io::Result<()> +async fn find_thread_meta_by_name_str_ignores_historical_name_after_rename() -> std::io::Result<()> { let temp = TempDir::new()?; let path = session_index_path(temp.path()); @@ -171,9 +174,9 @@ async fn find_thread_path_by_name_str_ignores_historical_name_after_rename() -> ]; write_index(&path, &lines)?; - let found = find_thread_path_by_name_str(temp.path(), "same").await?; + let found = find_thread_meta_by_name_str(temp.path(), "same").await?; - assert_eq!(found, Some(current_rollout_path)); + assert_eq!(found.map(|(path, _)| path), Some(current_rollout_path)); Ok(()) } diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index 8d35d393a..91c6755ff 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -33,7 +33,9 @@ pub fn apply_rollout_item( pub fn rollout_item_affects_thread_metadata(item: &RolloutItem) -> bool { match item { RolloutItem::SessionMeta(_) | RolloutItem::TurnContext(_) => true, - RolloutItem::EventMsg(EventMsg::TokenCount(_) | EventMsg::UserMessage(_)) => true, + RolloutItem::EventMsg( + EventMsg::TokenCount(_) | EventMsg::UserMessage(_) | EventMsg::ThreadNameUpdated(_), + ) => true, RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) | RolloutItem::Compacted(_) => { false } @@ -95,13 +97,18 @@ fn apply_event_msg(metadata: &mut ThreadMetadata, event: &EventMsg) { } } } + EventMsg::ThreadNameUpdated(updated) => { + if let Some(title) = updated.thread_name.as_deref() + && !title.trim().is_empty() + { + metadata.title = title.trim().to_string(); + } + } _ => {} } } -fn apply_response_item(_metadata: &mut ThreadMetadata, _item: &ResponseItem) { - // Title and first_user_message are derived from EventMsg::UserMessage only. -} +fn apply_response_item(_metadata: &mut ThreadMetadata, _item: &ResponseItem) {} fn strip_user_message_prefix(text: &str) -> &str { match text.find(USER_MESSAGE_BEGIN) { @@ -152,6 +159,7 @@ mod tests { use codex_protocol::protocol::SessionMeta; use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::SessionSource; + use codex_protocol::protocol::ThreadNameUpdatedEvent; use codex_protocol::protocol::TurnContextItem; use codex_protocol::protocol::USER_MESSAGE_BEGIN; use codex_protocol::protocol::UserMessageEvent; @@ -198,6 +206,25 @@ mod tests { assert_eq!(metadata.title, "actual user request"); } + #[test] + fn thread_name_update_replaces_title_without_changing_first_user_message() { + let mut metadata = metadata_for_test(); + metadata.title = "actual user request".to_string(); + metadata.first_user_message = Some("actual user request".to_string()); + let item = RolloutItem::EventMsg(EventMsg::ThreadNameUpdated(ThreadNameUpdatedEvent { + thread_id: metadata.id, + thread_name: Some("saved-session".to_string()), + })); + + apply_rollout_item(&mut metadata, &item, "test-provider"); + + assert_eq!( + metadata.first_user_message.as_deref(), + Some("actual user request") + ); + assert_eq!(metadata.title, "saved-session"); + } + #[test] fn event_msg_image_only_user_message_sets_image_placeholder_preview() { let mut metadata = metadata_for_test(); diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 09b23a431..e56ca3f38 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -326,6 +326,66 @@ ON CONFLICT(child_thread_id) DO NOTHING .map(PathBuf::from)) } + /// Find the newest thread whose user-facing title exactly matches `title`. + #[allow(clippy::too_many_arguments)] + pub async fn find_thread_by_exact_title( + &self, + title: &str, + allowed_sources: &[String], + model_providers: Option<&[String]>, + archived_only: bool, + cwd: Option<&Path>, + ) -> anyhow::Result> { + let mut builder = QueryBuilder::::new( + r#" +SELECT + id, + rollout_path, + created_at, + updated_at, + source, + agent_nickname, + agent_role, + agent_path, + model_provider, + model, + reasoning_effort, + cwd, + cli_version, + title, + sandbox_policy, + approval_mode, + tokens_used, + first_user_message, + archived_at, + git_sha, + git_branch, + git_origin_url +FROM threads + "#, + ); + push_thread_filters( + &mut builder, + archived_only, + allowed_sources, + model_providers, + /*anchor*/ None, + crate::SortKey::UpdatedAt, + /*search_term*/ None, + ); + builder.push(" AND title = "); + builder.push_bind(title); + if let Some(cwd) = cwd { + builder.push(" AND cwd = "); + builder.push_bind(cwd.display().to_string()); + } + push_thread_order_and_limit(&mut builder, crate::SortKey::UpdatedAt, /*limit*/ 1); + + let row = builder.build().fetch_optional(self.pool.as_ref()).await?; + row.map(|row| ThreadRow::try_from_row(&row).and_then(crate::ThreadMetadata::try_from)) + .transpose() + } + /// List threads using the underlying database. #[allow(clippy::too_many_arguments)] pub async fn list_threads( @@ -521,6 +581,19 @@ ON CONFLICT(id) DO NOTHING Ok(result.rows_affected() > 0) } + pub async fn update_thread_title( + &self, + thread_id: ThreadId, + title: &str, + ) -> anyhow::Result { + let result = sqlx::query("UPDATE threads SET title = ? WHERE id = ?") + .bind(title) + .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, diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index 6986b9e59..25c536b66 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -4219,7 +4219,13 @@ impl App { tui.frame_requester().schedule_frame(); } AppEvent::ResumeSessionByIdOrName(id_or_name) => { - match crate::lookup_session_target_with_app_server(app_server, &id_or_name).await? { + match crate::lookup_session_target_with_app_server( + app_server, + self.config.codex_home.as_path(), + &id_or_name, + ) + .await? + { Some(target_session) => { return self .resume_target_session(tui, app_server, target_session) diff --git a/codex-rs/tui/src/lib.rs b/codex-rs/tui/src/lib.rs index facae979f..eb46ba2c9 100644 --- a/codex-rs/tui/src/lib.rs +++ b/codex-rs/tui/src/lib.rs @@ -33,6 +33,7 @@ use codex_core::config_loader::CloudRequirementsLoader; use codex_core::config_loader::ConfigLoadError; use codex_core::config_loader::LoaderOverrides; use codex_core::config_loader::format_config_error_with_source; +use codex_core::find_thread_meta_by_name_str; use codex_core::format_exec_policy_error_with_source; use codex_core::path_utils; use codex_core::read_session_meta_line; @@ -492,6 +493,7 @@ fn session_target_from_app_server_thread( async fn lookup_session_target_by_name_with_app_server( app_server: &mut AppServerSession, + codex_home: &Path, name: &str, ) -> color_eyre::Result> { let mut cursor = None; @@ -505,10 +507,7 @@ async fn lookup_session_target_by_name_with_app_server( source_kinds: Some(vec![ThreadSourceKind::Cli, ThreadSourceKind::VsCode]), archived: Some(false), cwd: None, - // Thread names are hydrated after `thread/list` resolves rollout metadata, so - // name-based resume must scan the filtered list client-side instead of relying on - // the backend search index. - search_term: None, + search_term: Some(name.to_string()), }) .await?; if let Some(thread) = response @@ -519,7 +518,15 @@ async fn lookup_session_target_by_name_with_app_server( return Ok(session_target_from_app_server_thread(thread)); } if response.next_cursor.is_none() { - return Ok(None); + if app_server.is_remote() { + return Ok(None); + } + return Ok(find_thread_meta_by_name_str(codex_home, name).await?.map( + |(path, session_meta)| resume_picker::SessionTarget { + path: Some(path), + thread_id: session_meta.meta.id, + }, + )); } cursor = response.next_cursor; } @@ -527,6 +534,7 @@ async fn lookup_session_target_by_name_with_app_server( async fn lookup_session_target_with_app_server( app_server: &mut AppServerSession, + codex_home: &Path, id_or_name: &str, ) -> color_eyre::Result> { if Uuid::parse_str(id_or_name).is_ok() { @@ -557,7 +565,7 @@ async fn lookup_session_target_with_app_server( }; } - lookup_session_target_by_name_with_app_server(app_server, id_or_name).await + lookup_session_target_by_name_with_app_server(app_server, codex_home, id_or_name).await } async fn lookup_latest_session_target_with_app_server( @@ -1163,7 +1171,13 @@ async fn run_ratatui_app( let Some(startup_app_server) = app_server.as_mut() else { unreachable!("app server should be initialized for --fork "); }; - match lookup_session_target_with_app_server(startup_app_server, id_str).await? { + match lookup_session_target_with_app_server( + startup_app_server, + config.codex_home.as_path(), + id_str, + ) + .await? + { Some(target_session) => resume_picker::SessionSelection::Fork(target_session), None => { shutdown_app_server_if_present(app_server.take()).await; @@ -1224,7 +1238,13 @@ async fn run_ratatui_app( let Some(startup_app_server) = app_server.as_mut() else { unreachable!("app server should be initialized for --resume "); }; - match lookup_session_target_with_app_server(startup_app_server, id_str).await? { + match lookup_session_target_with_app_server( + startup_app_server, + config.codex_home.as_path(), + id_str, + ) + .await? + { Some(target_session) => resume_picker::SessionSelection::Resume(target_session), None => { shutdown_app_server_if_present(app_server.take()).await; @@ -1997,8 +2017,7 @@ mod tests { } #[tokio::test] - async fn lookup_session_target_by_name_ignores_backend_search_term_mismatch() - -> color_eyre::Result<()> { + async fn lookup_session_target_by_name_uses_backend_title_search() -> color_eyre::Result<()> { let temp_dir = TempDir::new()?; let config = build_config(&temp_dir).await?; let thread_id = ThreadId::new(); @@ -2034,22 +2053,76 @@ mod tests { ); builder.cwd = session_cwd; let mut metadata = builder.build(config.model_provider_id.as_str()); - metadata.title = "Different rollout title".to_string(); + metadata.title = "saved-session".to_string(); metadata.first_user_message = Some("preview text".to_string()); state_runtime .upsert_thread(&metadata) .await .map_err(std::io::Error::other)?; - codex_core::append_thread_name(&config.codex_home, thread_id, "saved-session").await?; + let mut app_server = + AppServerSession::new(codex_app_server_client::AppServerClient::InProcess( + start_test_embedded_app_server(config).await?, + )); + let target = lookup_session_target_by_name_with_app_server( + &mut app_server, + temp_dir.path(), + "saved-session", + ) + .await?; + let target = target.expect("name lookup should find the saved thread"); + assert_eq!(target.path, Some(rollout_path)); + assert_eq!(target.thread_id, thread_id); + + app_server.shutdown().await?; + Ok(()) + } + + #[tokio::test] + async fn lookup_session_target_by_name_falls_back_to_legacy_index() -> color_eyre::Result<()> { + let temp_dir = TempDir::new()?; + let config = build_config(&temp_dir).await?; + let thread_id = ThreadId::new(); + let rollout_path = temp_dir + .path() + .join("sessions/2025/02/01") + .join(format!("rollout-2025-02-01T10-00-00-{thread_id}.jsonl")); + std::fs::create_dir_all(rollout_path.parent().expect("rollout parent"))?; + let session_meta = SessionMeta { + id: thread_id, + timestamp: "2025-02-01T10:00:00Z".to_string(), + model_provider: Some(config.model_provider_id.clone()), + ..SessionMeta::default() + }; + let line = RolloutLine { + timestamp: session_meta.timestamp.clone(), + item: RolloutItem::SessionMeta(SessionMetaLine { + meta: session_meta, + git: None, + }), + }; + std::fs::write( + &rollout_path, + format!("{}\n", serde_json::to_string(&line)?), + )?; + std::fs::write( + temp_dir.path().join("session_index.jsonl"), + format!( + "{{\"id\":\"{thread_id}\",\"thread_name\":\"hello\",\"updated_at\":\"2025-02-02T10:00:00Z\"}}\n" + ), + )?; let mut app_server = AppServerSession::new(codex_app_server_client::AppServerClient::InProcess( start_test_embedded_app_server(config).await?, )); - let target = - lookup_session_target_by_name_with_app_server(&mut app_server, "saved-session").await?; - let target = target.expect("name lookup should find the saved thread"); + let target = lookup_session_target_by_name_with_app_server( + &mut app_server, + temp_dir.path(), + "hello", + ) + .await?; + let target = target.expect("legacy name lookup should find the saved thread"); assert_eq!(target.path, Some(rollout_path)); assert_eq!(target.thread_id, thread_id);