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 a42a2a91c7..00c1ded53b 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -1273,7 +1273,7 @@ async fn paginated_thread_name_set_is_reflected_in_read_list_and_metadata_resume .await?; // Set a user-facing thread title. - let new_name = "Saved user message"; + let new_name = "Custom saved name"; let set_id = mcp .send_thread_set_name_request(ThreadSetNameParams { thread_id: conversation_id.clone(), diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index d2105f5613..9cc5318341 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -77,6 +77,7 @@ use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ReasoningEffort; use codex_protocol::protocol::AgentMessageEvent; +use codex_protocol::protocol::CompactedItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ImageGenerationEndEvent; use codex_protocol::protocol::McpInvocation; @@ -135,7 +136,7 @@ const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals."; #[tokio::test] -async fn thread_resume_paginated_metadata_only_uses_model_context() -> Result<()> { +async fn thread_resume_paginated_model_context_preserves_original_metadata() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; let codex_home = TempDir::new()?; mock_responses_config(&server.uri()).write(codex_home.path())?; @@ -147,6 +148,19 @@ async fn thread_resume_paginated_metadata_only_uses_model_context() -> Result<() Some("mock_provider"), /*git_info*/ None, )?; + let path = rollout_path(codex_home.path(), "2025-01-05T12-00-00", &conversation_id); + append_rollout_item_to_path( + &path, + &RolloutItem::Compacted(CompactedItem { + message: "compacted history".to_string(), + replacement_history: Some(Vec::new()), + window_number: Some(1), + first_window_id: None, + previous_window_id: None, + window_id: None, + }), + ) + .await?; let mut primary = TestAppServer::builder() .with_codex_home(codex_home.path()) @@ -166,7 +180,76 @@ async fn thread_resume_paginated_metadata_only_uses_model_context() -> Result<() } = timeout(DEFAULT_READ_TIMEOUT, primary.read_response(resume_id)).await??; assert_eq!(resumed.id, conversation_id); assert_eq!(resumed.history_mode, ThreadHistoryMode::Paginated); + assert_eq!(resumed.preview, "Saved user message"); assert!(resumed.turns.is_empty()); + + timeout( + DEFAULT_READ_TIMEOUT, + primary.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: conversation_id.clone(), + input: vec![UserInput::Text { + text: "bounded suffix user message".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }), + ) + .await??; + timeout(DEFAULT_READ_TIMEOUT, primary.shutdown_gracefully()).await??; + + let mut secondary = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build_initialized() + .await?; + let resume_id = secondary + .send_thread_resume_request(ThreadResumeParams { + thread_id: conversation_id.clone(), + path: Some(path), + exclude_turns: true, + ..Default::default() + }) + .await?; + let ThreadResumeResponse { + thread: resumed, .. + } = timeout(DEFAULT_READ_TIMEOUT, secondary.read_response(resume_id)).await??; + assert_eq!(resumed.preview, "Saved user message"); + assert!(resumed.turns.is_empty()); + + timeout( + DEFAULT_READ_TIMEOUT, + secondary.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: conversation_id.clone(), + input: vec![UserInput::Text { + text: "resumed user message".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }), + ) + .await??; + + let state_db = StateRuntime::init( + codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()), + "mock_provider".into(), + ) + .await?; + let metadata = state_db + .get_thread(ThreadId::from_string(&conversation_id)?) + .await? + .expect("thread metadata should exist"); + assert_eq!( + ( + metadata.preview.as_deref(), + metadata.title.as_str(), + metadata.first_user_message.as_deref(), + ), + ( + Some("Saved user message"), + "Saved user message", + Some("Saved user message"), + ) + ); Ok(()) } diff --git a/codex-rs/thread-store/src/live_thread.rs b/codex-rs/thread-store/src/live_thread.rs index f2dc3cfeb0..509671d80d 100644 --- a/codex-rs/thread-store/src/live_thread.rs +++ b/codex-rs/thread-store/src/live_thread.rs @@ -151,7 +151,20 @@ impl LiveThread { let thread_id = params.thread_id; let should_load_history = params.history.is_none(); let include_archived = params.include_archived; - let mut metadata_sync = ThreadMetadataSync::for_resume(¶ms); + let metadata = if history_mode == ThreadHistoryMode::Paginated + && let Some(local_store) = thread_store.as_any().downcast_ref::() + && let Some(state_db) = local_store.state_db().await + { + state_db + .get_thread(thread_id) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to read thread metadata for {thread_id}: {err}"), + })? + } else { + None + }; + let mut metadata_sync = ThreadMetadataSync::for_resume(¶ms, metadata.as_ref()); thread_store.resume_thread(params).await?; if should_load_history { match thread_store diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 6f25492e9a..0249417eea 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -674,6 +674,89 @@ mod tests { assert_eq!(metadata.title, "observed append"); } + #[tokio::test] + async fn paginated_resume_prefers_explicit_rollout_path_over_stale_sqlite_path() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let runtime = codex_state::StateRuntime::init( + config.sqlite.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = Arc::new(LocalThreadStore::new(config, Some(runtime.clone()))); + let uuid = 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("paginated session file"); + let stale_rollout_path = home.path().join("stale-rollout.jsonl"); + tokio::fs::write(&stale_rollout_path, "malformed session metadata\n") + .await + .expect("write stale rollout"); + let mut builder = codex_state::ThreadMetadataBuilder::new( + thread_id, + stale_rollout_path, + chrono::Utc::now(), + SessionSource::Cli, + ); + builder.history_mode = ThreadHistoryMode::Paginated; + builder.cwd = home.path().to_path_buf(); + let mut metadata = builder.build("test-provider"); + metadata.preview = Some("original user message".to_string()); + metadata.first_user_message = Some("original user message".to_string()); + metadata.title = "original user message".to_string(); + runtime + .upsert_thread(&metadata) + .await + .expect("update stale sqlite rollout path"); + + let resumed = LiveThread::resume( + store, + ThreadHistoryMode::Paginated, + ResumeThreadParams { + thread_id, + rollout_path: Some(rollout_path.clone()), + history: Some(Arc::new(vec![user_message_item("bounded suffix")])), + include_archived: false, + metadata: ThreadPersistenceMetadata { + cwd: Some(home.path().to_path_buf()), + model_provider: "test-provider".to_string(), + memory_mode: ThreadMemoryMode::Enabled, + }, + }, + ) + .await + .expect("resume paginated thread from its requested rollout"); + assert_eq!( + resumed.local_rollout_path().await.expect("live rollout"), + Some(rollout_path) + ); + resumed.shutdown().await.expect("shutdown resumed writer"); + + let metadata = runtime + .get_thread(thread_id) + .await + .expect("sqlite metadata read") + .expect("sqlite metadata"); + assert_eq!( + ( + metadata.preview.as_deref(), + metadata.title.as_str(), + metadata.first_user_message.as_deref(), + ), + ( + Some("original user message"), + "original user message", + Some("original user message"), + ) + ); + } + #[tokio::test] async fn live_thread_does_not_derive_metadata_from_inherited_items() { 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 dc07f71e99..30aded9ff7 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -51,7 +51,10 @@ pub(super) async fn read_thread( { let metadata_sandbox_policy = metadata.sandbox_policy.clone(); let mut thread = stored_thread_from_sqlite_metadata(store, metadata).await?; - if !params.include_history + // Paginated history may contain only a suffix, so its display metadata lives in SQLite. + // Legacy display metadata remains rollout-derived. + if thread.history_mode == ThreadHistoryMode::Legacy + && !params.include_history && let Some(rollout_path) = thread.rollout_path.clone() && let Ok(mut rollout_thread) = read_thread_from_rollout_path(store, rollout_path).await && rollout_thread.thread_id == thread_id @@ -115,27 +118,21 @@ pub(super) async fn read_thread_by_rollout_path( include_history: bool, ) -> ThreadStoreResult { let path = resolve_requested_rollout_path(store, rollout_path).await?; - let mut thread = read_thread_from_rollout_path(store, path).await?; + let mut thread = read_thread_from_rollout_path(store, path.clone()).await?; if !include_archived && thread.archived_at.is_some() { return Err(ThreadStoreError::InvalidRequest { message: format!("thread {} is archived", thread.thread_id), }); } - if let Some(metadata) = read_sqlite_metadata(store, thread.thread_id).await { + if let Some(mut 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; - thread.is_pinned = metadata.is_pinned; - thread.git_info = if thread.history_mode == ThreadHistoryMode::Paginated { - // A paginated rollout only has the initial Git tuple. Do not turn an explicit SQLite - // clear back into the stale rollout value while reading by path. - git_info_from_parts( - metadata.git_sha, - metadata.git_branch, - metadata.git_origin_url, - ) + // Paginated display metadata lives in SQLite because rollout history may be partial. + metadata.rollout_path = path; + metadata.archived_at = thread.archived_at; + thread = stored_thread_from_sqlite_metadata(store, metadata).await?; } else { + thread.recency_at = metadata.recency_at; + thread.is_pinned = metadata.is_pinned; let (fallback_sha, fallback_branch, fallback_origin_url) = match thread.git_info.take() { Some(info) => ( @@ -145,12 +142,12 @@ pub(super) async fn read_thread_by_rollout_path( ), None => (None, None, None), }; - git_info_from_parts( + thread.git_info = git_info_from_parts( metadata.git_sha.or(fallback_sha), metadata.git_branch.or(fallback_branch), metadata.git_origin_url.or(fallback_origin_url), - ) - }; + ); + } } reject_paginated_history(&thread, include_history)?; attach_history_if_requested(&mut thread, include_history).await?; @@ -528,9 +525,15 @@ mod tests { use chrono::Utc; use codex_protocol::ThreadId; + use codex_protocol::items::TurnItem; + use codex_protocol::items::UserMessageItem; + use codex_protocol::protocol::EventMsg; + use codex_protocol::protocol::ItemCompletedEvent; + use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SandboxPolicy; use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::ThreadHistoryMode; + use codex_protocol::user_input::UserInput; use codex_state::ThreadMetadataBuilder; use pretty_assertions::assert_eq; use tempfile::TempDir; @@ -803,7 +806,7 @@ mod tests { } #[tokio::test] - async fn read_paginated_thread_uses_sqlite_name_over_legacy_compatibility() { + async fn read_paginated_thread_prefers_sqlite_display_metadata() { let home = TempDir::new().expect("temp dir"); let config = test_config(home.path()); let uuid = Uuid::from_u128(228); @@ -815,6 +818,20 @@ mod tests { ThreadHistoryMode::Paginated, ) .expect("session file"); + codex_rollout::append_rollout_item_to_path( + &rollout_path, + &RolloutItem::EventMsg(EventMsg::ItemCompleted(ItemCompletedEvent { + thread_id, + turn_id: "turn-1".to_string(), + item: TurnItem::UserMessage(UserMessageItem::new(&[UserInput::Text { + text: "Rollout user message".to_string(), + text_elements: Vec::new(), + }])), + completed_at_ms: 0, + })), + ) + .await + .expect("append rollout user message"); let runtime = codex_state::StateRuntime::init( config.sqlite.clone(), config.default_model_provider_id.clone(), @@ -822,14 +839,21 @@ mod tests { .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); + let mut builder = ThreadMetadataBuilder::new( + thread_id, + rollout_path.clone(), + Utc::now(), + SessionSource::Cli, + ); builder.history_mode = ThreadHistoryMode::Paginated; - builder.model_provider = Some(config.default_model_provider_id.clone()); + builder.model_provider = Some("Canonical SQLite provider".to_string()); 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(); + metadata.preview = Some("Canonical SQLite preview".to_string()); + metadata.first_user_message = Some("Canonical SQLite first user message".to_string()); + metadata.title = "Canonical SQLite derived title".to_string(); + metadata.model = Some("Canonical SQLite model".to_string()); runtime .upsert_thread(&metadata) .await @@ -838,7 +862,7 @@ mod tests { .await .expect("append legacy thread name"); - let thread = store + let thread_by_id = store .read_thread(ReadThreadParams { thread_id, include_archived: false, @@ -846,8 +870,33 @@ mod tests { }) .await .expect("read thread"); + let thread_by_path = store + .read_thread_by_rollout_path( + rollout_path, + /*include_archived*/ false, + /*include_history*/ false, + ) + .await + .expect("read thread by rollout path"); - assert_eq!(thread.name.as_deref(), Some("Canonical SQLite name")); + for thread in [thread_by_id, thread_by_path] { + assert_eq!( + ( + thread.name.as_deref(), + thread.preview.as_str(), + thread.first_user_message.as_deref(), + thread.model_provider.as_str(), + thread.model.as_deref(), + ), + ( + Some("Canonical SQLite name"), + "Canonical SQLite preview", + Some("Canonical SQLite first user message"), + "Canonical SQLite provider", + Some("Canonical SQLite model"), + ) + ); + } } #[tokio::test] diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index 2cdd837580..d320ecb5bc 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -16,6 +16,7 @@ use codex_protocol::protocol::ThreadMemoryMode; use codex_protocol::protocol::UserMessageEvent; use codex_protocol::protocol::strip_user_message_prefix; use codex_protocol::protocol::user_message_preview; +use codex_state::ThreadMetadata; use crate::CreateThreadParams; use crate::GitInfoPatch; @@ -90,7 +91,10 @@ impl ThreadMetadataSync { } } - pub(crate) fn for_resume(params: &ResumeThreadParams) -> Self { + pub(crate) fn for_resume( + params: &ResumeThreadParams, + metadata: Option<&ThreadMetadata>, + ) -> Self { let mut sync = Self { thread_id: params.thread_id, cwd_seen: params @@ -98,9 +102,12 @@ impl ThreadMetadataSync { .cwd .as_ref() .is_some_and(|cwd| !cwd.as_os_str().is_empty()), - preview_seen: false, - first_user_message_seen: false, - title_seen: false, + preview_seen: metadata + .and_then(|metadata| metadata.preview.as_deref()) + .is_some_and(|preview| !preview.is_empty()), + first_user_message_seen: metadata + .is_some_and(|metadata| metadata.first_user_message.is_some()), + title_seen: metadata.is_some_and(|metadata| !metadata.title.is_empty()), pending_update: None, pending_update_generation: 0, last_touch_persisted_at: None, @@ -426,13 +433,13 @@ mod tests { #[test] fn resume_history_keeps_derived_metadata_pending_until_applied() { let thread_id = ThreadId::new(); - let mut sync = ThreadMetadataSync::for_resume(&resume_params( + let mut sync = resume_sync( thread_id, vec![ RolloutItem::SessionMeta(session_meta(thread_id)), RolloutItem::EventMsg(EventMsg::UserMessage(user_message("hello metadata"))), ], - )); + ); let update = sync.take_pending_update().expect("pending metadata update"); assert_eq!( @@ -462,7 +469,7 @@ mod tests { #[test] fn goal_update_sets_preview_without_overriding_existing_preview() { let thread_id = ThreadId::new(); - let sync = ThreadMetadataSync::for_resume(&resume_params( + let sync = resume_sync( thread_id, vec![ RolloutItem::EventMsg(EventMsg::ThreadGoalUpdated(goal_update( @@ -471,26 +478,26 @@ mod tests { ))), RolloutItem::EventMsg(EventMsg::UserMessage(user_message("first user text"))), ], - )); + ); let update = sync.take_pending_update().expect("pending metadata update"); assert_eq!(update.patch.preview.as_deref(), Some("ship the refactor")); + assert_eq!(update.patch.title.as_deref(), Some("first user text")); assert_eq!( update.patch.first_user_message.as_deref(), Some("first user text") ); - assert_eq!(update.patch.title.as_deref(), Some("first user text")); } #[test] fn later_user_messages_do_not_emit_existing_preview_fields() { let thread_id = ThreadId::new(); - let mut sync = ThreadMetadataSync::for_resume(&resume_params( + let mut sync = resume_sync( thread_id, vec![RolloutItem::EventMsg(EventMsg::UserMessage(user_message( "first user text", )))], - )); + ); let pending = sync.take_pending_update().expect("pending resume metadata"); sync.mark_pending_update_applied(&pending); @@ -509,7 +516,7 @@ mod tests { #[test] fn completed_user_message_items_emit_metadata_fields() { let thread_id = ThreadId::new(); - let mut sync = ThreadMetadataSync::for_resume(&resume_params(thread_id, Vec::new())); + let mut sync = resume_sync(thread_id, Vec::new()); let update = sync .observe_appended_items(&[RolloutItem::EventMsg(EventMsg::ItemCompleted( ItemCompletedEvent { @@ -536,7 +543,7 @@ mod tests { #[test] fn metadata_irrelevant_items_coalesce_updated_at_touches() { let thread_id = ThreadId::new(); - let mut sync = ThreadMetadataSync::for_resume(&resume_params(thread_id, Vec::new())); + let mut sync = resume_sync(thread_id, Vec::new()); let item = RolloutItem::Compacted(CompactedItem { message: "compacted".to_string(), replacement_history: None, @@ -566,7 +573,7 @@ mod tests { #[test] fn turn_start_advances_recency_at_without_changing_updated_at_behavior() { let thread_id = ThreadId::new(); - let mut sync = ThreadMetadataSync::for_resume(&resume_params(thread_id, Vec::new())); + let mut sync = resume_sync(thread_id, Vec::new()); let update = sync .observe_appended_items(&[RolloutItem::EventMsg(EventMsg::TurnStarted( @@ -587,7 +594,7 @@ mod tests { #[test] fn thread_settings_applied_updates_live_metadata() { let thread_id = ThreadId::new(); - let mut sync = ThreadMetadataSync::for_resume(&resume_params(thread_id, Vec::new())); + let mut sync = resume_sync(thread_id, Vec::new()); let permission_profile = PermissionProfile::workspace_write(); let cwd = std::env::current_dir() .expect("current directory") @@ -655,13 +662,13 @@ mod tests { #[test] fn resume_history_waits_for_append_before_flushing_metadata() { let thread_id = ThreadId::new(); - let mut sync = ThreadMetadataSync::for_resume(&resume_params( + let mut sync = resume_sync( thread_id, vec![ RolloutItem::SessionMeta(session_meta(thread_id)), RolloutItem::EventMsg(EventMsg::UserMessage(user_message("hello metadata"))), ], - )); + ); assert!( sync.take_pending_update_for_existing_history().is_none(), @@ -687,13 +694,13 @@ mod tests { branch: Some("stale-branch".to_string()), repository_url: None, }); - let sync = ThreadMetadataSync::for_resume(&resume_params( + let sync = resume_sync( thread_id, vec![ RolloutItem::SessionMeta(meta), RolloutItem::EventMsg(EventMsg::UserMessage(user_message("hello metadata"))), ], - )); + ); let update = sync.take_pending_update().expect("pending metadata update"); assert_eq!(update.patch.git_info, None); @@ -701,6 +708,40 @@ mod tests { assert_eq!(update.patch.preview.as_deref(), Some("hello metadata")); } + #[test] + fn resume_preserves_existing_display_metadata_and_initializes_missing_title() { + let thread_id = ThreadId::new(); + let mut metadata = codex_state::ThreadMetadataBuilder::new( + thread_id, + Default::default(), + Utc::now(), + SessionSource::Exec, + ) + .build("test-provider"); + metadata.preview = Some("goal-first preview".to_string()); + metadata.first_user_message = Some("[Image]".to_string()); + let params = resume_params( + thread_id, + vec![RolloutItem::EventMsg(EventMsg::UserMessage(user_message( + "first textual user message", + )))], + ); + + let sync = ThreadMetadataSync::for_resume(¶ms, Some(&metadata)); + let update = sync.take_pending_update().expect("pending metadata update"); + + assert_eq!(update.patch.preview, None); + assert_eq!(update.patch.first_user_message, None); + assert_eq!( + update.patch.title.as_deref(), + Some("first textual user message") + ); + } + + fn resume_sync(thread_id: ThreadId, history: Vec) -> ThreadMetadataSync { + ThreadMetadataSync::for_resume(&resume_params(thread_id, history), /*metadata*/ None) + } + fn resume_params(thread_id: ThreadId, history: Vec) -> ResumeThreadParams { ResumeThreadParams { thread_id,