agent/codex
1
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

chore: merge name and title (#17116)

Browse Source 
Merge title and name concept to leverage the sqlite title column and
have more efficient queries

---------

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
jif-oai
2026-04-09 18:44:26 +01:00
committed by GitHub
Unverified
parent c0b5d8d24a
commit 12f0e0b0eb
16 changed files with 539 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+112 -30
View File
@@ -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<CodexThread>,
codex_home: &Path,
_codex_home: &Path,
thread_state_manager: &ThreadStateManager,
thread_state: &Arc<Mutex<ThreadState>>,
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<ConversationSummary> {
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<String> {
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<ThreadId>,
) -> HashMap<ThreadId, String> {
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<StateDbHandle> {
StateRuntime::init(config.sqlite_home.clone(), config.model_provider_id.clone())
.await
.ok()
}
fn non_empty_title(metadata: &ThreadMetadata) -> Option<String> {
let title = metadata.title.trim();
(!title.is_empty()).then(|| title.to_string())
}
fn distinct_title(metadata: &ThreadMetadata) -> Option<String> {
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)
}
codex-rs/app-server/tests/suite/v2/thread_name_websocket.rs
+52
View File
@@ -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<usize> {
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::<RolloutLine>(line).ok())
.filter(|line| {
matches!(
line.item,
RolloutItem::EventMsg(EventMsg::ThreadNameUpdated(_))
)
})
.count())
}
codex-rs/core/src/codex.rs
+73 -63
View File
@@ -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<String> {
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<Session>,
event: ThreadNameUpdatedEvent,
) -> anyhow::Result<EventMsg> {
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<Session>, 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<Session>, sub_id: String) -> bool {
codex-rs/core/src/lib.rs
+1 -1
View File
@@ -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;
codex-rs/core/src/rollout.rs
+1 -6
View File
@@ -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 {
codex-rs/core/tests/suite/rollout_list_find.rs
+4 -3
View File
@@ -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()));
codex-rs/exec/src/lib.rs
+23 -4
View File
@@ -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",
codex-rs/rollout/src/lib.rs
+2 -1
View File
@@ -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)]
codex-rs/rollout/src/policy.rs
+16 -1
View File
@@ -97,6 +97,7 @@ fn event_msg_persistence_mode(ev: &EventMsg) -> Option<EventPersistenceMode> {
| 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<EventPersistenceMode> {
| 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<EventPersistenceMode> {
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
));
}
}
codex-rs/rollout/src/recorder.rs
+37 -1
View File
@@ -225,6 +225,26 @@ impl RolloutRecorder {
search_term: Option<&str>,
) -> std::io::Result<ThreadsPage> {
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,
}
codex-rs/rollout/src/session_index.rs
+7 -6
View File
@@ -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<Option<PathBuf>> {
) -> std::io::Result<Option<(PathBuf, SessionMetaLine)>> {
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)??;
codex-rs/rollout/src/session_index_tests.rs
+12 -9
View File
@@ -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(())
}
codex-rs/state/src/extract.rs
+31 -4
View File
@@ -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();
codex-rs/state/src/runtime/threads.rs
+73
View File
@@ -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<Option<crate::ThreadMetadata>> {
let mut builder = QueryBuilder::<Sqlite>::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<bool> {
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,
codex-rs/tui/src/app.rs
+7 -1
View File
@@ -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)
codex-rs/tui/src/lib.rs
+88 -15
View File
@@ -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<Option<resume_picker::SessionTarget>> {
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<Option<resume_picker::SessionTarget>> {
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 <id>");
};
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 <id>");
};
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);