diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index e232148ad1..9e7ed84b15 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -2824,8 +2824,10 @@ dependencies = [ "codex-protocol", "codex-rollout", "codex-state", + "pretty_assertions", "serde", "serde_json", + "tempfile", "thiserror 2.0.18", "tokio", ] diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 1d5ba79773..a6f2e2df40 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -198,6 +198,7 @@ use codex_core::ArchiveThreadParams as StoreArchiveThreadParams; use codex_core::CodexThread; use codex_core::ForkSnapshot; use codex_core::GitInfoPatch; +use codex_core::MissingThreadMetadata; use codex_core::NewThread; use codex_core::ReadThreadParams as StoreReadThreadParams; use codex_core::RolloutRecorder; @@ -210,13 +211,12 @@ use codex_core::StoredThread; use codex_core::ThreadConfigSnapshot; use codex_core::ThreadManager; use codex_core::ThreadMetadataPatch; +use codex_core::ThreadMetadataSeed; use codex_core::ThreadOwner; use codex_core::ThreadSortKey as CoreThreadSortKey; use codex_core::ThreadStore; use codex_core::ThreadStoreError; use codex_core::UpdateThreadMetadataParams as StoreUpdateThreadMetadataParams; -use codex_core::WorkspaceRole as CoreWorkspaceRole; -use codex_core::append_thread_name; use codex_core::config::Config; use codex_core::config::ConfigOverrides; use codex_core::config::NetworkProxyAuditMetadata; @@ -232,10 +232,8 @@ use codex_core::exec::ExecCapturePolicy; use codex_core::exec::ExecExpiration; use codex_core::exec::ExecParams; use codex_core::exec_env::create_env; -use codex_core::find_thread_name_by_id; use codex_core::find_thread_path_by_id_str; use codex_core::local_thread_store; -use codex_core::parse_cursor; use codex_core::path_utils; use codex_core::plugins::MarketplaceError; use codex_core::plugins::MarketplacePluginSource; @@ -313,11 +311,9 @@ 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_state::StateRuntime; -use codex_state::ThreadMetadata; -use codex_state::ThreadMetadataBuilder; use codex_state::log_db::LogDbLayer; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_json_to_toml::json_to_toml; @@ -334,7 +330,6 @@ use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; use std::time::Duration; use std::time::Instant; -use std::time::SystemTime; use tokio::sync::Mutex; use tokio::sync::broadcast; use tokio::sync::oneshot; @@ -2982,8 +2977,8 @@ impl CodexMessageProcessor { return; } - let state_db_ctx = open_state_db_for_direct_thread_lookup(&self.config).await; - reconcile_rollout( + let state_db_ctx = get_state_db(&self.config).await; + codex_rollout::state_db::reconcile_rollout( state_db_ctx.as_deref(), rollout_path.as_path(), self.config.model_provider_id.as_str(), @@ -3090,7 +3085,7 @@ impl CodexMessageProcessor { None => None, }; - if let Ok(loaded_thread) = self.thread_manager.get_thread(thread_uuid).await { + let missing = if let Ok(loaded_thread) = self.thread_manager.get_thread(thread_uuid).await { let Some(rollout_path) = loaded_thread.rollout_path() else { self.send_invalid_request_error( request_id, @@ -3099,60 +3094,27 @@ impl CodexMessageProcessor { .await; return; }; - let state_db = match loaded_thread.state_db() { - Some(state_db) => Some(state_db), - None => get_state_db(&self.config).await, - }; - let Some(state_db) = state_db else { - self.send_internal_error( - request_id, - format!("sqlite state db unavailable for thread {thread_uuid}"), - ) - .await; - return; - }; - match state_db.get_thread(thread_uuid).await { - Ok(Some(_)) => {} - Ok(None) => { - let config_snapshot = loaded_thread.config_snapshot().await; - let model_provider = config_snapshot.model_provider_id.clone(); - let mut builder = ThreadMetadataBuilder::new( - thread_uuid, - rollout_path, - Utc::now(), - config_snapshot.session_source.clone(), - ); - builder.model_provider = Some(model_provider.clone()); - builder.cwd = config_snapshot.cwd.clone(); - builder.cli_version = Some(env!("CARGO_PKG_VERSION").to_string()); - builder.sandbox_policy = config_snapshot.sandbox_policy.clone(); - builder.approval_mode = config_snapshot.approval_policy; - let metadata = builder.build(model_provider.as_str()); - if let Err(err) = state_db.insert_thread_if_absent(&metadata).await { - self.send_internal_error( - request_id, - format!("failed to create thread metadata for {thread_uuid}: {err}"), - ) - .await; - return; - } - } - Err(err) => { - self.send_internal_error( - request_id, - format!("failed to load thread metadata for {thread_uuid}: {err}"), - ) - .await; - return; - } - } - } + let config_snapshot = loaded_thread.config_snapshot().await; + MissingThreadMetadata::Create(ThreadMetadataSeed { + rollout_path, + created_at: Utc::now(), + source: config_snapshot.session_source.clone(), + model_provider: config_snapshot.model_provider_id.clone(), + cwd: config_snapshot.cwd.clone(), + cli_version: env!("CARGO_PKG_VERSION").to_string(), + sandbox_policy: config_snapshot.sandbox_policy.clone(), + approval_mode: config_snapshot.approval_policy, + }) + } else { + MissingThreadMetadata::Error + }; let thread_store = local_thread_store(&self.config); let stored_thread = match thread_store .update_thread_metadata(StoreUpdateThreadMetadataParams { thread_id: thread_uuid, owner: ThreadOwner::default(), + missing, patch: ThreadMetadataPatch { name: None, git_info: Some(GitInfoPatch { @@ -3599,13 +3561,19 @@ impl CodexMessageProcessor { } } } - let loaded_thread_state_db = loaded_thread.as_ref().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_uuid).await - } else { - read_summary_from_state_db_by_thread_id(&self.config, thread_uuid).await - }; - let mut rollout_path = db_summary.as_ref().map(|summary| summary.path.clone()); + let thread_store = local_thread_store(&self.config); + let stored_thread = thread_store + .read_thread(StoreReadThreadParams { + thread_id: thread_uuid, + owner: ThreadOwner::default(), + include_archived: true, + include_history: false, + }) + .await + .ok(); + let mut rollout_path = stored_thread + .as_ref() + .and_then(|thread| thread.legacy_path.clone()); if rollout_path.is_none() || include_turns { rollout_path = match find_thread_path_by_id_str(&self.config.codex_home, &thread_uuid.to_string()) @@ -3630,7 +3598,7 @@ impl CodexMessageProcessor { }; } - if include_turns && rollout_path.is_none() && db_summary.is_some() { + if include_turns && rollout_path.is_none() && stored_thread.is_some() { self.send_internal_error( request_id, format!("failed to locate rollout for thread {thread_uuid}"), @@ -3639,8 +3607,8 @@ impl CodexMessageProcessor { return; } - let mut thread = if let Some(summary) = db_summary { - summary_to_thread(summary) + let mut thread = if let Some(stored_thread) = stored_thread { + stored_thread_to_api_thread(stored_thread) } else if let Some(rollout_path) = rollout_path.as_ref() { let fallback_provider = self.config.model_provider_id.as_str(); match read_summary_from_rollout(rollout_path, fallback_provider).await { @@ -4019,16 +3987,20 @@ impl CodexMessageProcessor { thread_history: &InitialHistory, request_overrides: &mut Option>, typesafe_overrides: &mut ConfigOverrides, - ) -> Option { + ) -> Option { let InitialHistory::Resumed(resumed_history) = thread_history else { return None; }; - let state_db_ctx = get_state_db(&self.config).await?; - let persisted_metadata = state_db_ctx - .get_thread(resumed_history.conversation_id) + let thread_store = local_thread_store(&self.config); + let persisted_metadata = thread_store + .read_thread(StoreReadThreadParams { + thread_id: resumed_history.conversation_id, + owner: ThreadOwner::default(), + include_archived: true, + include_history: false, + }) .await - .ok() - .flatten()?; + .ok()?; merge_persisted_resume_metadata(request_overrides, typesafe_overrides, &persisted_metadata); Some(persisted_metadata) } @@ -4298,7 +4270,7 @@ impl CodexMessageProcessor { thread_history: &InitialHistory, rollout_path: &Path, fallback_provider: &str, - persisted_resume_metadata: Option<&ThreadMetadata>, + persisted_resume_metadata: Option<&StoredThread>, ) -> std::result::Result { let thread = match thread_history { InitialHistory::Resumed(resumed) => { @@ -4340,7 +4312,7 @@ impl CodexMessageProcessor { } async fn attach_thread_name(&self, thread_id: ThreadId, thread: &mut Thread) { - if let Some(title) = title_from_state_db(&self.config, thread_id).await { + if let Some(title) = thread_name_from_thread_store(&self.config, thread_id).await { set_thread_name_from_title(thread, title); } } @@ -4404,9 +4376,12 @@ impl CodexMessageProcessor { } }; - let history_cwd = - read_history_cwd_from_state_db(&self.config, source_thread_id, rollout_path.as_path()) - .await; + let history_cwd = read_history_cwd_from_thread_store( + &self.config, + source_thread_id, + rollout_path.as_path(), + ) + .await; // Persist Windows sandbox mode. let mut cli_overrides = cli_overrides.unwrap_or_default(); @@ -4654,7 +4629,7 @@ impl CodexMessageProcessor { ) { if let GetConversationSummaryParams::ThreadId { conversation_id } = ¶ms && let Some(summary) = - read_summary_from_state_db_by_thread_id(&self.config, *conversation_id).await + read_stored_thread_summary_by_thread_id(&self.config, *conversation_id).await { let response = GetConversationSummaryResponse { summary }; self.outgoing.send_response(request_id, response).await; @@ -8315,7 +8290,7 @@ fn collect_resume_override_mismatches( fn merge_persisted_resume_metadata( request_overrides: &mut Option>, typesafe_overrides: &mut ConfigOverrides, - persisted_metadata: &ThreadMetadata, + persisted_metadata: &StoredThread, ) { if has_model_resume_override(request_overrides.as_ref(), typesafe_overrides) { return; @@ -8626,16 +8601,15 @@ async fn derive_config_for_cwd( Ok(config) } -async fn read_history_cwd_from_state_db( +async fn read_history_cwd_from_thread_store( config: &Config, thread_id: Option, rollout_path: &Path, ) -> Option { - if let Some(state_db_ctx) = get_state_db(config).await - && let Some(thread_id) = thread_id - && let Ok(Some(metadata)) = state_db_ctx.get_thread(thread_id).await + if let Some(thread_id) = thread_id + && let Some(thread) = read_stored_thread_by_thread_id(config, thread_id).await { - return Some(metadata.cwd); + return Some(thread.cwd); } match read_session_meta_line(rollout_path).await { @@ -8648,58 +8622,35 @@ async fn read_history_cwd_from_state_db( } } -async fn read_summary_from_state_db_by_thread_id( +async fn read_stored_thread_by_thread_id( + config: &Config, + thread_id: ThreadId, +) -> Option { + let thread_store = local_thread_store(config); + thread_store + .read_thread(StoreReadThreadParams { + thread_id, + owner: ThreadOwner::default(), + include_archived: true, + include_history: false, + }) + .await + .ok() +} + +async fn read_stored_thread_summary_by_thread_id( config: &Config, thread_id: ThreadId, ) -> Option { - 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 -} - -async fn read_summary_from_state_db_context_by_thread_id( - state_db_ctx: Option<&StateDbHandle>, - thread_id: ThreadId, -) -> Option { - let state_db_ctx = state_db_ctx?; - - let metadata = match state_db_ctx.get_thread(thread_id).await { - Ok(Some(metadata)) => metadata, - Ok(None) | Err(_) => return None, - }; - Some(summary_from_thread_metadata(&metadata)) -} - -async fn title_from_state_db(config: &Config, thread_id: ThreadId) -> Option { - if let Some(state_db_ctx) = open_state_db_for_direct_thread_lookup(config).await - && let Some(metadata) = state_db_ctx.get_thread(thread_id).await.ok().flatten() - && let Some(title) = distinct_title(&metadata) - { - return Some(title); - } - find_thread_name_by_id(&config.codex_home, &thread_id) + read_stored_thread_by_thread_id(config, thread_id) .await - .ok() - .flatten() + .map(stored_thread_to_conversation_summary) } -async fn open_state_db_for_direct_thread_lookup(config: &Config) -> Option { - StateRuntime::init(config.sqlite_home.clone(), config.model_provider_id.clone()) +async fn thread_name_from_thread_store(config: &Config, thread_id: ThreadId) -> Option { + read_stored_thread_by_thread_id(config, thread_id) .await - .ok() -} - -fn non_empty_title(metadata: &ThreadMetadata) -> Option { - let title = metadata.title.trim(); - (!title.is_empty()).then(|| title.to_string()) -} - -fn distinct_title(metadata: &ThreadMetadata) -> Option { - let title = non_empty_title(metadata)?; - if metadata.first_user_message.as_deref().map(str::trim) == Some(title.as_str()) { - None - } else { - Some(title) - } + .and_then(|thread| thread.name) } fn set_thread_name_from_title(thread: &mut Thread, title: String) { @@ -8709,74 +8660,31 @@ fn set_thread_name_from_title(thread: &mut Thread, title: String) { thread.name = Some(title); } -#[allow(clippy::too_many_arguments)] -fn summary_from_state_db_metadata( - conversation_id: ThreadId, - path: PathBuf, - first_user_message: Option, - timestamp: String, - updated_at: String, - model_provider: String, - cwd: PathBuf, - cli_version: String, - source: String, - agent_nickname: Option, - agent_role: Option, - git_sha: Option, - git_branch: Option, - git_origin_url: Option, -) -> ConversationSummary { - let preview = first_user_message.unwrap_or_default(); - let source = serde_json::from_str(&source) - .or_else(|_| serde_json::from_value(serde_json::Value::String(source.clone()))) - .unwrap_or(codex_protocol::protocol::SessionSource::Unknown); - let source = with_thread_spawn_agent_metadata(source, agent_nickname, agent_role); - let git_info = if git_sha.is_none() && git_branch.is_none() && git_origin_url.is_none() { - None - } else { - Some(ConversationGitInfo { - sha: git_sha, - branch: git_branch, - origin_url: git_origin_url, - }) - }; +fn stored_thread_to_conversation_summary(thread: StoredThread) -> ConversationSummary { + let source = with_thread_spawn_agent_metadata( + thread.source, + thread.agent_nickname.clone(), + thread.agent_role.clone(), + ); + let git_info = thread.git_info.map(|git_info| ConversationGitInfo { + sha: git_info.commit_hash.map(|sha| sha.0), + branch: git_info.branch, + origin_url: git_info.repository_url, + }); ConversationSummary { - conversation_id, - path, - preview, - timestamp: Some(timestamp), - updated_at: Some(updated_at), - model_provider, - cwd, - cli_version, + conversation_id: thread.thread_id, + path: thread.legacy_path.unwrap_or_default(), + preview: thread.first_user_message.unwrap_or(thread.preview), + timestamp: Some(thread.created_at.to_rfc3339_opts(SecondsFormat::Secs, true)), + updated_at: Some(thread.updated_at.to_rfc3339_opts(SecondsFormat::Secs, true)), + model_provider: thread.model_provider, + cwd: thread.cwd, + cli_version: thread.cli_version, source, git_info, } } -fn summary_from_thread_metadata(metadata: &ThreadMetadata) -> ConversationSummary { - summary_from_state_db_metadata( - metadata.id, - metadata.rollout_path.clone(), - metadata.first_user_message.clone(), - metadata - .created_at - .to_rfc3339_opts(SecondsFormat::Secs, true), - metadata - .updated_at - .to_rfc3339_opts(SecondsFormat::Secs, true), - metadata.model_provider.clone(), - metadata.cwd.clone(), - metadata.cli_version.clone(), - metadata.source.clone(), - metadata.agent_nickname.clone(), - metadata.agent_role.clone(), - metadata.git_sha.clone(), - metadata.git_branch.clone(), - metadata.git_origin_url.clone(), - ) -} - pub(crate) async fn read_summary_from_rollout( path: &Path, fallback_provider: &str, @@ -8924,7 +8832,7 @@ async fn load_thread_summary_for_rollout( thread_id: ThreadId, rollout_path: &Path, fallback_provider: &str, - persisted_metadata: Option<&ThreadMetadata>, + persisted_metadata: Option<&StoredThread>, ) -> std::result::Result { let mut thread = read_summary_from_rollout(rollout_path, fallback_provider) .await @@ -8939,15 +8847,15 @@ async fn load_thread_summary_for_rollout( if let Some(persisted_metadata) = persisted_metadata { merge_mutable_thread_metadata( &mut thread, - summary_to_thread(summary_from_thread_metadata(persisted_metadata)), + stored_thread_to_api_thread(persisted_metadata.clone()), ); - } 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)); + } else if let Some(stored_thread) = read_stored_thread_by_thread_id(config, thread_id).await { + merge_mutable_thread_metadata(&mut thread, stored_thread_to_api_thread(stored_thread)); } let title = if let Some(metadata) = persisted_metadata { - non_empty_title(metadata) + metadata.name.clone() } else { - title_from_state_db(config, thread_id).await + thread_name_from_thread_store(config, thread_id).await }; if let Some(title) = title { set_thread_name_from_title(&mut thread, title); @@ -9316,19 +9224,37 @@ mod tests { fn test_thread_metadata( model: Option<&str>, reasoning_effort: Option, - ) -> Result { + ) -> Result { let thread_id = ThreadId::from_string("3f941c35-29b3-493b-b0a4-e25800d9aeb0")?; - let mut builder = ThreadMetadataBuilder::new( + let now = Utc::now(); + Ok(StoredThread { thread_id, - PathBuf::from("/tmp/rollout.jsonl"), - Utc::now(), - codex_protocol::protocol::SessionSource::default(), - ); - builder.model_provider = Some("mock_provider".to_string()); - let mut metadata = builder.build("mock_provider"); - metadata.model = model.map(ToString::to_string); - metadata.reasoning_effort = reasoning_effort; - Ok(metadata) + forked_from_id: None, + legacy_path: Some(PathBuf::from("/tmp/rollout.jsonl")), + owner: ThreadOwner::default(), + preview: String::new(), + name: None, + model_provider: "mock_provider".to_string(), + model: model.map(ToString::to_string), + service_tier: None, + reasoning_effort, + created_at: now, + updated_at: now, + archived_at: None, + cwd: PathBuf::new(), + cli_version: String::new(), + source: codex_protocol::protocol::SessionSource::default(), + agent_nickname: None, + agent_role: None, + agent_path: None, + git_info: None, + approval_mode: codex_protocol::protocol::AskForApproval::OnRequest, + sandbox_policy: codex_protocol::protocol::SandboxPolicy::new_read_only_policy(), + token_usage: None, + first_user_message: None, + memory_mode: None, + history: None, + }) } #[test] @@ -9750,33 +9676,46 @@ mod tests { } #[test] - fn summary_from_state_db_metadata_preserves_agent_nickname() -> Result<()> { + fn stored_thread_to_conversation_summary_preserves_agent_nickname() -> Result<()> { let conversation_id = ThreadId::from_string("bfd12a78-5900-467b-9bc5-d3d35df08191")?; - let source = - serde_json::to_string(&SessionSource::SubAgent(SubAgentSource::ThreadSpawn { - parent_thread_id: ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?, - depth: 1, - agent_path: None, - agent_nickname: None, - agent_role: None, - }))?; + let source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id: ThreadId::from_string("ad7f0408-99b8-4f6e-a46f-bd0eec433370")?, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }); + let created_at = DateTime::parse_from_rfc3339("2025-09-05T16:53:11Z")?.to_utc(); + let updated_at = DateTime::parse_from_rfc3339("2025-09-05T16:53:12Z")?.to_utc(); - let summary = summary_from_state_db_metadata( - conversation_id, - PathBuf::from("/tmp/rollout.jsonl"), - Some("hi".to_string()), - "2025-09-05T16:53:11Z".to_string(), - "2025-09-05T16:53:12Z".to_string(), - "test-provider".to_string(), - PathBuf::from("/"), - "0.0.0".to_string(), + let summary = stored_thread_to_conversation_summary(StoredThread { + thread_id: conversation_id, + forked_from_id: None, + legacy_path: Some(PathBuf::from("/tmp/rollout.jsonl")), + owner: ThreadOwner::default(), + preview: String::new(), + name: None, + model_provider: "test-provider".to_string(), + model: None, + service_tier: None, + reasoning_effort: None, + created_at, + updated_at, + archived_at: None, + cwd: PathBuf::from("/"), + cli_version: "0.0.0".to_string(), source, - Some("atlas".to_string()), - Some("explorer".to_string()), - /*git_sha*/ None, - /*git_branch*/ None, - /*git_origin_url*/ None, - ); + agent_nickname: Some("atlas".to_string()), + agent_role: Some("explorer".to_string()), + agent_path: None, + git_info: None, + approval_mode: codex_protocol::protocol::AskForApproval::OnRequest, + sandbox_policy: codex_protocol::protocol::SandboxPolicy::new_read_only_policy(), + token_usage: None, + first_user_message: Some("hi".to_string()), + memory_mode: None, + history: None, + }); let thread = summary_to_thread(summary); diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index fa92001c7d..3e432dc056 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -161,6 +161,7 @@ pub use rollout::GitInfoPatch; pub use rollout::INTERACTIVE_SESSION_SOURCES; pub use rollout::ListThreadsParams as StoreListThreadsParams; pub use rollout::LocalThreadStore; +pub use rollout::MissingThreadMetadata; pub use rollout::ReadThreadParams; pub use rollout::RolloutRecorder; pub use rollout::RolloutRecorderParams; @@ -171,6 +172,7 @@ pub use rollout::StoreThreadSortKey; pub use rollout::StoredThread; pub use rollout::ThreadItem; pub use rollout::ThreadMetadataPatch; +pub use rollout::ThreadMetadataSeed; pub use rollout::ThreadOwner; pub use rollout::ThreadSortKey; pub use rollout::ThreadStore; diff --git a/codex-rs/core/src/rollout.rs b/codex-rs/core/src/rollout.rs index 7e0eebfa52..7135b57954 100644 --- a/codex-rs/core/src/rollout.rs +++ b/codex-rs/core/src/rollout.rs @@ -27,10 +27,12 @@ pub use codex_thread_store::FindThreadByNameParams; pub use codex_thread_store::GitInfoPatch; pub use codex_thread_store::ListThreadsParams; pub use codex_thread_store::LocalThreadStore; +pub use codex_thread_store::MissingThreadMetadata; pub use codex_thread_store::ReadThreadParams; pub use codex_thread_store::SetThreadNameParams; pub use codex_thread_store::StoredThread; pub use codex_thread_store::ThreadMetadataPatch; +pub use codex_thread_store::ThreadMetadataSeed; pub use codex_thread_store::ThreadOwner; pub use codex_thread_store::ThreadSortKey as StoreThreadSortKey; pub use codex_thread_store::ThreadStore; diff --git a/codex-rs/thread-store/Cargo.toml b/codex-rs/thread-store/Cargo.toml index 2467fb86ff..6c1acee3cb 100644 --- a/codex-rs/thread-store/Cargo.toml +++ b/codex-rs/thread-store/Cargo.toml @@ -22,3 +22,7 @@ serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, features = ["fs", "macros", "rt"] } + +[dev-dependencies] +pretty_assertions = { workspace = true } +tempfile = { workspace = true } diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index d750a28454..8cf93a7f62 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -1,7 +1,7 @@ //! Storage-neutral thread persistence interfaces. //! //! Application code should treat [`codex_protocol::ThreadId`] as the only durable thread handle. -//! Implementations are responsible for resolving that id to local rollout files, HTTP requests, or +//! Implementations are responsible for resolving that id to local rollout files, RPC requests, or //! any other backing store. mod error; @@ -18,24 +18,23 @@ pub use store::ThreadStore; pub use types::AppendThreadItemsParams; pub use types::ArchiveThreadParams; pub use types::CreateThreadParams; -pub use types::DynamicToolsParams; pub use types::FindThreadByNameParams; pub use types::FindThreadSpawnByPathParams; pub use types::GitInfoPatch; pub use types::ListThreadSpawnEdgesParams; pub use types::ListThreadsParams; pub use types::LoadThreadHistoryParams; +pub use types::MissingThreadMetadata; pub use types::OptionalStringPatch; pub use types::ReadThreadParams; pub use types::ResolveLegacyPathParams; pub use types::ResumeThreadRecorderParams; -pub use types::SetThreadMemoryModeParams; pub use types::SetThreadNameParams; pub use types::StoredThread; pub use types::StoredThreadHistory; pub use types::ThreadEventPersistenceMode; -pub use types::ThreadMemoryModeParams; pub use types::ThreadMetadataPatch; +pub use types::ThreadMetadataSeed; pub use types::ThreadOwner; pub use types::ThreadPage; pub use types::ThreadSortKey; diff --git a/codex-rs/thread-store/src/local/helpers.rs b/codex-rs/thread-store/src/local/helpers.rs index f3f2ffbe43..a5ea9753f5 100644 --- a/codex-rs/thread-store/src/local/helpers.rs +++ b/codex-rs/thread-store/src/local/helpers.rs @@ -7,7 +7,6 @@ use chrono::DateTime; use chrono::Utc; use codex_git_utils::GitSha; use codex_protocol::ThreadId; -use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::GitInfo; use codex_protocol::protocol::RolloutItem; @@ -70,6 +69,7 @@ pub(crate) fn stored_thread_from_metadata( ) -> StoredThread { let thread_id = metadata.id; let git_info = git_info_from_metadata(&metadata); + let name = name.or_else(|| distinct_title(&metadata)); StoredThread { thread_id, forked_from_id: None, @@ -103,6 +103,15 @@ pub(crate) fn stored_thread_from_metadata( } } +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 git_info_from_metadata(metadata: &ThreadMetadata) -> Option { if metadata.git_sha.is_none() && metadata.git_branch.is_none() @@ -147,16 +156,6 @@ pub(crate) fn memory_mode_from_items(items: &[RolloutItem]) -> Option { }) } -pub(crate) fn dynamic_tools_from_items(items: &[RolloutItem]) -> Option> { - items.iter().find_map(|item| match item { - RolloutItem::SessionMeta(meta) => meta.meta.dynamic_tools.clone(), - RolloutItem::ResponseItem(_) - | RolloutItem::Compacted(_) - | RolloutItem::TurnContext(_) - | RolloutItem::EventMsg(_) => None, - }) -} - pub(crate) fn parse_cursor_param(cursor: Option<&str>) -> ThreadStoreResult> { cursor .map(|cursor| { diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index ed122a6e8b..22dcdbc49f 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -5,7 +5,6 @@ use std::time::SystemTime; use async_trait::async_trait; use chrono::Utc; use codex_protocol::ThreadId; -use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::ThreadNameUpdatedEvent; @@ -22,6 +21,8 @@ use codex_rollout::find_archived_thread_path_by_id_str; use codex_rollout::find_thread_meta_by_name_str; use codex_rollout::find_thread_path_by_id_str; use codex_rollout::rollout_date_parts; +use codex_state::StateRuntime; +use codex_state::ThreadMetadataBuilder; mod helpers; mod read; @@ -29,10 +30,8 @@ mod recorder; use self::helpers::checked_rollout_file_name; use self::helpers::display_error; -use self::helpers::dynamic_tools_from_items; use self::helpers::edge_status_to_state; use self::helpers::io_error; -use self::helpers::memory_mode_from_items; use self::helpers::parse_cursor_param; use self::helpers::rollout_sort_key; use self::helpers::serialize_cursor; @@ -42,21 +41,20 @@ use self::recorder::RolloutThreadRecorder; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; -use crate::DynamicToolsParams; use crate::FindThreadByNameParams; use crate::FindThreadSpawnByPathParams; use crate::ListThreadSpawnEdgesParams; use crate::ListThreadsParams; use crate::LoadThreadHistoryParams; +use crate::MissingThreadMetadata; use crate::ReadThreadParams; use crate::ResolveLegacyPathParams; use crate::ResumeThreadRecorderParams; -use crate::SetThreadMemoryModeParams; use crate::SetThreadNameParams; use crate::StoredThread; use crate::StoredThreadHistory; -use crate::ThreadMemoryModeParams; use crate::ThreadMetadataPatch; +use crate::ThreadMetadataSeed; use crate::ThreadPage; use crate::ThreadRecorder; use crate::ThreadSpawnEdge; @@ -141,10 +139,11 @@ impl LocalThreadStore { "local_thread_store_find_path", ) .await - && path.exists() { - let archived = path.starts_with(self.archived_root()); - return Ok((path, archived)); - } + && path.exists() + { + let archived = path.starts_with(self.archived_root()); + return Ok((path, archived)); + } match find_thread_path_by_id_str(&self.config.codex_home, &thread_id.to_string()).await { Ok(Some(path)) => return Ok((path, false)), @@ -175,6 +174,49 @@ impl LocalThreadStore { pub(crate) fn archived_root(&self) -> PathBuf { self.config.codex_home.join(ARCHIVED_SESSIONS_SUBDIR) } + + async fn ensure_metadata_for_update( + &self, + thread_id: ThreadId, + missing: MissingThreadMetadata, + state_db: &StateRuntime, + ) -> ThreadStoreResult<()> { + if state_db + .get_thread(thread_id) + .await + .map_err(display_error)? + .is_some() + { + return Ok(()); + } + + let MissingThreadMetadata::Create(seed) = missing else { + return Err(ThreadStoreError::ThreadNotFound { thread_id }); + }; + let ThreadMetadataSeed { + rollout_path, + created_at, + source, + model_provider, + cwd, + cli_version, + sandbox_policy, + approval_mode, + } = seed; + + let mut builder = ThreadMetadataBuilder::new(thread_id, rollout_path, created_at, source); + builder.model_provider = Some(model_provider.clone()); + builder.cwd = cwd; + builder.cli_version = Some(cli_version); + builder.sandbox_policy = sandbox_policy; + builder.approval_mode = approval_mode; + let metadata = builder.build(model_provider.as_str()); + state_db + .insert_thread_if_absent(&metadata) + .await + .map_err(display_error)?; + Ok(()) + } } #[async_trait] @@ -207,7 +249,7 @@ impl ThreadStore for LocalThreadStore { event_persistence_mode.into(), ), self.state_db().await, - None, + /*state_builder*/ None, ) .await .map_err(io_error)?; @@ -228,7 +270,7 @@ impl ThreadStore for LocalThreadStore { &self.config, RolloutRecorderParams::resume(path, params.event_persistence_mode.into()), self.state_db().await, - None, + /*state_builder*/ None, ) .await .map_err(io_error)?; @@ -239,7 +281,9 @@ impl ThreadStore for LocalThreadStore { } async fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreResult<()> { - let (path, archived) = self.find_path(params.thread_id, true).await?; + let (path, archived) = self + .find_path(params.thread_id, /*include_archived*/ true) + .await?; for item in ¶ms.items { append_rollout_item_to_path(path.as_path(), item) .await @@ -250,7 +294,7 @@ impl ThreadStore for LocalThreadStore { state_db.as_deref(), path.as_path(), self.config.model_provider_id.as_str(), - None, + /*builder*/ None, params.items.as_slice(), "local_thread_store_append_items", params.new_thread_memory_mode.as_deref(), @@ -306,7 +350,7 @@ impl ThreadStore for LocalThreadStore { thread_id: params.thread_id, }); }; - self.stored_thread_from_state_metadata(metadata, false) + self.stored_thread_from_state_metadata(metadata, /*include_history*/ false) .await } Err(err) => Err(err), @@ -460,13 +504,19 @@ impl ThreadStore for LocalThreadStore { { return Ok(None); } - self.stored_thread_from_path(path.as_path(), false, false) - .await - .map(Some) + self.stored_thread_from_path( + path.as_path(), + /*archived*/ false, + /*include_history*/ false, + ) + .await + .map(Some) } async fn set_thread_name(&self, params: SetThreadNameParams) -> ThreadStoreResult<()> { - let (path, _) = self.find_path(params.thread_id, false).await?; + let (path, _) = self + .find_path(params.thread_id, /*include_archived*/ false) + .await?; let item = RolloutItem::EventMsg(EventMsg::ThreadNameUpdated(ThreadNameUpdatedEvent { thread_id: params.thread_id, thread_name: Some(params.name.clone()), @@ -486,10 +536,10 @@ impl ThreadStore for LocalThreadStore { state_db.as_deref(), path.as_path(), self.config.model_provider_id.as_str(), - None, + /*builder*/ None, &[], - None, - None, + /*archived_only*/ None, + /*new_thread_memory_mode*/ None, ) .await; Ok(()) @@ -499,6 +549,7 @@ impl ThreadStore for LocalThreadStore { &self, params: UpdateThreadMetadataParams, ) -> ThreadStoreResult { + let missing = params.missing.clone(); let ThreadMetadataPatch { name, git_info } = params.patch; if let Some(name) = name && let Some(name) = name @@ -518,33 +569,30 @@ impl ThreadStore for LocalThreadStore { message: "sqlite state db unavailable for git metadata update".to_string(), }); }; - match self.find_path(params.thread_id, true).await { + match self + .find_path(params.thread_id, /*include_archived*/ true) + .await + { Ok((path, archived)) => { codex_rollout::state_db::reconcile_rollout( Some(ctx), path.as_path(), self.config.model_provider_id.as_str(), - None, + /*builder*/ None, &[], Some(archived), - None, + /*new_thread_memory_mode*/ None, ) .await; } Err(ThreadStoreError::ThreadNotFound { .. }) => { - if ctx - .get_thread(params.thread_id) - .await - .map_err(display_error)? - .is_none() - { - return Err(ThreadStoreError::ThreadNotFound { - thread_id: params.thread_id, - }); - } + self.ensure_metadata_for_update(params.thread_id, missing.clone(), ctx) + .await?; } Err(err) => return Err(err), } + self.ensure_metadata_for_update(params.thread_id, missing, ctx) + .await?; ctx.update_thread_git_info( params.thread_id, git_info.sha.as_ref().map(|value| value.as_deref()), @@ -559,7 +607,9 @@ impl ThreadStore for LocalThreadStore { } async fn archive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreResult<()> { - let (path, _) = self.find_path(params.thread_id, false).await?; + let (path, _) = self + .find_path(params.thread_id, /*include_archived*/ false) + .await?; let canonical_sessions_dir = tokio::fs::canonicalize(self.sessions_root()) .await .map_err(io_error)?; @@ -595,7 +645,9 @@ impl ThreadStore for LocalThreadStore { &self, params: ArchiveThreadParams, ) -> ThreadStoreResult { - let (path, _) = self.find_path(params.thread_id, true).await?; + let (path, _) = self + .find_path(params.thread_id, /*include_archived*/ true) + .await?; let canonical_archived_dir = tokio::fs::canonicalize(self.archived_root()) .await .map_err(io_error)?; @@ -646,8 +698,12 @@ impl ThreadStore for LocalThreadStore { .await .map_err(display_error)?; } - self.stored_thread_from_path(restored_path.as_path(), false, false) - .await + self.stored_thread_from_path( + restored_path.as_path(), + /*archived*/ false, + /*include_history*/ false, + ) + .await } async fn resolve_legacy_path( @@ -663,80 +719,6 @@ impl ThreadStore for LocalThreadStore { Ok(thread_id) } - async fn dynamic_tools( - &self, - params: DynamicToolsParams, - ) -> ThreadStoreResult>> { - let state_db = self.state_db().await; - if let Some(tools) = codex_rollout::state_db::get_dynamic_tools( - state_db.as_deref(), - params.thread_id, - "local_thread_store_dynamic_tools", - ) - .await - { - return Ok(Some(tools)); - } - let history = self - .load_history(LoadThreadHistoryParams { - thread_id: params.thread_id, - owner: params.owner, - include_archived: true, - }) - .await?; - Ok(dynamic_tools_from_items(&history.items)) - } - - async fn memory_mode( - &self, - params: ThreadMemoryModeParams, - ) -> ThreadStoreResult> { - let state_db = self.state_db().await; - if let Some(ctx) = state_db.as_deref() { - return ctx - .get_thread_memory_mode(params.thread_id) - .await - .map_err(display_error); - } - let history = self - .load_history(LoadThreadHistoryParams { - thread_id: params.thread_id, - owner: params.owner, - include_archived: true, - }) - .await?; - Ok(memory_mode_from_items(&history.items)) - } - - async fn set_memory_mode(&self, params: SetThreadMemoryModeParams) -> ThreadStoreResult<()> { - let state_db = self.state_db().await; - let Some(ctx) = state_db.as_deref() else { - return Err(ThreadStoreError::Unavailable { - message: "sqlite state db unavailable for memory mode update".to_string(), - }); - }; - ctx.set_thread_memory_mode(params.thread_id, params.memory_mode.as_str()) - .await - .map_err(display_error)?; - Ok(()) - } - - async fn mark_memory_mode_polluted( - &self, - params: ThreadMemoryModeParams, - ) -> ThreadStoreResult<()> { - let state_db = self.state_db().await; - let Some(ctx) = state_db.as_deref() else { - return Err(ThreadStoreError::Unavailable { - message: "sqlite state db unavailable for memory mode pollution update".to_string(), - }); - }; - ctx.mark_thread_memory_mode_polluted(params.thread_id) - .await - .map_err(display_error)?; - Ok(()) - } - async fn upsert_thread_spawn_edge(&self, edge: ThreadSpawnEdge) -> ThreadStoreResult<()> { let state_db = self.state_db().await; let Some(ctx) = state_db.as_deref() else { @@ -822,3 +804,231 @@ impl ThreadStore for LocalThreadStore { path.starts_with(self.sessions_root()) || path.starts_with(self.archived_root()) } } + +#[cfg(test)] +mod tests { + use std::path::Path; + + use chrono::TimeZone; + use chrono::Utc; + use codex_git_utils::GitSha; + use codex_protocol::ThreadId; + use codex_protocol::models::BaseInstructions; + use codex_protocol::protocol::AgentMessageEvent; + use codex_protocol::protocol::AskForApproval; + use codex_protocol::protocol::EventMsg; + use codex_protocol::protocol::RolloutItem; + use codex_protocol::protocol::SandboxPolicy; + use codex_protocol::protocol::SessionSource; + use codex_protocol::protocol::UserMessageEvent; + use codex_rollout::RolloutConfig; + use pretty_assertions::assert_eq; + use tempfile::TempDir; + + use crate::CreateThreadParams; + use crate::GitInfoPatch; + use crate::MissingThreadMetadata; + use crate::ReadThreadParams; + use crate::ThreadEventPersistenceMode; + use crate::ThreadMetadataPatch; + use crate::ThreadMetadataSeed; + use crate::ThreadOwner; + use crate::ThreadStore; + use crate::ThreadStoreError; + use crate::UpdateThreadMetadataParams; + + use super::LocalThreadStore; + + fn test_config(codex_home: &Path) -> RolloutConfig { + RolloutConfig { + codex_home: codex_home.to_path_buf(), + sqlite_home: codex_home.to_path_buf(), + cwd: codex_home.to_path_buf(), + model_provider_id: "test-provider".to_string(), + generate_memories: true, + } + } + + fn metadata_seed(thread_id: ThreadId, codex_home: &Path, cwd: &Path) -> ThreadMetadataSeed { + ThreadMetadataSeed { + rollout_path: codex_home.join(format!("sessions/seeded-{thread_id}.jsonl")), + created_at: Utc + .with_ymd_and_hms(2025, 1, 3, 4, 5, 6) + .single() + .expect("valid timestamp"), + source: SessionSource::Exec, + model_provider: "seed-provider".to_string(), + cwd: cwd.to_path_buf(), + cli_version: "seed-cli-version".to_string(), + sandbox_policy: SandboxPolicy::new_workspace_write_policy(), + approval_mode: AskForApproval::Never, + } + } + + fn git_patch( + sha: impl Into, + branch: impl Into, + origin_url: impl Into, + ) -> ThreadMetadataPatch { + ThreadMetadataPatch { + name: None, + git_info: Some(GitInfoPatch { + sha: Some(Some(sha.into())), + branch: Some(Some(branch.into())), + origin_url: Some(Some(origin_url.into())), + }), + } + } + + async fn create_materialized_thread(store: &LocalThreadStore, cwd: &Path) -> ThreadId { + let thread_id = ThreadId::new(); + let recorder = store + .create_thread(CreateThreadParams { + thread_id, + owner: ThreadOwner::default(), + forked_from_id: None, + source: SessionSource::Exec, + cwd: cwd.to_path_buf(), + originator: "test-originator".to_string(), + cli_version: "test-cli-version".to_string(), + model_provider: "test-provider".to_string(), + model: None, + service_tier: None, + reasoning_effort: None, + approval_mode: AskForApproval::OnRequest, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + memory_mode: None, + git_info: None, + event_persistence_mode: ThreadEventPersistenceMode::Limited, + }) + .await + .expect("create thread recorder"); + recorder + .record_items(&[ + RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent { + message: "hello from the local store".to_string(), + images: None, + local_images: Vec::new(), + text_elements: Vec::new(), + })), + RolloutItem::EventMsg(EventMsg::AgentMessage(AgentMessageEvent { + message: "hello back".to_string(), + phase: None, + memory_citation: None, + })), + ]) + .await + .expect("record rollout items"); + recorder.flush().await.expect("flush rollout"); + thread_id + } + + #[tokio::test(flavor = "current_thread")] + async fn update_thread_metadata_create_backfills_missing_sqlite_row() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::with_state_db(test_config(home.path())).await; + let thread_id = ThreadId::new(); + let seed = metadata_seed(thread_id, home.path(), home.path()); + + let stored = store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + owner: ThreadOwner::default(), + patch: git_patch("abc123", "main", "https://example.com/repo.git"), + missing: MissingThreadMetadata::Create(seed.clone()), + }) + .await + .expect("metadata update succeeds"); + + assert_eq!(stored.thread_id, thread_id); + assert_eq!(stored.legacy_path, Some(seed.rollout_path)); + assert_eq!(stored.model_provider, "seed-provider"); + assert_eq!(stored.cwd, home.path()); + assert_eq!(stored.cli_version, "seed-cli-version"); + assert_eq!(stored.source, SessionSource::Exec); + assert_eq!( + stored.sandbox_policy, + SandboxPolicy::new_workspace_write_policy() + ); + assert_eq!(stored.approval_mode, AskForApproval::Never); + let git_info = stored.git_info.expect("git info"); + assert_eq!(git_info.commit_hash, Some(GitSha::new("abc123"))); + assert_eq!(git_info.branch, Some("main".to_string())); + assert_eq!( + git_info.repository_url, + Some("https://example.com/repo.git".to_string()) + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn update_thread_metadata_error_does_not_backfill_missing_sqlite_row() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::with_state_db(test_config(home.path())).await; + let thread_id = ThreadId::new(); + + let err = store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + owner: ThreadOwner::default(), + patch: git_patch("abc123", "main", "https://example.com/repo.git"), + missing: MissingThreadMetadata::Error, + }) + .await + .expect_err("metadata update should fail"); + + assert!( + matches!(err, ThreadStoreError::ThreadNotFound { thread_id: id } if id == thread_id) + ); + } + + #[tokio::test(flavor = "current_thread")] + async fn read_thread_overlays_sqlite_metadata_on_rollout_summary() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::with_state_db(test_config(home.path())).await; + let thread_id = create_materialized_thread(&store, home.path()).await; + + let stored = store + .update_thread_metadata(UpdateThreadMetadataParams { + thread_id, + owner: ThreadOwner::default(), + patch: git_patch("def456", "feature/thread-store", "ssh://example/repo"), + missing: MissingThreadMetadata::Error, + }) + .await + .expect("metadata update succeeds"); + + let git_info = stored.git_info.expect("git info"); + assert_eq!(git_info.commit_hash, Some(GitSha::new("def456"))); + assert_eq!(git_info.branch, Some("feature/thread-store".to_string())); + assert_eq!( + git_info.repository_url, + Some("ssh://example/repo".to_string()) + ); + + let reread = store + .read_thread(ReadThreadParams { + thread_id, + owner: ThreadOwner::default(), + include_archived: false, + include_history: true, + }) + .await + .expect("read thread"); + + let git_info = reread.git_info.expect("git info"); + assert_eq!(git_info.commit_hash, Some(GitSha::new("def456"))); + assert_eq!(git_info.branch, Some("feature/thread-store".to_string())); + assert_eq!( + git_info.repository_url, + Some("ssh://example/repo".to_string()) + ); + assert_eq!(reread.preview, "hello from the local store"); + assert_eq!( + reread.history.expect("history").items.len(), + 3, + "session metadata plus two recorded events should be replayable" + ); + } +} diff --git a/codex-rs/thread-store/src/local/read.rs b/codex-rs/thread-store/src/local/read.rs index 41ca869ab9..fcec75de63 100644 --- a/codex-rs/thread-store/src/local/read.rs +++ b/codex-rs/thread-store/src/local/read.rs @@ -131,7 +131,20 @@ impl LocalThreadStore { item: ThreadItem, archived: bool, ) -> ThreadStoreResult { - self.stored_thread_from_path(item.path.as_path(), archived, false) - .await + if let Some(thread_id) = item.thread_id + && let Some(ctx) = self.state_db().await.as_deref() + && let Some(metadata) = ctx.get_thread(thread_id).await.map_err(display_error)? + { + return self + .stored_thread_from_state_metadata(metadata, /*include_history*/ false) + .await; + } + + self.stored_thread_from_path( + item.path.as_path(), + archived, + /*include_history*/ false, + ) + .await } } diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index 87f7c26c49..177d41e648 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -2,12 +2,10 @@ use std::path::Path; use async_trait::async_trait; use codex_protocol::ThreadId; -use codex_protocol::dynamic_tools::DynamicToolSpec; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; -use crate::DynamicToolsParams; use crate::FindThreadByNameParams; use crate::FindThreadSpawnByPathParams; use crate::ListThreadSpawnEdgesParams; @@ -16,11 +14,9 @@ use crate::LoadThreadHistoryParams; use crate::ReadThreadParams; use crate::ResolveLegacyPathParams; use crate::ResumeThreadRecorderParams; -use crate::SetThreadMemoryModeParams; use crate::SetThreadNameParams; use crate::StoredThread; use crate::StoredThreadHistory; -use crate::ThreadMemoryModeParams; use crate::ThreadPage; use crate::ThreadRecorder; use crate::ThreadSpawnEdge; @@ -87,27 +83,6 @@ pub trait ThreadStore: Send + Sync { params: ResolveLegacyPathParams, ) -> ThreadStoreResult>; - /// Reads dynamic tools persisted for a thread. - async fn dynamic_tools( - &self, - params: DynamicToolsParams, - ) -> ThreadStoreResult>>; - - /// Reads a thread's memory mode. - async fn memory_mode( - &self, - params: ThreadMemoryModeParams, - ) -> ThreadStoreResult>; - - /// Updates a thread's memory mode. - async fn set_memory_mode(&self, params: SetThreadMemoryModeParams) -> ThreadStoreResult<()>; - - /// Marks a thread's memory context as polluted. - async fn mark_memory_mode_polluted( - &self, - params: ThreadMemoryModeParams, - ) -> ThreadStoreResult<()>; - /// Persists or replaces a thread-spawn parent-child edge. async fn upsert_thread_spawn_edge(&self, edge: ThreadSpawnEdge) -> ThreadStoreResult<()>; diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 1d535169c9..5105ead00d 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -277,6 +277,37 @@ pub struct ThreadMetadataPatch { pub git_info: Option, } +/// Metadata needed to create a missing persisted metadata row/object for an existing thread. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ThreadMetadataSeed { + /// Legacy local rollout path for implementations that still materialize rollouts on disk. + pub rollout_path: PathBuf, + /// Timestamp to use if the backing store must create metadata from this seed. + pub created_at: DateTime, + /// Runtime source for the thread. + pub source: SessionSource, + /// Model provider id associated with the thread. + pub model_provider: String, + /// Working directory captured for the thread. + pub cwd: PathBuf, + /// Version of the CLI/app that created the thread. + pub cli_version: String, + /// Sandbox policy captured for the thread. + pub sandbox_policy: SandboxPolicy, + /// Approval mode captured for the thread. + pub approval_mode: AskForApproval, +} + +/// Controls how metadata updates behave when the backing metadata record is missing. +#[allow(clippy::large_enum_variant)] +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum MissingThreadMetadata { + /// Return a not-found error if the backing metadata record cannot be found or reconstructed. + Error, + /// Create backing metadata from the supplied seed before applying the patch. + Create(ThreadMetadataSeed), +} + /// Parameters for patching mutable thread metadata. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct UpdateThreadMetadataParams { @@ -286,6 +317,8 @@ pub struct UpdateThreadMetadataParams { pub owner: ThreadOwner, /// Patch to apply. pub patch: ThreadMetadataPatch, + /// Behavior when the backing metadata record is missing. + pub missing: MissingThreadMetadata, } /// Parameters for archiving or unarchiving a thread. @@ -326,35 +359,6 @@ pub struct ResolveLegacyPathParams { pub include_archived: bool, } -/// Parameters for reading dynamic tool specs persisted for a thread. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct DynamicToolsParams { - /// Thread id to read. - pub thread_id: ThreadId, - /// Tenant/owner metadata for multi-tenant stores. - pub owner: ThreadOwner, -} - -/// Parameters for memory-mode lookups and pollution markers. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct ThreadMemoryModeParams { - /// Thread id to read or update. - pub thread_id: ThreadId, - /// Tenant/owner metadata for multi-tenant stores. - pub owner: ThreadOwner, -} - -/// Parameters for setting an explicit memory mode. -#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] -pub struct SetThreadMemoryModeParams { - /// Thread id to update. - pub thread_id: ThreadId, - /// Tenant/owner metadata for multi-tenant stores. - pub owner: ThreadOwner, - /// Memory mode value to store, for example `disabled`. - pub memory_mode: String, -} - /// Status of a parent-child thread-spawn edge. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum ThreadSpawnEdgeStatus { diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index a1f813e9b6..90f7b62981 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -2938,8 +2938,10 @@ impl App { } fn is_terminal_thread_read_error(err: &color_eyre::Report) -> bool { - err.chain() - .any(|cause| cause.to_string().contains("thread not loaded:")) + err.chain().any(|cause| { + let message = cause.to_string(); + message.contains("thread not loaded:") || message.contains(" not found") + }) } fn closed_state_for_thread_read_error(