diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 8ea52e9304..a6c83a9b4b 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3465,7 +3465,6 @@ dependencies = [ "chrono", "codex-file-search", "codex-git-utils", - "codex-login", "codex-otel", "codex-protocol", "codex-state", diff --git a/codex-rs/app-server/tests/suite/conversation_summary.rs b/codex-rs/app-server/tests/suite/conversation_summary.rs index 754d1f9467..bc5511904a 100644 --- a/codex-rs/app-server/tests/suite/conversation_summary.rs +++ b/codex-rs/app-server/tests/suite/conversation_summary.rs @@ -125,6 +125,7 @@ async fn get_conversation_summary_by_thread_id_reads_pathless_store_thread() -> thread_id, forked_from_id: None, source: SessionSource::Cli, + originator: "test_originator".to_string(), thread_source: None, base_instructions: BaseInstructions::default(), dynamic_tools: Vec::new(), 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 52420c0c80..614bcd3069 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_read.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_read.rs @@ -1283,6 +1283,7 @@ async fn seed_pathless_store_thread( thread_id, forked_from_id: None, source: ProtocolSessionSource::Cli, + originator: "test_originator".to_string(), thread_source: None, base_instructions: BaseInstructions::default(), dynamic_tools: Vec::new(), diff --git a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs index 5b421dcec5..91b13ddf97 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_unarchive.rs @@ -211,6 +211,7 @@ async fn thread_unarchive_preserves_pathless_store_metadata() -> Result<()> { thread_id, forked_from_id: Some(parent_thread_id), source: SessionSource::Cli, + originator: "test_originator".to_string(), thread_source: None, base_instructions: BaseInstructions::default(), dynamic_tools: Vec::new(), diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 1a790314d5..4a4f8a4b3e 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -418,6 +418,7 @@ impl Session { thread_id, forked_from_id, source: session_source, + originator: originator().value, thread_source: session_configuration.thread_source, base_instructions: BaseInstructions { text: session_configuration.base_instructions.clone(), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 5090fc3f07..b7890a9246 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -2946,6 +2946,7 @@ async fn attach_thread_persistence(session: &mut Session) -> PathBuf { thread_id: session.conversation_id, forked_from_id: None, source: SessionSource::Exec, + originator: "test_originator".to_string(), thread_source: None, base_instructions: BaseInstructions::default(), dynamic_tools: Vec::new(), @@ -5166,6 +5167,7 @@ async fn shutdown_complete_does_not_append_to_thread_store_after_shutdown() { thread_id: session.conversation_id, forked_from_id: None, source: SessionSource::Exec, + originator: "test_originator".to_string(), thread_source: None, base_instructions: BaseInstructions::default(), dynamic_tools: Vec::new(), diff --git a/codex-rs/core/tests/suite/rollout_list_find.rs b/codex-rs/core/tests/suite/rollout_list_find.rs index 51b3f2db1a..82115e89ec 100644 --- a/codex-rs/core/tests/suite/rollout_list_find.rs +++ b/codex-rs/core/tests/suite/rollout_list_find.rs @@ -182,15 +182,16 @@ async fn find_locates_rollout_file_written_by_recorder() -> std::io::Result<()> let thread_name = "named thread"; let recorder = RolloutRecorder::new( &config, - RolloutRecorderParams::new( - thread_id, - /*forked_from_id*/ None, - SessionSource::Exec, - /*thread_source*/ None, - BaseInstructions::default(), - Vec::new(), - EventPersistenceMode::Limited, - ), + RolloutRecorderParams::Create { + conversation_id: thread_id, + forked_from_id: None, + source: SessionSource::Exec, + originator: "test_originator".to_string(), + thread_source: None, + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + event_persistence_mode: EventPersistenceMode::Limited, + }, /*state_db_ctx*/ None, /*state_builder*/ None, ) diff --git a/codex-rs/rollout/Cargo.toml b/codex-rs/rollout/Cargo.toml index ef5a8dc22a..e3f30ccfd0 100644 --- a/codex-rs/rollout/Cargo.toml +++ b/codex-rs/rollout/Cargo.toml @@ -18,7 +18,6 @@ async-trait = { workspace = true } chrono = { workspace = true, features = ["serde"] } codex-file-search = { workspace = true } codex-git-utils = { workspace = true } -codex-login = { workspace = true } codex-otel = { workspace = true } codex-protocol = { workspace = true } codex-state = { workspace = true } diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index d65ddd3d5b..95c58d4cc2 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -12,10 +12,6 @@ pub(crate) mod recorder; pub(crate) mod session_index; pub mod state_db; -pub(crate) mod default_client { - pub use codex_login::default_client::*; -} - pub(crate) use codex_protocol::protocol; pub const SESSIONS_SUBDIR: &str = "sessions"; @@ -54,6 +50,8 @@ pub use list::rollout_date_parts; pub use metadata::builder_from_items; pub use policy::EventPersistenceMode; pub use policy::is_persisted_rollout_item; +pub use policy::persisted_rollout_items; +pub use policy::sanitize_rollout_item_for_persistence; pub use policy::should_persist_response_item_for_memories; pub use recorder::RolloutRecorder; pub use recorder::RolloutRecorderParams; diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 558c3fef98..af93c1d0c0 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -1,6 +1,7 @@ use crate::protocol::EventMsg; use crate::protocol::RolloutItem; use codex_protocol::models::ResponseItem; +use codex_utils_string::truncate_middle_chars; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub enum EventPersistenceMode { @@ -22,6 +23,46 @@ pub fn is_persisted_rollout_item(item: &RolloutItem, mode: EventPersistenceMode) } } +const PERSISTED_EXEC_AGGREGATED_OUTPUT_MAX_BYTES: usize = 10_000; + +pub fn sanitize_rollout_item_for_persistence( + item: RolloutItem, + mode: EventPersistenceMode, +) -> RolloutItem { + if mode != EventPersistenceMode::Extended { + return item; + } + + match item { + RolloutItem::EventMsg(EventMsg::ExecCommandEnd(mut event)) => { + // Persist only a bounded aggregated summary of command output. + event.aggregated_output = truncate_middle_chars( + &event.aggregated_output, + PERSISTED_EXEC_AGGREGATED_OUTPUT_MAX_BYTES, + ); + // Drop unnecessary fields from rollout storage since aggregated_output is all we need. + event.stdout.clear(); + event.stderr.clear(); + event.formatted_output.clear(); + RolloutItem::EventMsg(EventMsg::ExecCommandEnd(event)) + } + _ => item, + } +} + +pub fn persisted_rollout_items( + items: &[RolloutItem], + mode: EventPersistenceMode, +) -> Vec { + let mut filtered = Vec::new(); + for item in items { + if is_persisted_rollout_item(item, mode) { + filtered.push(sanitize_rollout_item_for_persistence(item.clone(), mode)); + } + } + filtered +} + /// Whether a `ResponseItem` should be persisted in rollout files. #[inline] pub fn should_persist_response_item(item: &ResponseItem) -> bool { diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 6b549b745c..5555072845 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -51,7 +51,6 @@ use super::policy::EventPersistenceMode; use super::policy::is_persisted_rollout_item; use super::session_index::find_thread_names_by_ids; use crate::config::RolloutConfigView; -use crate::default_client::originator; use crate::state_db; use crate::state_db::StateDbHandle; use codex_git_utils::collect_git_info; @@ -92,6 +91,7 @@ pub enum RolloutRecorderParams { conversation_id: ThreadId, forked_from_id: Option, source: SessionSource, + originator: String, thread_source: Option, base_instructions: BaseInstructions, dynamic_tools: Vec, @@ -165,26 +165,6 @@ fn clone_io_error(err: &IoError) -> IoError { } impl RolloutRecorderParams { - pub fn new( - conversation_id: ThreadId, - forked_from_id: Option, - source: SessionSource, - thread_source: Option, - base_instructions: BaseInstructions, - dynamic_tools: Vec, - event_persistence_mode: EventPersistenceMode, - ) -> Self { - Self::Create { - conversation_id, - forked_from_id, - source, - thread_source, - base_instructions, - dynamic_tools, - event_persistence_mode, - } - } - pub fn resume(path: PathBuf, event_persistence_mode: EventPersistenceMode) -> Self { Self::Resume { path, @@ -674,6 +654,7 @@ impl RolloutRecorder { conversation_id, forked_from_id, source, + originator, thread_source, base_instructions, dynamic_tools, @@ -697,7 +678,7 @@ impl RolloutRecorder { forked_from_id, timestamp, cwd: config.cwd().to_path_buf(), - originator: originator().value, + originator, cli_version: env!("CARGO_PKG_VERSION").to_string(), agent_nickname: source.get_nickname(), agent_role: source.get_agent_role(), diff --git a/codex-rs/rollout/src/recorder_tests.rs b/codex-rs/rollout/src/recorder_tests.rs index a8b946934b..be16d29049 100644 --- a/codex-rs/rollout/src/recorder_tests.rs +++ b/codex-rs/rollout/src/recorder_tests.rs @@ -365,15 +365,16 @@ async fn recorder_materializes_on_flush_with_pending_items() -> std::io::Result< let thread_id = ThreadId::new(); let recorder = RolloutRecorder::new( &config, - RolloutRecorderParams::new( - thread_id, - /*forked_from_id*/ None, - SessionSource::Exec, - /*thread_source*/ None, - BaseInstructions::default(), - Vec::new(), - EventPersistenceMode::Limited, - ), + RolloutRecorderParams::Create { + conversation_id: thread_id, + forked_from_id: None, + source: SessionSource::Exec, + originator: "test_originator".to_string(), + thread_source: None, + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + event_persistence_mode: EventPersistenceMode::Limited, + }, /*state_db_ctx*/ None, /*state_builder*/ None, ) @@ -446,15 +447,16 @@ async fn persist_reports_filesystem_error_and_retries_buffered_items() -> std::i let thread_id = ThreadId::new(); let recorder = RolloutRecorder::new( &config, - RolloutRecorderParams::new( - thread_id, - /*forked_from_id*/ None, - SessionSource::Exec, - /*thread_source*/ None, - BaseInstructions::default(), - Vec::new(), - EventPersistenceMode::Limited, - ), + RolloutRecorderParams::Create { + conversation_id: thread_id, + forked_from_id: None, + source: SessionSource::Exec, + originator: "test_originator".to_string(), + thread_source: None, + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + event_persistence_mode: EventPersistenceMode::Limited, + }, /*state_db_ctx*/ None, /*state_builder*/ None, ) @@ -546,15 +548,16 @@ async fn metadata_irrelevant_events_coalesce_state_db_updated_at() -> std::io::R let thread_id = ThreadId::new(); let recorder = RolloutRecorder::new( &config, - RolloutRecorderParams::new( - thread_id, - /*forked_from_id*/ None, - SessionSource::Cli, - /*thread_source*/ None, - BaseInstructions::default(), - Vec::new(), - EventPersistenceMode::Limited, - ), + RolloutRecorderParams::Create { + conversation_id: thread_id, + forked_from_id: None, + source: SessionSource::Cli, + originator: "test_originator".to_string(), + thread_source: None, + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + event_persistence_mode: EventPersistenceMode::Limited, + }, Some(state_db.clone()), /*state_builder*/ None, ) diff --git a/codex-rs/rollout/src/session_index.rs b/codex-rs/rollout/src/session_index.rs index e227515357..fa950b3404 100644 --- a/codex-rs/rollout/src/session_index.rs +++ b/codex-rs/rollout/src/session_index.rs @@ -103,9 +103,13 @@ pub async fn find_thread_names_by_ids( let Ok(entry) = serde_json::from_str::(trimmed) else { continue; }; - let name = entry.thread_name.trim(); - if !name.is_empty() && thread_ids.contains(&entry.id) { - names.insert(entry.id, name.to_string()); + if thread_ids.contains(&entry.id) { + let name = entry.thread_name.trim(); + if name.is_empty() { + names.remove(&entry.id); + } else { + names.insert(entry.id, name.to_string()); + } } } diff --git a/codex-rs/rollout/src/session_index_tests.rs b/codex-rs/rollout/src/session_index_tests.rs index 757b08b4d4..562231636a 100644 --- a/codex-rs/rollout/src/session_index_tests.rs +++ b/codex-rs/rollout/src/session_index_tests.rs @@ -266,6 +266,33 @@ async fn find_thread_names_by_ids_prefers_latest_entry() -> std::io::Result<()> Ok(()) } +#[tokio::test] +async fn find_thread_names_by_ids_respects_empty_tombstone() -> std::io::Result<()> { + let temp = TempDir::new()?; + let path = session_index_path(temp.path()); + let id = ThreadId::new(); + let lines = vec![ + SessionIndexEntry { + id, + thread_name: "first".to_string(), + updated_at: "2024-01-01T00:00:00Z".to_string(), + }, + SessionIndexEntry { + id, + thread_name: String::new(), + updated_at: "2024-01-02T00:00:00Z".to_string(), + }, + ]; + write_index(&path, &lines)?; + + let mut ids = HashSet::new(); + ids.insert(id); + + let found = find_thread_names_by_ids(temp.path(), &ids).await?; + assert!(found.is_empty()); + Ok(()) +} + #[test] fn scan_index_finds_latest_match_among_mixed_entries() -> std::io::Result<()> { let temp = TempDir::new()?; diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 1795a864d8..3fd833cbd5 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -830,6 +830,51 @@ ON CONFLICT(thread_id, position) DO NOTHING Ok(()) } + /// Replace dynamic tools for a thread. + /// + /// Unlike [`Self::persist_dynamic_tools`], this is for explicit metadata updates where the + /// caller owns the full current tool list. + pub async fn replace_dynamic_tools( + &self, + thread_id: ThreadId, + tools: &[DynamicToolSpec], + ) -> anyhow::Result<()> { + let thread_id = thread_id.to_string(); + let mut tx = self.pool.begin().await?; + sqlx::query("DELETE FROM thread_dynamic_tools WHERE thread_id = ?") + .bind(thread_id.as_str()) + .execute(&mut *tx) + .await?; + for (idx, tool) in tools.iter().enumerate() { + let position = i64::try_from(idx).unwrap_or(i64::MAX); + let input_schema = serde_json::to_string(&tool.input_schema)?; + sqlx::query( + r#" +INSERT INTO thread_dynamic_tools ( + thread_id, + position, + namespace, + name, + description, + input_schema, + defer_loading +) VALUES (?, ?, ?, ?, ?, ?, ?) + "#, + ) + .bind(thread_id.as_str()) + .bind(position) + .bind(tool.namespace.as_deref()) + .bind(tool.name.as_str()) + .bind(tool.description.as_str()) + .bind(input_schema) + .bind(tool.defer_loading) + .execute(&mut *tx) + .await?; + } + tx.commit().await?; + Ok(()) + } + /// Apply rollout items incrementally using the underlying database. pub async fn apply_rollout_items( &self, diff --git a/codex-rs/thread-store/README.md b/codex-rs/thread-store/README.md new file mode 100644 index 0000000000..2acb7d15be --- /dev/null +++ b/codex-rs/thread-store/README.md @@ -0,0 +1,28 @@ +# Thread Store + +`codex-thread-store` is the storage boundary for Codex threads. It defines the +`ThreadStore` trait plus local and in-memory implementations. Other storage +implementations may live outside this repository. + +## Responsibilities + +- `LiveThread` is the preferred API for active session persistence. It owns a + `ThreadStore` and serializes live operations. +- `ThreadStore::append_items` is the raw canonical history append API. It does + not infer metadata from item contents. +- `ThreadStore::apply_thread_metadata` is the explicit metadata API. Local + storage writes these facts to SQLite; other stores can route them to a + metadata endpoint independent of rollout item contents. +- `LocalThreadStore` persists history through `codex-rollout` JSONL files and + persists queryable metadata through the SQLite state database. +- `RolloutRecorder` is the local JSONL writer. It writes already-canonical + items and does not decide which rollout events should be persisted. +- `core/session` creates or resumes `LiveThread` handles and should not need to + know whether persistence is backed by local files or a remote store. +- App-server code should use `LiveThread` for loaded threads. + +## Direction + +New metadata semantics should live above `ThreadStore`. The store is +allowed to persist explicit metadata fields, but it should not derive metadata by +inspecting rollout item payloads. diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 1483906267..29ca7753a3 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -9,10 +9,12 @@ use async_trait::async_trait; use chrono::Utc; use codex_protocol::ThreadId; use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::GitInfo; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SandboxPolicy; use crate::AppendThreadItemsParams; +use crate::ApplyThreadMetadataParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::ListThreadsParams; @@ -22,6 +24,7 @@ use crate::ReadThreadParams; use crate::ResumeThreadParams; use crate::StoredThread; use crate::StoredThreadHistory; +use crate::ThreadMetadataUpdate; use crate::ThreadPage; use crate::ThreadStore; use crate::ThreadStoreError; @@ -37,11 +40,19 @@ fn stores() -> &'static Mutex>> { #[cfg(test)] mod tests { + use codex_protocol::models::BaseInstructions; + use codex_protocol::protocol::SessionSource; + use codex_protocol::protocol::ThreadMemoryMode; + use codex_protocol::protocol::TokenUsage; + use pretty_assertions::assert_eq; + use super::*; use crate::ListItemsParams; use crate::ListTurnsParams; use crate::SortDirection; use crate::StoredTurnItemsView; + use crate::ThreadEventPersistenceMode; + use crate::ThreadPersistenceMetadata; #[tokio::test] async fn default_turn_pagination_methods_return_unsupported() { @@ -84,6 +95,117 @@ mod tests { } )); } + + #[tokio::test] + async fn apply_thread_metadata_updates_readable_metadata() { + let store = InMemoryThreadStore::default(); + let thread_id = ThreadId::new(); + store + .create_thread(create_thread_params(thread_id)) + .await + .expect("create thread"); + let token_usage = TokenUsage { + total_tokens: 42, + ..Default::default() + }; + let git_info = GitInfo { + commit_hash: None, + branch: Some("main".to_string()), + repository_url: Some("https://example.test/repo.git".to_string()), + }; + let created_at = Utc::now(); + let updated_at = Utc::now(); + + store + .apply_thread_metadata(ApplyThreadMetadataParams { + thread_id, + update: ThreadMetadataUpdate { + preview: Some("preview".to_string()), + name: Some(Some("name".to_string())), + model_provider: Some("provider".to_string()), + model: Some(Some("model".to_string())), + created_at: Some(created_at), + updated_at: Some(updated_at), + cwd: Some(PathBuf::from("/tmp/project")), + cli_version: Some("cli-version".to_string()), + token_usage: Some(Some(token_usage.clone())), + first_user_message: Some(Some("first".to_string())), + git_info: Some(Some(git_info.clone())), + ..Default::default() + }, + }) + .await + .expect("apply metadata"); + + let stored = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: false, + }) + .await + .expect("read thread"); + + assert_eq!(stored.preview, "preview"); + assert_eq!(stored.name.as_deref(), Some("name")); + assert_eq!(stored.model_provider, "provider"); + assert_eq!(stored.model.as_deref(), Some("model")); + assert_eq!(stored.created_at, created_at); + assert_eq!(stored.updated_at, updated_at); + assert_eq!(stored.cwd, PathBuf::from("/tmp/project")); + assert_eq!(stored.cli_version, "cli-version"); + assert_eq!(stored.token_usage, Some(token_usage)); + assert_eq!(stored.first_user_message.as_deref(), Some("first")); + let stored_git_info = stored.git_info.expect("git info"); + assert_eq!(stored_git_info.branch, git_info.branch); + assert_eq!(stored_git_info.repository_url, git_info.repository_url); + + store + .apply_thread_metadata(ApplyThreadMetadataParams { + thread_id, + update: ThreadMetadataUpdate { + name: Some(None), + token_usage: Some(None), + first_user_message: Some(None), + git_info: Some(None), + ..Default::default() + }, + }) + .await + .expect("clear metadata"); + + let stored = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: true, + include_history: false, + }) + .await + .expect("read thread"); + + assert_eq!(stored.name, None); + assert_eq!(stored.token_usage, None); + assert_eq!(stored.first_user_message, None); + assert!(stored.git_info.is_none()); + } + + fn create_thread_params(thread_id: ThreadId) -> CreateThreadParams { + CreateThreadParams { + thread_id, + forked_from_id: None, + source: SessionSource::Exec, + originator: "test".to_string(), + thread_source: None, + base_instructions: BaseInstructions::default(), + dynamic_tools: Vec::new(), + metadata: ThreadPersistenceMetadata { + cwd: Some(PathBuf::from("/tmp/project")), + model_provider: "test-provider".to_string(), + memory_mode: ThreadMemoryMode::Enabled, + }, + event_persistence_mode: ThreadEventPersistenceMode::Limited, + } + } } fn stores_guard() -> MutexGuard<'static, HashMap>> { @@ -108,6 +230,7 @@ pub struct InMemoryThreadStoreCalls { pub read_thread_by_rollout_path: usize, pub list_threads: usize, pub update_thread_metadata: usize, + pub apply_thread_metadata: usize, pub archive_thread: usize, pub unarchive_thread: usize, } @@ -127,7 +250,9 @@ struct InMemoryThreadStoreState { calls: InMemoryThreadStoreCalls, created_threads: HashMap, histories: HashMap>, + metadata_updates: HashMap, names: HashMap>, + git_infos: HashMap>, rollout_paths: HashMap, } @@ -188,6 +313,36 @@ impl ThreadStore for InMemoryThreadStore { Ok(()) } + async fn apply_thread_metadata( + &self, + params: ApplyThreadMetadataParams, + ) -> ThreadStoreResult<()> { + let mut state = self.state.lock().await; + state.calls.apply_thread_metadata += 1; + if !state.histories.contains_key(¶ms.thread_id) { + return Err(ThreadStoreError::ThreadNotFound { + thread_id: params.thread_id, + }); + } + let update = params.update; + if let Some(rollout_path) = update.rollout_path.as_ref() { + state + .rollout_paths + .insert(rollout_path.clone(), params.thread_id); + } + if let Some(name) = update.name.clone() { + state.names.insert(params.thread_id, name); + } + if let Some(git_info) = update.git_info.clone() { + state.git_infos.insert(params.thread_id, git_info); + } + merge_metadata_update( + state.metadata_updates.entry(params.thread_id).or_default(), + update, + ); + Ok(()) + } + async fn persist_thread(&self, _thread_id: ThreadId) -> ThreadStoreResult<()> { self.state.lock().await.calls.persist_thread += 1; Ok(()) @@ -272,6 +427,15 @@ impl ThreadStore for InMemoryThreadStore { let mut state = self.state.lock().await; state.calls.update_thread_metadata += 1; if let Some(name) = params.patch.name { + let update = ThreadMetadataUpdate { + name: Some(Some(name.clone())), + updated_at: Some(Utc::now()), + ..Default::default() + }; + merge_metadata_update( + state.metadata_updates.entry(params.thread_id).or_default(), + update, + ); state.names.insert(params.thread_id, Some(name)); } stored_thread_from_state(&state, params.thread_id, /*include_history*/ false) @@ -307,37 +471,178 @@ fn stored_thread_from_state( items: history_items.clone(), }); let name = state.names.get(&thread_id).cloned().flatten(); + let git_info = state.git_infos.get(&thread_id).cloned().flatten(); + let update = state.metadata_updates.get(&thread_id); + let name = update + .and_then(|update| update.name.clone()) + .unwrap_or(name); + let git_info = update + .and_then(|update| update.git_info.clone()) + .unwrap_or(git_info); let rollout_path = state - .rollout_paths - .iter() - .find_map(|(path, mapped_thread_id)| { - (*mapped_thread_id == thread_id).then(|| path.clone()) + .metadata_updates + .get(&thread_id) + .and_then(|update| update.rollout_path.clone()) + .or_else(|| { + state + .rollout_paths + .iter() + .find_map(|(path, mapped_thread_id)| { + (*mapped_thread_id == thread_id).then(|| path.clone()) + }) }); Ok(StoredThread { thread_id, rollout_path, forked_from_id: created.forked_from_id, - preview: String::new(), + preview: update + .and_then(|update| update.preview.clone()) + .unwrap_or_default(), name, - model_provider: "test".to_string(), - model: None, - reasoning_effort: None, - created_at: Utc::now(), - updated_at: Utc::now(), + model_provider: update + .and_then(|update| update.model_provider.clone()) + .unwrap_or_else(|| "test".to_string()), + model: update + .and_then(|update| update.model.clone()) + .unwrap_or_default(), + reasoning_effort: update + .and_then(|update| update.reasoning_effort) + .unwrap_or_default(), + created_at: update + .and_then(|update| update.created_at) + .unwrap_or_else(Utc::now), + updated_at: update + .and_then(|update| update.updated_at) + .unwrap_or_else(Utc::now), archived_at: None, - cwd: PathBuf::new(), - cli_version: "test".to_string(), - source: created.source.clone(), - thread_source: created.thread_source, - agent_nickname: None, - agent_role: None, - agent_path: None, - git_info: None, - approval_mode: AskForApproval::Never, - sandbox_policy: SandboxPolicy::new_read_only_policy(), - token_usage: None, - first_user_message: None, + cwd: update + .and_then(|update| update.cwd.clone()) + .unwrap_or_default(), + cli_version: update + .and_then(|update| update.cli_version.clone()) + .unwrap_or_else(|| "test".to_string()), + source: update + .and_then(|update| update.source.clone()) + .unwrap_or_else(|| created.source.clone()), + thread_source: update + .and_then(|update| update.thread_source) + .unwrap_or(created.thread_source), + agent_nickname: update + .and_then(|update| update.agent_nickname.clone()) + .unwrap_or_default(), + agent_role: update + .and_then(|update| update.agent_role.clone()) + .unwrap_or_default(), + agent_path: update + .and_then(|update| update.agent_path.clone()) + .unwrap_or_default(), + git_info, + approval_mode: update + .and_then(|update| update.approval_mode) + .unwrap_or(AskForApproval::Never), + sandbox_policy: update + .and_then(|update| update.sandbox_policy.clone()) + .unwrap_or_else(SandboxPolicy::new_read_only_policy), + token_usage: update + .and_then(|update| update.token_usage.clone()) + .unwrap_or_default(), + first_user_message: update + .and_then(|update| update.first_user_message.clone()) + .unwrap_or_default(), history, }) } + +fn merge_metadata_update(target: &mut ThreadMetadataUpdate, update: ThreadMetadataUpdate) { + let ThreadMetadataUpdate { + rollout_path, + preview, + name, + model_provider, + model, + reasoning_effort, + created_at, + updated_at, + source, + thread_source, + agent_nickname, + agent_role, + agent_path, + cwd, + cli_version, + approval_mode, + sandbox_policy, + token_usage, + first_user_message, + git_info, + memory_mode, + dynamic_tools, + } = update; + if rollout_path.is_some() { + target.rollout_path = rollout_path; + } + if preview.is_some() { + target.preview = preview; + } + if name.is_some() { + target.name = name; + } + if model_provider.is_some() { + target.model_provider = model_provider; + } + if model.is_some() { + target.model = model; + } + if reasoning_effort.is_some() { + target.reasoning_effort = reasoning_effort; + } + if created_at.is_some() { + target.created_at = created_at; + } + if updated_at.is_some() { + target.updated_at = updated_at; + } + if source.is_some() { + target.source = source; + } + if thread_source.is_some() { + target.thread_source = thread_source; + } + if agent_nickname.is_some() { + target.agent_nickname = agent_nickname; + } + if agent_role.is_some() { + target.agent_role = agent_role; + } + if agent_path.is_some() { + target.agent_path = agent_path; + } + if cwd.is_some() { + target.cwd = cwd; + } + if cli_version.is_some() { + target.cli_version = cli_version; + } + if approval_mode.is_some() { + target.approval_mode = approval_mode; + } + if sandbox_policy.is_some() { + target.sandbox_policy = sandbox_policy; + } + if token_usage.is_some() { + target.token_usage = token_usage; + } + if first_user_message.is_some() { + target.first_user_message = first_user_message; + } + if git_info.is_some() { + target.git_info = git_info; + } + if memory_mode.is_some() { + target.memory_mode = memory_mode; + } + if dynamic_tools.is_some() { + target.dynamic_tools = dynamic_tools; + } +} diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index c7f6b4eaed..3c3a179c4d 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -21,6 +21,7 @@ pub use local::LocalThreadStore; pub use local::LocalThreadStoreConfig; pub use store::ThreadStore; pub use types::AppendThreadItemsParams; +pub use types::ApplyThreadMetadataParams; pub use types::ArchiveThreadParams; pub use types::CreateThreadParams; pub use types::GitInfoPatch; @@ -42,6 +43,7 @@ pub use types::StoredTurnItemsView; pub use types::StoredTurnStatus; pub use types::ThreadEventPersistenceMode; pub use types::ThreadMetadataPatch; +pub use types::ThreadMetadataUpdate; pub use types::ThreadPage; pub use types::ThreadPersistenceMetadata; pub use types::ThreadSortKey; diff --git a/codex-rs/thread-store/src/local/apply_thread_metadata.rs b/codex-rs/thread-store/src/local/apply_thread_metadata.rs new file mode 100644 index 0000000000..8fdd77a82d --- /dev/null +++ b/codex-rs/thread-store/src/local/apply_thread_metadata.rs @@ -0,0 +1,558 @@ +use chrono::Utc; +use codex_protocol::ThreadId; +use codex_protocol::protocol::GitInfo; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::ThreadMemoryMode; +use codex_rollout::append_thread_name; +use codex_state::ThreadMetadata; + +use super::LocalThreadStore; +use super::helpers::rollout_path_is_archived; +use super::live_writer; +use crate::ApplyThreadMetadataParams; +use crate::ThreadMetadataUpdate; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +pub(super) async fn apply_thread_metadata( + store: &LocalThreadStore, + params: ApplyThreadMetadataParams, +) -> ThreadStoreResult<()> { + let thread_id = params.thread_id; + let update = params.update; + let _metadata_permit = store.acquire_metadata_permit(thread_id).await?; + let Some(state_db) = store.state_db().await else { + if let Some(name) = update.name.as_ref() { + append_thread_name_update(store, thread_id, name.as_deref().unwrap_or_default()) + .await?; + } + return Ok(()); + }; + flush_live_writer_if_present(store, thread_id).await?; + let existing = + state_db + .get_thread(thread_id) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to read thread metadata for {thread_id}: {err}"), + })?; + let metadata = merge_thread_metadata( + store, + thread_id, + existing, + update.clone(), + update.memory_mode, + ) + .await?; + state_db + .upsert_thread(&metadata) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to apply thread metadata for {thread_id}: {err}"), + })?; + if let Some(name) = update.name.as_ref() { + append_thread_name_update(store, thread_id, name.as_deref().unwrap_or_default()).await?; + } + if let Some(git_info) = update.git_info.as_ref() { + apply_git_info_update(state_db.as_ref(), thread_id, git_info.as_ref()).await?; + } + if let Some(memory_mode) = update.memory_mode { + state_db + .set_thread_memory_mode(thread_id, memory_mode_to_str(memory_mode)) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to apply thread memory mode for {thread_id}: {err}"), + })?; + } + if let Some(dynamic_tools) = update.dynamic_tools.as_deref() { + state_db + .replace_dynamic_tools(thread_id, dynamic_tools) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to replace dynamic tools for {thread_id}: {err}"), + })?; + } + Ok(()) +} + +async fn append_thread_name_update( + store: &LocalThreadStore, + thread_id: ThreadId, + name: &str, +) -> ThreadStoreResult<()> { + append_thread_name(store.config.codex_home.as_path(), thread_id, name) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to update thread name index: {err}"), + }) +} + +async fn flush_live_writer_if_present( + store: &LocalThreadStore, + thread_id: ThreadId, +) -> ThreadStoreResult<()> { + match live_writer::flush_thread(store, thread_id).await { + Ok(()) | Err(ThreadStoreError::ThreadNotFound { .. }) => Ok(()), + Err(err) => Err(err), + } +} + +async fn apply_git_info_update( + state_db: &codex_state::StateRuntime, + thread_id: ThreadId, + git_info: Option<&GitInfo>, +) -> ThreadStoreResult<()> { + let (git_sha, git_branch, git_origin_url) = match git_info { + Some(git) => ( + git.commit_hash.as_ref().map(|sha| sha.0.as_str()), + git.branch.as_deref(), + git.repository_url.as_deref(), + ), + None => (Some(""), Some(""), Some("")), + }; + state_db + .update_thread_git_info( + thread_id, + Some(git_sha), + Some(git_branch), + Some(git_origin_url), + ) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to apply thread git info for {thread_id}: {err}"), + })?; + Ok(()) +} + +async fn merge_thread_metadata( + store: &LocalThreadStore, + thread_id: ThreadId, + existing: Option, + update: ThreadMetadataUpdate, + memory_mode: Option, +) -> ThreadStoreResult { + let live_rollout_path = live_writer::rollout_path(store, thread_id).await.ok(); + let rollout_path = update + .rollout_path + .clone() + .or_else(|| { + existing + .as_ref() + .map(|metadata| metadata.rollout_path.clone()) + }) + .or(live_rollout_path) + .ok_or_else(|| ThreadStoreError::Internal { + message: format!("thread metadata update missing rollout path for {thread_id}"), + })?; + let now = Utc::now(); + let created_at = update + .created_at + .or_else(|| existing.as_ref().map(|metadata| metadata.created_at)) + .unwrap_or(now); + let updated_at = update + .updated_at + .or_else(|| existing.as_ref().map(|metadata| metadata.updated_at)) + .unwrap_or(created_at); + let tokens_used = token_count_from_update(&update) + .or_else(|| existing.as_ref().map(|metadata| metadata.tokens_used)) + .unwrap_or_default(); + let archived_at = existing + .as_ref() + .and_then(|metadata| metadata.archived_at) + .or_else(|| { + rollout_path_is_archived(store.config.codex_home.as_path(), &rollout_path) + .then_some(updated_at) + }); + let mut metadata = ThreadMetadata { + id: thread_id, + rollout_path, + created_at, + updated_at, + source: update + .source + .as_ref() + .map(enum_to_string) + .or_else(|| existing.as_ref().map(|metadata| metadata.source.clone())) + .unwrap_or_else(|| enum_to_string(&SessionSource::Unknown)), + thread_source: nested_or_existing( + update.thread_source, + existing + .as_ref() + .and_then(|metadata| metadata.thread_source), + ), + agent_nickname: nested_or_existing( + update.agent_nickname, + existing + .as_ref() + .and_then(|metadata| metadata.agent_nickname.clone()), + ), + agent_role: nested_or_existing( + update.agent_role, + existing + .as_ref() + .and_then(|metadata| metadata.agent_role.clone()), + ), + agent_path: nested_or_existing( + update.agent_path, + existing + .as_ref() + .and_then(|metadata| metadata.agent_path.clone()), + ), + model_provider: update + .model_provider + .or_else(|| { + existing + .as_ref() + .map(|metadata| metadata.model_provider.clone()) + }) + .unwrap_or_else(|| store.config.default_model_provider_id.clone()), + model: nested_or_existing( + update.model, + existing + .as_ref() + .and_then(|metadata| metadata.model.clone()), + ), + reasoning_effort: nested_or_existing( + update.reasoning_effort, + existing + .as_ref() + .and_then(|metadata| metadata.reasoning_effort), + ), + cwd: update + .cwd + .or_else(|| existing.as_ref().map(|metadata| metadata.cwd.clone())) + .unwrap_or_default(), + cli_version: update + .cli_version + .or_else(|| { + existing + .as_ref() + .map(|metadata| metadata.cli_version.clone()) + }) + .unwrap_or_default(), + title: existing + .as_ref() + .map(|metadata| metadata.title.clone()) + .unwrap_or_default(), + sandbox_policy: update + .sandbox_policy + .as_ref() + .map(enum_to_string) + .or_else(|| { + existing + .as_ref() + .map(|metadata| metadata.sandbox_policy.clone()) + }) + .unwrap_or_default(), + approval_mode: update + .approval_mode + .as_ref() + .map(enum_to_string) + .or_else(|| { + existing + .as_ref() + .map(|metadata| metadata.approval_mode.clone()) + }) + .unwrap_or_default(), + tokens_used, + first_user_message: nested_or_existing( + update.first_user_message, + existing + .as_ref() + .and_then(|metadata| metadata.first_user_message.clone()), + ), + archived_at, + git_sha: None, + git_branch: None, + git_origin_url: None, + }; + let name_was_explicit = update.name.is_some(); + if let Some(name) = update.name { + metadata.title = name.unwrap_or_default(); + } else if metadata.title.is_empty() + && let Some(preview) = update + .preview + .or_else(|| metadata.first_user_message.clone()) + { + metadata.title = preview; + } + apply_git_info(&mut metadata, update.git_info, existing.as_ref()); + if memory_mode == Some(ThreadMemoryMode::Disabled) + && !name_was_explicit + && metadata.title.is_empty() + { + metadata.title = existing + .as_ref() + .map(|metadata| metadata.title.clone()) + .unwrap_or_default(); + } + metadata.cwd = codex_rollout::state_db::normalize_cwd_for_state_db(&metadata.cwd); + Ok(metadata) +} + +fn apply_git_info( + metadata: &mut ThreadMetadata, + update: Option>, + existing: Option<&ThreadMetadata>, +) { + match update { + Some(Some(git)) => { + metadata.git_sha = git.commit_hash.map(|sha| sha.0); + metadata.git_branch = git.branch; + metadata.git_origin_url = git.repository_url; + } + Some(None) => {} + None => { + metadata.git_sha = existing.and_then(|metadata| metadata.git_sha.clone()); + metadata.git_branch = existing.and_then(|metadata| metadata.git_branch.clone()); + metadata.git_origin_url = existing.and_then(|metadata| metadata.git_origin_url.clone()); + } + } +} + +fn token_count_from_update(update: &ThreadMetadataUpdate) -> Option { + match update.token_usage.as_ref() { + Some(Some(usage)) => Some(usage.total_tokens.max(0)), + Some(None) => Some(0), + None => None, + } +} + +fn nested_or_existing(update: Option>, existing: Option) -> Option { + match update { + Some(value) => value, + None => existing, + } +} + +fn memory_mode_to_str(mode: ThreadMemoryMode) -> &'static str { + match mode { + ThreadMemoryMode::Enabled => "enabled", + ThreadMemoryMode::Disabled => "disabled", + } +} + +fn enum_to_string(value: &T) -> String { + match serde_json::to_value(value) { + Ok(serde_json::Value::String(s)) => s, + Ok(other) => other.to_string(), + Err(_) => String::new(), + } +} + +#[cfg(test)] +mod tests { + use chrono::TimeZone; + use codex_protocol::dynamic_tools::DynamicToolSpec; + use codex_protocol::protocol::GitInfo; + use codex_protocol::protocol::TokenUsage; + use pretty_assertions::assert_eq; + use tempfile::TempDir; + + use super::*; + use crate::ThreadStore; + use crate::local::LocalThreadStore; + use crate::local::test_support::test_config; + + #[tokio::test] + async fn apply_thread_metadata_updates_sqlite_metadata_and_side_tables() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config, Some(runtime.clone())); + let thread_id = ThreadId::default(); + let rollout_path = home.path().join("rollout.jsonl"); + let created_at = Utc + .with_ymd_and_hms(2026, 1, 2, 3, 4, 5) + .single() + .expect("valid timestamp"); + let updated_at = Utc + .with_ymd_and_hms(2026, 1, 2, 4, 5, 6) + .single() + .expect("valid timestamp"); + + store + .apply_thread_metadata(ApplyThreadMetadataParams { + thread_id, + update: ThreadMetadataUpdate { + rollout_path: Some(rollout_path.clone()), + preview: Some("first preview".to_string()), + name: Some(Some("named thread".to_string())), + model_provider: Some("test-provider".to_string()), + model: Some(Some("test-model".to_string())), + created_at: Some(created_at), + updated_at: Some(updated_at), + source: Some(SessionSource::Exec), + cwd: Some(home.path().to_path_buf()), + cli_version: Some("test-version".to_string()), + token_usage: Some(Some(TokenUsage { + input_tokens: 40, + cached_input_tokens: 0, + output_tokens: 60, + reasoning_output_tokens: 23, + total_tokens: 123, + })), + first_user_message: Some(Some("first preview".to_string())), + git_info: Some(Some(GitInfo { + commit_hash: Some(codex_git_utils::GitSha::new("abcdef")), + branch: Some("main".to_string()), + repository_url: Some("https://example.com/repo.git".to_string()), + })), + memory_mode: Some(ThreadMemoryMode::Disabled), + dynamic_tools: Some(vec![DynamicToolSpec { + namespace: Some("test".to_string()), + name: "tool".to_string(), + description: "tool description".to_string(), + input_schema: serde_json::json!({"type": "object"}), + defer_loading: true, + }]), + ..Default::default() + }, + }) + .await + .expect("apply metadata"); + + let metadata = runtime + .get_thread(thread_id) + .await + .expect("read metadata") + .expect("metadata stored"); + assert_eq!(metadata.id, thread_id); + assert_eq!(metadata.rollout_path, rollout_path); + assert_eq!(metadata.title, "named thread"); + assert_eq!( + metadata.first_user_message.as_deref(), + Some("first preview") + ); + assert_eq!(metadata.model_provider, "test-provider"); + assert_eq!(metadata.model.as_deref(), Some("test-model")); + assert_eq!(metadata.created_at, created_at); + assert_eq!(metadata.updated_at, updated_at); + assert_eq!(metadata.cli_version, "test-version"); + assert_eq!(metadata.tokens_used, 123); + assert_eq!(metadata.git_sha.as_deref(), Some("abcdef")); + assert_eq!(metadata.git_branch.as_deref(), Some("main")); + assert_eq!( + metadata.git_origin_url.as_deref(), + Some("https://example.com/repo.git") + ); + + let memory_mode = runtime + .get_thread_memory_mode(thread_id) + .await + .expect("read memory mode"); + assert_eq!(memory_mode.as_deref(), Some("disabled")); + let dynamic_tools = runtime + .get_dynamic_tools(thread_id) + .await + .expect("read dynamic tools") + .expect("dynamic tools stored"); + assert_eq!(dynamic_tools.len(), 1); + assert_eq!(dynamic_tools[0].name, "tool"); + + store + .apply_thread_metadata(ApplyThreadMetadataParams { + thread_id, + update: ThreadMetadataUpdate { + name: Some(None), + git_info: Some(None), + token_usage: Some(None), + dynamic_tools: Some(Vec::new()), + ..Default::default() + }, + }) + .await + .expect("clear metadata"); + + let metadata = runtime + .get_thread(thread_id) + .await + .expect("read metadata") + .expect("metadata stored"); + assert_eq!(metadata.title, ""); + assert_eq!(metadata.tokens_used, 0); + assert_eq!(metadata.git_sha.as_deref(), Some("")); + assert_eq!(metadata.git_branch.as_deref(), Some("")); + assert_eq!(metadata.git_origin_url.as_deref(), Some("")); + let dynamic_tools = runtime + .get_dynamic_tools(thread_id) + .await + .expect("read dynamic tools"); + assert_eq!(dynamic_tools, None); + } + + #[tokio::test] + async fn apply_thread_metadata_without_sqlite_updates_name_index() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let thread_id = ThreadId::default(); + + store + .apply_thread_metadata(ApplyThreadMetadataParams { + thread_id, + update: ThreadMetadataUpdate { + name: Some(Some("index-only name".to_string())), + ..Default::default() + }, + }) + .await + .expect("apply metadata"); + + let name = codex_rollout::find_thread_name_by_id(home.path(), &thread_id) + .await + .expect("read thread name"); + assert_eq!(name.as_deref(), Some("index-only name")); + + store + .apply_thread_metadata(ApplyThreadMetadataParams { + thread_id, + update: ThreadMetadataUpdate { + name: Some(None), + ..Default::default() + }, + }) + .await + .expect("clear metadata"); + + let name = codex_rollout::find_thread_name_by_id(home.path(), &thread_id) + .await + .expect("read thread name"); + assert_eq!(name.as_deref(), Some("")); + } + + #[tokio::test] + async fn apply_thread_metadata_does_not_index_name_when_validation_fails() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let runtime = codex_state::StateRuntime::init( + home.path().to_path_buf(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config, Some(runtime)); + let thread_id = ThreadId::default(); + + let err = store + .apply_thread_metadata(ApplyThreadMetadataParams { + thread_id, + update: ThreadMetadataUpdate { + name: Some(Some("should not be indexed".to_string())), + ..Default::default() + }, + }) + .await + .expect_err("metadata update should fail without a rollout path"); + + assert!(matches!(err, ThreadStoreError::Internal { .. })); + let name = codex_rollout::find_thread_name_by_id(home.path(), &thread_id) + .await + .expect("read thread name"); + assert_eq!(name, None); + } +} diff --git a/codex-rs/thread-store/src/local/create_thread.rs b/codex-rs/thread-store/src/local/create_thread.rs index d181149406..897e2518af 100644 --- a/codex-rs/thread-store/src/local/create_thread.rs +++ b/codex-rs/thread-store/src/local/create_thread.rs @@ -30,15 +30,16 @@ pub(super) async fn create_thread( let state_db_ctx = store.state_db().await; let recorder = RolloutRecorder::new( &config, - RolloutRecorderParams::new( - params.thread_id, - params.forked_from_id, - params.source, - params.thread_source, - params.base_instructions, - params.dynamic_tools, - event_persistence_mode(params.event_persistence_mode), - ), + RolloutRecorderParams::Create { + conversation_id: params.thread_id, + forked_from_id: params.forked_from_id, + source: params.source, + originator: params.originator, + thread_source: params.thread_source, + base_instructions: params.base_instructions, + dynamic_tools: params.dynamic_tools, + event_persistence_mode: event_persistence_mode(params.event_persistence_mode), + }, state_db_ctx, /*state_builder*/ None, ) diff --git a/codex-rs/thread-store/src/local/helpers.rs b/codex-rs/thread-store/src/local/helpers.rs index bb46287123..9d40ba97b4 100644 --- a/codex-rs/thread-store/src/local/helpers.rs +++ b/codex-rs/thread-store/src/local/helpers.rs @@ -170,6 +170,9 @@ pub(super) fn git_info_from_parts( branch: Option, origin_url: Option, ) -> Option { + let sha = non_empty_string(sha); + let branch = non_empty_string(branch); + let origin_url = non_empty_string(origin_url); if sha.is_none() && branch.is_none() && origin_url.is_none() { return None; } @@ -180,6 +183,10 @@ pub(super) fn git_info_from_parts( }) } +pub(super) fn non_empty_string(value: Option) -> Option { + value.filter(|value| !value.is_empty()) +} + fn thread_id_from_rollout_path(path: &Path) -> Option { let file_name = path.file_name()?.to_str()?; let stem = file_name.strip_suffix(".jsonl")?; diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 07aa5e925f..b2a35eefa8 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -1,3 +1,4 @@ +mod apply_thread_metadata; mod archive_thread; mod create_thread; mod helpers; @@ -19,8 +20,11 @@ use std::collections::hash_map::Entry; use std::path::PathBuf; use std::sync::Arc; use tokio::sync::Mutex; +use tokio::sync::OwnedSemaphorePermit; +use tokio::sync::Semaphore; use crate::AppendThreadItemsParams; +use crate::ApplyThreadMetadataParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::ListThreadsParams; @@ -41,6 +45,7 @@ use crate::UpdateThreadMetadataParams; pub struct LocalThreadStore { pub(super) config: LocalThreadStoreConfig, live_recorders: Arc>>, + metadata_apply_semaphore: Arc, state_db: Option, } @@ -80,6 +85,7 @@ impl LocalThreadStore { Self { config, live_recorders: Arc::new(Mutex::new(HashMap::new())), + metadata_apply_semaphore: Arc::new(Semaphore::new(1)), state_db, } } @@ -122,6 +128,19 @@ impl LocalThreadStore { .ok_or(ThreadStoreError::ThreadNotFound { thread_id }) } + pub(super) async fn acquire_metadata_permit( + &self, + thread_id: ThreadId, + ) -> ThreadStoreResult { + self.metadata_apply_semaphore + .clone() + .acquire_owned() + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("metadata semaphore closed for thread {thread_id}: {err}"), + }) + } + pub(super) async fn ensure_live_recorder_absent( &self, thread_id: ThreadId, @@ -169,6 +188,13 @@ impl ThreadStore for LocalThreadStore { live_writer::append_items(self, params).await } + async fn apply_thread_metadata( + &self, + params: ApplyThreadMetadataParams, + ) -> ThreadStoreResult<()> { + apply_thread_metadata::apply_thread_metadata(self, params).await + } + async fn persist_thread(&self, thread_id: ThreadId) -> ThreadStoreResult<()> { live_writer::persist_thread(self, thread_id).await } @@ -738,6 +764,7 @@ mod tests { thread_id, forked_from_id: None, source: SessionSource::Exec, + originator: "test_originator".to_string(), thread_source: None, base_instructions: BaseInstructions::default(), dynamic_tools: Vec::new(), diff --git a/codex-rs/thread-store/src/local/read_thread.rs b/codex-rs/thread-store/src/local/read_thread.rs index 9d685aace6..7dce05515f 100644 --- a/codex-rs/thread-store/src/local/read_thread.rs +++ b/codex-rs/thread-store/src/local/read_thread.rs @@ -15,6 +15,7 @@ use codex_state::ThreadMetadata; use super::LocalThreadStore; use super::helpers::distinct_thread_metadata_title; use super::helpers::git_info_from_parts; +use super::helpers::non_empty_string; use super::helpers::rollout_path_is_archived; use super::helpers::set_thread_name_from_title; use super::helpers::stored_thread_from_rollout_item; @@ -109,20 +110,7 @@ pub(super) async fn read_thread_by_rollout_path( }); } if let Some(metadata) = read_sqlite_metadata(store, thread.thread_id).await { - let existing_git_info = thread.git_info.take(); - let (fallback_sha, fallback_branch, fallback_origin_url) = match existing_git_info { - Some(info) => ( - info.commit_hash.map(|sha| sha.0), - info.branch, - info.repository_url, - ), - None => (None, None, None), - }; - 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), - ); + thread.git_info = sqlite_git_info_with_rollout_fallback(&metadata, thread.git_info); } attach_history_if_requested(&mut thread, include_history).await?; Ok(thread) @@ -273,7 +261,8 @@ async fn stored_thread_from_sqlite_metadata( None => find_thread_name_by_id(store.config.codex_home.as_path(), &metadata.id) .await .ok() - .flatten(), + .flatten() + .and_then(|title| non_empty_string(Some(title))), }; let session_meta = read_session_meta_line(metadata.rollout_path.as_path()) .await @@ -319,6 +308,38 @@ async fn stored_thread_from_sqlite_metadata( } } +fn sqlite_git_info_with_rollout_fallback( + metadata: &ThreadMetadata, + rollout_git_info: Option, +) -> Option { + let Some(rollout_git_info) = rollout_git_info else { + return git_info_from_parts( + metadata.git_sha.clone(), + metadata.git_branch.clone(), + metadata.git_origin_url.clone(), + ); + }; + git_info_from_parts( + sqlite_git_part( + metadata.git_sha.clone(), + rollout_git_info.commit_hash.map(|sha| sha.0), + ), + sqlite_git_part(metadata.git_branch.clone(), rollout_git_info.branch), + sqlite_git_part( + metadata.git_origin_url.clone(), + rollout_git_info.repository_url, + ), + ) +} + +fn sqlite_git_part(sqlite_value: Option, rollout_value: Option) -> Option { + match sqlite_value.as_deref() { + Some("") => None, + Some(_) => sqlite_value, + None => rollout_value, + } +} + async fn stored_thread_from_session_meta( store: &LocalThreadStore, path: std::path::PathBuf, @@ -527,6 +548,48 @@ mod tests { ); } + #[tokio::test] + async fn read_thread_by_rollout_path_respects_sqlite_git_clear() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let uuid = Uuid::from_u128(224); + 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"); + let runtime = codex_state::StateRuntime::init( + config.sqlite_home.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config.clone(), Some(runtime.clone())); + let builder = ThreadMetadataBuilder::new( + thread_id, + active_path.clone(), + Utc::now(), + SessionSource::Cli, + ); + runtime + .upsert_thread(&builder.build(config.default_model_provider_id.as_str())) + .await + .expect("state db upsert should succeed"); + runtime + .update_thread_git_info(thread_id, Some(Some("")), Some(Some("")), Some(Some(""))) + .await + .expect("state db git clear should succeed"); + + let thread = store + .read_thread_by_rollout_path( + active_path, + /*include_archived*/ false, + /*include_history*/ false, + ) + .await + .expect("read thread by rollout path"); + + assert!(thread.git_info.is_none()); + } + #[tokio::test] async fn read_thread_returns_archived_rollout_when_requested() { let home = TempDir::new().expect("temp dir"); @@ -764,6 +827,46 @@ mod tests { assert_eq!(thread.name, Some("Legacy title".to_string())); } + #[tokio::test] + async fn read_thread_treats_legacy_empty_name_as_absent() { + let home = TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let uuid = Uuid::from_u128(225); + 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.default_model_provider_id.clone(), + ) + .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); + builder.model_provider = Some(config.default_model_provider_id.clone()); + let mut metadata = builder.build(config.default_model_provider_id.as_str()); + metadata.first_user_message = Some("Hello from user".to_string()); + runtime + .upsert_thread(&metadata) + .await + .expect("state db upsert should succeed"); + codex_rollout::append_thread_name(home.path(), thread_id, "") + .await + .expect("append legacy empty thread name"); + + let thread = store + .read_thread(ReadThreadParams { + thread_id, + include_archived: false, + include_history: false, + }) + .await + .expect("read thread"); + + assert_eq!(thread.name, None); + } + #[tokio::test] async fn read_thread_uses_sqlite_metadata_for_rollout_without_user_preview() { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index bd5e3e7d3d..6a232d82ce 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -3,6 +3,7 @@ use codex_protocol::ThreadId; use std::any::Any; use crate::AppendThreadItemsParams; +use crate::ApplyThreadMetadataParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::ItemPage; @@ -33,9 +34,22 @@ pub trait ThreadStore: Any + Send + Sync { /// Reopens an existing thread for live appends. async fn resume_thread(&self, params: ResumeThreadParams) -> ThreadStoreResult<()>; - /// Appends items to a live thread. + /// Appends canonical rollout items to a live thread. + /// + /// This is the raw history API. It does not infer or persist metadata by itself; callers that + /// need metadata correctness should go through [`crate::LiveThread`] or explicitly call + /// [`ThreadStore::apply_thread_metadata`] with metadata derived above the store. async fn append_items(&self, params: AppendThreadItemsParams) -> ThreadStoreResult<()>; + /// Applies metadata inferred from canonical live thread items. + /// + /// Raw item appends remain the canonical history API. Implementations must persist the + /// explicit metadata facts prepared by the live thread layer. + async fn apply_thread_metadata( + &self, + params: ApplyThreadMetadataParams, + ) -> ThreadStoreResult<()>; + /// Materializes the thread if persistence is lazy, then persists all queued items. async fn persist_thread(&self, thread_id: ThreadId) -> ThreadStoreResult<()>; diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 1fed7bc829..38967aba6b 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -49,6 +49,8 @@ pub struct CreateThreadParams { pub forked_from_id: Option, /// Runtime source for the thread. pub source: SessionSource, + /// Product originator captured in the synthesized session metadata. + pub originator: String, /// Optional analytics source classification for this thread. pub thread_source: Option, /// Base instructions persisted in session metadata. @@ -87,6 +89,67 @@ pub struct AppendThreadItemsParams { pub items: Vec, } +/// Metadata fields derived from canonical rollout items by a live thread metadata handler. +/// +/// Each field is optional so callers can send only the metadata facts learned from a batch of +/// accepted items. Nested `Option` fields distinguish omitted/no-op fields from explicit clears. +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct ThreadMetadataUpdate { + /// Local rollout path when the backing store is filesystem-based. + pub rollout_path: Option, + /// Best available user-facing preview, usually the first user message. + pub preview: Option, + /// Optional user-facing thread name/title. + pub name: Option>, + /// Model provider id associated with the thread. + pub model_provider: Option, + /// Latest observed model, if known. + pub model: Option>, + /// Latest observed reasoning effort, if known. + pub reasoning_effort: Option>, + /// Thread creation timestamp. + pub created_at: Option>, + /// Thread last-update timestamp. + pub updated_at: Option>, + /// Runtime source for the thread. + pub source: Option, + /// Optional analytics source classification for this thread. + pub thread_source: Option>, + /// Optional random nickname for thread-spawn sub-agents. + pub agent_nickname: Option>, + /// Optional role for thread-spawn sub-agents. + pub agent_role: Option>, + /// Optional canonical path for thread-spawn sub-agents. + pub agent_path: Option>, + /// Working directory captured for the thread. + pub cwd: Option, + /// CLI version captured for the thread. + pub cli_version: Option, + /// Approval mode captured for the thread. + pub approval_mode: Option, + /// Sandbox policy captured for the thread. + pub sandbox_policy: Option, + /// Last observed token usage. + pub token_usage: Option>, + /// First user message observed for this thread, if any. + pub first_user_message: Option>, + /// Optional Git metadata captured for the thread. + pub git_info: Option>, + /// Explicit thread memory behavior. + pub memory_mode: Option, + /// Dynamic tools available to the thread. + pub dynamic_tools: Option>, +} + +/// Parameters for applying metadata inferred by the live thread layer. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct ApplyThreadMetadataParams { + /// Thread id to update. + pub thread_id: ThreadId, + /// Metadata facts to apply. + pub update: ThreadMetadataUpdate, +} + /// Parameters for loading persisted history for resume, fork, rollback, and memory jobs. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct LoadThreadHistoryParams { @@ -243,7 +306,7 @@ pub struct StoredTurn { pub items: Vec, /// Amount of item detail included in `items`. pub items_view: StoredTurnItemsView, - /// Store-owned status for API layer projection. + /// Store-owned status used by API layer summaries. pub status: StoredTurnStatus, /// Error message when the turn failed. pub error: Option,