diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 3b21a2e82..04a066e88 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -3817,7 +3817,7 @@ impl CodexMessageProcessor { ) -> Result { let loaded_thread = self.load_live_thread_for_read(thread_id).await; let mut thread = if let Some(thread) = self - .load_persisted_thread_for_read(thread_id, include_turns, loaded_thread.as_ref()) + .load_persisted_thread_for_read(thread_id, include_turns) .await? { thread @@ -3859,73 +3859,38 @@ impl CodexMessageProcessor { &self, thread_id: ThreadId, include_turns: bool, - loaded_thread: Option<&Arc>, ) -> Result, ThreadReadViewError> { - let loaded_thread_state_db = loaded_thread.and_then(|thread| thread.state_db()); - let db_summary = if let Some(state_db_ctx) = loaded_thread_state_db.as_ref() { - read_summary_from_state_db_context_by_thread_id(Some(state_db_ctx), thread_id).await - } else { - read_summary_from_state_db_by_thread_id(&self.config, thread_id).await - }; - let mut rollout_path = db_summary.as_ref().map(|summary| summary.path.clone()); - if rollout_path.is_none() || include_turns { - rollout_path = - match find_thread_path_by_id_str(&self.config.codex_home, &thread_id.to_string()) - .await - { - Ok(Some(path)) => Some(path), - Ok(None) => { - if include_turns { - None - } else { - rollout_path - } - } - Err(err) => { - return Err(ThreadReadViewError::InvalidRequest(format!( - "failed to locate thread id {thread_id}: {err}" - ))); - } - }; - } - - if include_turns && rollout_path.is_none() && db_summary.is_some() { - return Err(ThreadReadViewError::Internal(format!( - "failed to locate rollout for thread {thread_id}" - ))); - } - - if let Some(summary) = db_summary { - let mut thread = summary_to_thread(summary, &self.config.cwd); - self.apply_thread_read_rollout_fields( - thread_id, - &mut thread, - rollout_path.as_deref(), - include_turns, - ) - .await?; - return Ok(Some(thread)); - } - - let Some(rollout_path) = rollout_path else { - return Ok(None); - }; let fallback_provider = self.config.model_provider_id.as_str(); - match read_summary_from_rollout(&rollout_path, fallback_provider).await { - Ok(summary) => { - let mut thread = summary_to_thread(summary, &self.config.cwd); - self.apply_thread_read_rollout_fields( - thread_id, - &mut thread, - Some(rollout_path.as_path()), - include_turns, - ) - .await?; + match self + .thread_store + .read_thread(StoreReadThreadParams { + thread_id, + include_archived: true, + include_history: include_turns, + }) + .await + { + Ok(stored_thread) => { + let (mut thread, history) = + thread_from_stored_thread(stored_thread, fallback_provider, &self.config.cwd); + if include_turns && let Some(history) = history { + thread.turns = build_turns_from_rollout_items(&history.items); + } Ok(Some(thread)) } + Err(ThreadStoreError::InvalidRequest { message }) + if message == format!("no rollout found for thread id {thread_id}") => + { + Ok(None) + } + Err(ThreadStoreError::ThreadNotFound { + thread_id: missing_thread_id, + }) if missing_thread_id == thread_id => Ok(None), + Err(ThreadStoreError::InvalidRequest { message }) => { + Err(ThreadReadViewError::InvalidRequest(message)) + } Err(err) => Err(ThreadReadViewError::Internal(format!( - "failed to load rollout `{}` for thread {thread_id}: {err}", - rollout_path.display() + "failed to read thread: {err}" ))), } } @@ -3946,7 +3911,6 @@ impl CodexMessageProcessor { "ephemeral threads do not support includeTurns".to_string(), )); } - let mut thread = build_thread_from_snapshot(thread_id, &config_snapshot, loaded_rollout_path.clone()); self.apply_thread_read_rollout_fields( @@ -9291,6 +9255,56 @@ fn thread_store_list_error(err: ThreadStoreError) -> JSONRPCErrorError { } } +fn thread_from_stored_thread( + thread: StoredThread, + fallback_provider: &str, + fallback_cwd: &AbsolutePathBuf, +) -> (Thread, Option) { + let path = thread.rollout_path; + let git_info = thread.git_info.map(|info| ApiGitInfo { + sha: info.commit_hash.map(|sha| sha.0), + branch: info.branch, + origin_url: info.repository_url, + }); + let cwd = AbsolutePathBuf::relative_to_current_dir(path_utils::normalize_for_native_workdir( + thread.cwd, + )) + .unwrap_or_else(|err| { + warn!("failed to normalize thread cwd while reading stored thread: {err}"); + fallback_cwd.clone() + }); + let source = with_thread_spawn_agent_metadata( + thread.source, + thread.agent_nickname.clone(), + thread.agent_role.clone(), + ); + let history = thread.history; + let thread = Thread { + id: thread.thread_id.to_string(), + forked_from_id: thread.forked_from_id.map(|id| id.to_string()), + preview: thread.first_user_message.unwrap_or(thread.preview), + ephemeral: false, + model_provider: if thread.model_provider.is_empty() { + fallback_provider.to_string() + } else { + thread.model_provider + }, + created_at: thread.created_at.timestamp(), + updated_at: thread.updated_at.timestamp(), + status: ThreadStatus::NotLoaded, + path, + cwd, + cli_version: thread.cli_version, + agent_nickname: source.get_nickname(), + agent_role: source.get_agent_role(), + source: source.into(), + git_info, + name: thread.name, + turns: Vec::new(), + }; + (thread, history) +} + fn thread_store_archive_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError { match err { ThreadStoreError::InvalidRequest { message } => JSONRPCErrorError { 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 4ab118162..37fd07d56 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -2,6 +2,7 @@ use anyhow::Result; use app_test_support::McpProcess; use app_test_support::create_fake_rollout_with_text_elements; use app_test_support::create_mock_responses_server_repeating_assistant; +use app_test_support::rollout_path; use app_test_support::test_absolute_path; use app_test_support::to_response; use codex_app_server_protocol::JSONRPCError; @@ -27,6 +28,7 @@ use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::UserInput; +use codex_core::ARCHIVED_SESSIONS_SUBDIR; use codex_protocol::user_input::ByteRange; use codex_protocol::user_input::TextElement; use core_test_support::responses; @@ -157,6 +159,54 @@ async fn thread_read_can_include_turns() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_read_can_return_archived_threads_by_id() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + create_config_toml(codex_home.path(), &server.uri())?; + + let filename_ts = "2025-01-05T12-00-00"; + let preview = "Archived saved user message"; + let conversation_id = create_fake_rollout_with_text_elements( + codex_home.path(), + filename_ts, + "2025-01-05T12:00:00Z", + preview, + vec![], + Some("mock_provider"), + /*git_info*/ None, + )?; + let active_rollout_path = rollout_path(codex_home.path(), filename_ts, &conversation_id); + let archived_dir = codex_home.path().join(ARCHIVED_SESSIONS_SUBDIR); + std::fs::create_dir_all(&archived_dir)?; + let archived_rollout_path = + archived_dir.join(active_rollout_path.file_name().expect("rollout file name")); + std::fs::rename(&active_rollout_path, &archived_rollout_path)?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let read_id = mcp + .send_thread_read_request(ThreadReadParams { + thread_id: conversation_id.clone(), + include_turns: false, + }) + .await?; + let read_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(read_id)), + ) + .await??; + let ThreadReadResponse { thread } = to_response::(read_resp)?; + + assert_eq!(thread.id, conversation_id); + assert_eq!(thread.preview, preview); + let path = thread.path.expect("thread path"); + assert_eq!(path.canonicalize()?, archived_rollout_path.canonicalize()?); + + Ok(()) +} + #[tokio::test] async fn thread_read_returns_forked_from_id_for_forked_threads() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/thread-store/Cargo.toml b/codex-rs/thread-store/Cargo.toml index 3698332b9..61e4e4145 100644 --- a/codex-rs/thread-store/Cargo.toml +++ b/codex-rs/thread-store/Cargo.toml @@ -21,6 +21,7 @@ chrono = { workspace = true, features = ["serde"] } codex-git-utils = { workspace = true } codex-protocol = { workspace = true } codex-rollout = { workspace = true } +codex-state = { workspace = true } prost = "0.14.3" serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } @@ -29,7 +30,6 @@ tonic = { workspace = true } tonic-prost = { workspace = true } [dev-dependencies] -codex-state = { workspace = true } pretty_assertions = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/codex-rs/thread-store/src/local/helpers.rs b/codex-rs/thread-store/src/local/helpers.rs index e28ecc661..108218e9e 100644 --- a/codex-rs/thread-store/src/local/helpers.rs +++ b/codex-rs/thread-store/src/local/helpers.rs @@ -139,7 +139,7 @@ fn parse_rfc3339(value: Option<&str>) -> Option> { .map(|dt| dt.with_timezone(&Utc)) } -fn git_info_from_parts( +pub(super) fn git_info_from_parts( sha: Option, branch: Option, origin_url: Option, diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index 189bc1938..33a41336b 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -1,9 +1,20 @@ +use chrono::DateTime; +use chrono::Utc; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::SessionMetaLine; +use codex_protocol::protocol::SessionSource; use codex_rollout::RolloutRecorder; use codex_rollout::find_archived_thread_path_by_id_str; +use codex_rollout::find_thread_name_by_id; use codex_rollout::find_thread_path_by_id_str; +use codex_rollout::read_session_meta_line; use codex_rollout::read_thread_item_from_rollout; +use codex_state::StateRuntime; +use codex_state::ThreadMetadata; use super::LocalThreadStore; +use super::helpers::git_info_from_parts; use super::helpers::stored_thread_from_rollout_item; use crate::ReadThreadParams; use crate::StoredThread; @@ -16,13 +27,53 @@ pub(super) async fn read_thread( params: ReadThreadParams, ) -> ThreadStoreResult { let thread_id = params.thread_id; - let path = if params.include_archived { + if let Some(metadata) = read_sqlite_metadata(store, thread_id).await + && (params.include_archived || metadata.archived_at.is_none()) + { + let mut thread = stored_thread_from_sqlite_metadata(store, metadata).await; + if params.include_history { + let Some(path) = thread.rollout_path.clone() else { + return Err(ThreadStoreError::Internal { + message: format!("failed to locate rollout for thread {thread_id}"), + }); + }; + let items = load_history_items(&path).await?; + thread.history = Some(StoredThreadHistory { thread_id, items }); + } + return Ok(thread); + } + + let path = resolve_rollout_path(store, thread_id, params.include_archived) + .await? + .ok_or_else(|| ThreadStoreError::InvalidRequest { + message: format!("no rollout found for thread id {thread_id}"), + })?; + + let mut thread = read_thread_from_rollout_path(store, thread_id, path).await?; + if params.include_history { + let Some(path) = thread.rollout_path.clone() else { + return Err(ThreadStoreError::Internal { + message: format!("failed to load thread history for thread {thread_id}"), + }); + }; + let items = load_history_items(&path).await?; + thread.history = Some(StoredThreadHistory { thread_id, items }); + } + Ok(thread) +} + +async fn resolve_rollout_path( + store: &LocalThreadStore, + thread_id: codex_protocol::ThreadId, + include_archived: bool, +) -> ThreadStoreResult> { + if include_archived { match find_thread_path_by_id_str(store.config.codex_home.as_path(), &thread_id.to_string()) .await .map_err(|err| ThreadStoreError::InvalidRequest { message: format!("failed to locate thread id {thread_id}: {err}"), })? { - Some(path) => Some(path), + Some(path) => Ok(Some(path)), None => find_archived_thread_path_by_id_str( store.config.codex_home.as_path(), &thread_id.to_string(), @@ -30,25 +81,26 @@ pub(super) async fn read_thread( .await .map_err(|err| ThreadStoreError::InvalidRequest { message: format!("failed to locate archived thread id {thread_id}: {err}"), - })?, + }), } } else { find_thread_path_by_id_str(store.config.codex_home.as_path(), &thread_id.to_string()) .await .map_err(|err| ThreadStoreError::InvalidRequest { message: format!("failed to locate thread id {thread_id}: {err}"), - })? + }) } - .ok_or_else(|| ThreadStoreError::InvalidRequest { - message: format!("no rollout found for thread id {thread_id}"), - })?; +} - let item = read_thread_item_from_rollout(path.clone()) - .await - .ok_or_else(|| ThreadStoreError::Internal { - message: format!("failed to read thread {}", path.display()), - })?; - let archived = item.path.starts_with( +async fn read_thread_from_rollout_path( + store: &LocalThreadStore, + thread_id: codex_protocol::ThreadId, + path: std::path::PathBuf, +) -> ThreadStoreResult { + let Some(item) = read_thread_item_from_rollout(path.clone()).await else { + return stored_thread_from_session_meta(store, path).await; + }; + let archived = path.starts_with( store .config .codex_home @@ -59,20 +111,203 @@ pub(super) async fn read_thread( .ok_or_else(|| ThreadStoreError::Internal { message: format!("failed to read thread id from {}", path.display()), })?; - if params.include_history { - let (items, _, _) = RolloutRecorder::load_rollout_items(path.as_path()) - .await - .map_err(|err| ThreadStoreError::Internal { - message: format!("failed to load thread history {}: {err}", path.display()), - })?; - thread.history = Some(StoredThreadHistory { thread_id, items }); + thread.forked_from_id = read_session_meta_line(path.as_path()) + .await + .ok() + .and_then(|meta_line| meta_line.meta.forked_from_id); + if let Ok(Some(title)) = + find_thread_name_by_id(store.config.codex_home.as_path(), &thread_id).await + { + set_thread_name_from_title(&mut thread, title); } Ok(thread) } +async fn load_history_items( + path: &std::path::Path, +) -> ThreadStoreResult> { + let (items, _, _) = RolloutRecorder::load_rollout_items(path) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to load thread history {}: {err}", path.display()), + })?; + Ok(items) +} + +async fn read_sqlite_metadata( + store: &LocalThreadStore, + thread_id: codex_protocol::ThreadId, +) -> Option { + let runtime = StateRuntime::init( + store.config.sqlite_home.clone(), + store.config.model_provider_id.clone(), + ) + .await + .ok()?; + runtime.get_thread(thread_id).await.ok().flatten() +} + +async fn stored_thread_from_sqlite_metadata( + store: &LocalThreadStore, + metadata: ThreadMetadata, +) -> StoredThread { + let name = match distinct_title(&metadata) { + Some(title) => Some(title), + None => find_thread_name_by_id(store.config.codex_home.as_path(), &metadata.id) + .await + .ok() + .flatten(), + }; + let forked_from_id = read_session_meta_line(metadata.rollout_path.as_path()) + .await + .ok() + .and_then(|meta_line| meta_line.meta.forked_from_id); + StoredThread { + thread_id: metadata.id, + rollout_path: Some(metadata.rollout_path), + forked_from_id, + preview: metadata.first_user_message.clone().unwrap_or_default(), + name, + model_provider: if metadata.model_provider.is_empty() { + store.config.model_provider_id.clone() + } else { + metadata.model_provider + }, + model: metadata.model, + reasoning_effort: metadata.reasoning_effort, + created_at: metadata.created_at, + updated_at: metadata.updated_at, + archived_at: metadata.archived_at, + cwd: metadata.cwd, + cli_version: metadata.cli_version, + source: parse_session_source(&metadata.source), + agent_nickname: metadata.agent_nickname, + agent_role: metadata.agent_role, + agent_path: metadata.agent_path, + git_info: git_info_from_parts( + metadata.git_sha, + metadata.git_branch, + metadata.git_origin_url, + ), + approval_mode: parse_or_default(&metadata.approval_mode, AskForApproval::OnRequest), + sandbox_policy: parse_or_default( + &metadata.sandbox_policy, + SandboxPolicy::new_read_only_policy(), + ), + token_usage: None, + first_user_message: metadata.first_user_message, + history: None, + } +} + +async fn stored_thread_from_session_meta( + store: &LocalThreadStore, + path: std::path::PathBuf, +) -> ThreadStoreResult { + let meta_line = read_session_meta_line(path.as_path()) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to read thread {}: {err}", path.display()), + })?; + let archived = path.starts_with( + store + .config + .codex_home + .join(codex_rollout::ARCHIVED_SESSIONS_SUBDIR), + ); + Ok(stored_thread_from_meta_line( + store, meta_line, path, archived, + )) +} + +fn stored_thread_from_meta_line( + store: &LocalThreadStore, + meta_line: SessionMetaLine, + path: std::path::PathBuf, + archived: bool, +) -> StoredThread { + let created_at = parse_rfc3339_non_optional(&meta_line.meta.timestamp).unwrap_or_else(Utc::now); + let updated_at = std::fs::metadata(path.as_path()) + .ok() + .and_then(|meta| meta.modified().ok()) + .map(DateTime::::from) + .unwrap_or(created_at); + StoredThread { + thread_id: meta_line.meta.id, + rollout_path: Some(path), + forked_from_id: meta_line.meta.forked_from_id, + preview: String::new(), + name: None, + model_provider: meta_line + .meta + .model_provider + .filter(|provider| !provider.is_empty()) + .unwrap_or_else(|| store.config.model_provider_id.clone()), + model: None, + reasoning_effort: None, + created_at, + updated_at, + archived_at: archived.then_some(updated_at), + cwd: meta_line.meta.cwd, + cli_version: meta_line.meta.cli_version, + source: meta_line.meta.source, + agent_nickname: meta_line.meta.agent_nickname, + agent_role: meta_line.meta.agent_role, + agent_path: meta_line.meta.agent_path, + git_info: meta_line.git, + approval_mode: AskForApproval::OnRequest, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + token_usage: None, + first_user_message: None, + history: None, + } +} + +fn distinct_title(metadata: &ThreadMetadata) -> Option { + let title = metadata.title.trim(); + if title.is_empty() || metadata.first_user_message.as_deref().map(str::trim) == Some(title) { + None + } else { + Some(title.to_string()) + } +} + +fn set_thread_name_from_title(thread: &mut StoredThread, title: String) { + if title.trim().is_empty() || thread.preview.trim() == title.trim() { + return; + } + thread.name = Some(title); +} + +fn parse_session_source(source: &str) -> SessionSource { + serde_json::from_str(source) + .or_else(|_| serde_json::from_value(serde_json::Value::String(source.to_string()))) + .unwrap_or(SessionSource::Unknown) +} + +fn parse_or_default(value: &str, default: T) -> T +where + T: serde::de::DeserializeOwned, +{ + serde_json::from_str(value) + .or_else(|_| serde_json::from_value(serde_json::Value::String(value.to_string()))) + .unwrap_or(default) +} + +fn parse_rfc3339_non_optional(value: &str) -> Option> { + DateTime::parse_from_rfc3339(value) + .ok() + .map(|dt| dt.with_timezone(&Utc)) +} + #[cfg(test)] mod tests { + use std::io::Write; + + use chrono::Utc; use codex_protocol::ThreadId; + use codex_protocol::protocol::SessionSource; + use codex_state::ThreadMetadataBuilder; use pretty_assertions::assert_eq; use tempfile::TempDir; use uuid::Uuid; @@ -81,7 +316,9 @@ mod tests { use crate::ThreadStore; use crate::local::LocalThreadStore; use crate::local::test_support::test_config; + use crate::local::test_support::write_archived_session_file; use crate::local::test_support::write_session_file; + use crate::local::test_support::write_session_file_with_fork; #[tokio::test] async fn read_thread_returns_active_rollout_summary() { @@ -111,6 +348,451 @@ mod tests { ); } + #[tokio::test] + async fn read_thread_returns_archived_rollout_when_requested() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let uuid = Uuid::from_u128(207); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let archived_path = write_archived_session_file(home.path(), "2025-01-03T12-00-00", uuid) + .expect("archived session file"); + + let active_only_err = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect_err("active-only read should fail for archived rollout"); + let ThreadStoreError::InvalidRequest { message } = active_only_err else { + panic!("expected invalid request error"); + }; + assert_eq!( + message, + format!("no rollout found for thread id {thread_id}") + ); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: false, + }) + .await + .expect("read archived thread"); + + assert_eq!(thread.thread_id, thread_id); + assert_eq!(thread.rollout_path, Some(archived_path)); + assert!(thread.archived_at.is_some()); + assert_eq!(thread.preview, "Archived user message"); + assert!(thread.history.is_none()); + } + + #[tokio::test] + async fn read_thread_prefers_active_rollout_over_archived() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let uuid = Uuid::from_u128(208); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let active_path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + write_archived_session_file(home.path(), "2025-01-03T12-00-00", uuid) + .expect("archived session file"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: false, + }) + .await + .expect("read thread"); + + assert_eq!(thread.rollout_path, Some(active_path)); + assert_eq!(thread.archived_at, None); + assert_eq!(thread.preview, "Hello from user"); + } + + #[tokio::test] + async fn read_thread_returns_forked_from_id() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let uuid = Uuid::from_u128(209); + let parent_uuid = Uuid::from_u128(210); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let parent_thread_id = + ThreadId::from_string(&parent_uuid.to_string()).expect("valid parent thread id"); + write_session_file_with_fork( + home.path(), + home.path().join("sessions/2025/01/03"), + "2025-01-03T12-00-00", + uuid, + "Forked user message", + Some("test-provider"), + Some(parent_uuid), + ) + .expect("forked session file"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect("read thread"); + + assert_eq!(thread.forked_from_id, Some(parent_thread_id)); + } + + #[tokio::test] + async fn read_thread_applies_sqlite_thread_name() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(212); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let rollout_path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let mut builder = + ThreadMetadataBuilder::new(thread_id, rollout_path, Utc::now(), SessionSource::Cli); + builder.model_provider = Some(config.model_provider_id.clone()); + builder.cwd = home.path().to_path_buf(); + builder.cli_version = Some("test_version".to_string()); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.title = "Saved title".to_string(); + metadata.first_user_message = Some("Hello from user".to_string()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect("read thread"); + + assert_eq!(thread.name, Some("Saved title".to_string())); + } + + #[tokio::test] + async fn read_thread_uses_legacy_thread_name_when_sqlite_title_is_missing() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let uuid = Uuid::from_u128(213); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + codex_rollout::append_thread_name(home.path(), thread_id, "Legacy title") + .await + .expect("append legacy thread name"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect("read thread"); + + assert_eq!(thread.name, Some("Legacy title".to_string())); + } + + #[tokio::test] + async fn read_thread_uses_sqlite_metadata_for_rollout_without_user_preview() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(217); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let day_dir = home.path().join("sessions/2025/01/03"); + std::fs::create_dir_all(&day_dir).expect("sessions dir"); + let rollout_path = day_dir.join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl")); + let mut file = std::fs::File::create(&rollout_path).expect("session file"); + let meta = serde_json::json!({ + "timestamp": "2025-01-03T12-00-00", + "type": "session_meta", + "payload": { + "id": uuid, + "timestamp": "2025-01-03T12-00-00", + "cwd": home.path(), + "originator": "test_originator", + "cli_version": "test_version", + "source": "cli", + "model_provider": "rollout-provider" + }, + }); + writeln!(file, "{meta}").expect("write session meta"); + + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let mut builder = ThreadMetadataBuilder::new( + thread_id, + rollout_path.clone(), + Utc::now(), + SessionSource::Cli, + ); + builder.model_provider = Some("sqlite-provider".to_string()); + builder.cwd = home.path().join("workspace"); + builder.cli_version = Some("sqlite-cli".to_string()); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.title = "Command-only thread".to_string(); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: true, + }) + .await + .expect("read thread"); + + assert_eq!(thread.thread_id, thread_id); + assert_eq!(thread.rollout_path, Some(rollout_path)); + assert_eq!(thread.preview, ""); + assert_eq!(thread.name.as_deref(), Some("Command-only thread")); + assert_eq!(thread.model_provider, "sqlite-provider"); + assert_eq!(thread.cwd, home.path().join("workspace")); + assert_eq!(thread.cli_version, "sqlite-cli"); + let history = thread.history.expect("history should load"); + assert_eq!(history.thread_id, thread_id); + assert_eq!(history.items.len(), 1); + } + + #[tokio::test] + async fn read_thread_uses_session_meta_for_rollout_without_user_preview_or_sqlite_metadata() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path())); + let uuid = Uuid::from_u128(218); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let day_dir = home.path().join("sessions/2025/01/03"); + std::fs::create_dir_all(&day_dir).expect("sessions dir"); + let rollout_path = day_dir.join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl")); + let mut file = std::fs::File::create(&rollout_path).expect("session file"); + let meta = serde_json::json!({ + "timestamp": "2025-01-03T12:00:00Z", + "type": "session_meta", + "payload": { + "id": uuid, + "timestamp": "2025-01-03T12:00:00Z", + "cwd": home.path(), + "originator": "test_originator", + "cli_version": "test_version", + "source": "cli", + "model_provider": "rollout-provider" + }, + }); + writeln!(file, "{meta}").expect("write session meta"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: true, + }) + .await + .expect("read thread"); + + assert_eq!(thread.thread_id, thread_id); + assert_eq!(thread.rollout_path, Some(rollout_path)); + assert_eq!(thread.preview, ""); + assert_eq!(thread.name, None); + assert_eq!(thread.model_provider, "rollout-provider"); + assert_eq!( + thread.created_at, + parse_rfc3339_non_optional("2025-01-03T12:00:00Z").unwrap() + ); + assert!(thread.updated_at >= thread.created_at); + assert_eq!(thread.archived_at, None); + assert_eq!(thread.cwd, home.path()); + assert_eq!(thread.cli_version, "test_version"); + assert_eq!(thread.source, SessionSource::Cli); + let history = thread.history.expect("history should load"); + assert_eq!(history.thread_id, thread_id); + assert_eq!(history.items.len(), 1); + } + + #[tokio::test] + async fn read_thread_falls_back_to_sqlite_summary() { + let home = TempDir::new().expect("temp dir"); + let external = TempDir::new().expect("external temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(214); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let rollout_path = external + .path() + .join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl")); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let mut builder = ThreadMetadataBuilder::new( + thread_id, + rollout_path.clone(), + Utc::now(), + SessionSource::Exec, + ); + builder.model_provider = Some("sqlite-provider".to_string()); + builder.cwd = external.path().join("workspace"); + builder.cli_version = Some("sqlite-cli".to_string()); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.title = "SQLite title".to_string(); + metadata.first_user_message = Some("SQLite preview".to_string()); + metadata.model = Some("sqlite-model".to_string()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect("read thread"); + + assert_eq!(thread.thread_id, thread_id); + assert_eq!(thread.rollout_path, Some(rollout_path)); + assert_eq!(thread.preview, "SQLite preview"); + assert_eq!(thread.first_user_message.as_deref(), Some("SQLite preview")); + assert_eq!(thread.name.as_deref(), Some("SQLite title")); + assert_eq!(thread.model_provider, "sqlite-provider"); + assert_eq!(thread.model.as_deref(), Some("sqlite-model")); + assert_eq!(thread.cwd, external.path().join("workspace")); + assert_eq!(thread.cli_version, "sqlite-cli"); + assert_eq!(thread.source, SessionSource::Exec); + assert_eq!(thread.archived_at, None); + assert!(thread.history.is_none()); + } + + #[tokio::test] + async fn read_thread_sqlite_fallback_respects_include_archived() { + let home = TempDir::new().expect("temp dir"); + let external = TempDir::new().expect("external temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(216); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let rollout_path = external + .path() + .join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl")); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let mut builder = + ThreadMetadataBuilder::new(thread_id, rollout_path, Utc::now(), SessionSource::Cli); + builder.archived_at = Some(Utc::now()); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.first_user_message = Some("Archived SQLite preview".to_string()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let active_only_err = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect_err("active-only read should fail for archived metadata"); + let ThreadStoreError::InvalidRequest { message } = active_only_err else { + panic!("expected invalid request error"); + }; + assert_eq!( + message, + format!("no rollout found for thread id {thread_id}") + ); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: false, + }) + .await + .expect("read archived thread"); + + assert_eq!(thread.thread_id, thread_id); + assert_eq!(thread.preview, "Archived SQLite preview"); + assert!(thread.archived_at.is_some()); + } + + #[tokio::test] + async fn read_thread_sqlite_fallback_loads_archived_history() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let store = LocalThreadStore::new(config.clone()); + let uuid = Uuid::from_u128(219); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let archived_path = write_archived_session_file(home.path(), "2025-01-03T12-00-00", uuid) + .expect("archived session file"); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let mut builder = ThreadMetadataBuilder::new( + thread_id, + archived_path.clone(), + Utc::now(), + SessionSource::Cli, + ); + builder.archived_at = Some(Utc::now()); + let mut metadata = builder.build(config.model_provider_id.as_str()); + metadata.first_user_message = Some("Archived SQLite preview".to_string()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: true, + }) + .await + .expect("read archived thread"); + + assert_eq!(thread.thread_id, thread_id); + assert_eq!(thread.rollout_path, Some(archived_path)); + assert_eq!(thread.preview, "Archived SQLite preview"); + assert!(thread.archived_at.is_some()); + let history = thread.history.expect("history should load"); + assert_eq!(history.thread_id, thread_id); + assert_eq!(history.items.len(), 2); + } + #[tokio::test] async fn read_thread_fails_without_rollout() { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/thread-store/src/local/test_support.rs b/codex-rs/thread-store/src/local/test_support.rs index d8ddcf8dd..4656503a6 100644 --- a/codex-rs/thread-store/src/local/test_support.rs +++ b/codex-rs/thread-store/src/local/test_support.rs @@ -50,6 +50,26 @@ pub(super) fn write_session_file_with( uuid: Uuid, first_user_message: &str, model_provider: Option<&str>, +) -> std::io::Result { + write_session_file_with_fork( + root, + day_dir, + ts, + uuid, + first_user_message, + model_provider, + /*forked_from_id*/ None, + ) +} + +pub(super) fn write_session_file_with_fork( + root: &Path, + day_dir: PathBuf, + ts: &str, + uuid: Uuid, + first_user_message: &str, + model_provider: Option<&str>, + forked_from_id: Option, ) -> std::io::Result { fs::create_dir_all(&day_dir)?; let path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl")); @@ -59,6 +79,7 @@ pub(super) fn write_session_file_with( "type": "session_meta", "payload": { "id": uuid, + "forked_from_id": forked_from_id, "timestamp": ts, "cwd": root, "originator": "test_originator",