openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add thread metadata store API

Browse Source
This commit is contained in:
Tom Wiltzius
2026-05-11 15:27:13 -07:00
parent bd42660cb4
commit fca2a68f85
25 changed files with 1327 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
-1
View File
@@ -3465,7 +3465,6 @@ dependencies = [
"chrono",
"codex-file-search",
"codex-git-utils",
"codex-login",
"codex-otel",
"codex-protocol",
"codex-state",
codex-rs/app-server/tests/suite/conversation_summary.rs
+1
View File
@@ -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(),
codex-rs/app-server/tests/suite/v2/thread_read.rs
+1
View File
@@ -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(),
codex-rs/app-server/tests/suite/v2/thread_unarchive.rs
+1
View File
@@ -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(),
codex-rs/core/src/session/session.rs
+1
View File
@@ -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(),
codex-rs/core/src/session/tests.rs
+2
View File
@@ -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(),
codex-rs/core/tests/suite/rollout_list_find.rs
+10 -9
View File
@@ -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,
)
codex-rs/rollout/Cargo.toml
-1
View File
@@ -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 }
codex-rs/rollout/src/lib.rs
+2 -4
View File
@@ -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;
codex-rs/rollout/src/policy.rs
+41
View File
@@ -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<RolloutItem> {
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 {
codex-rs/rollout/src/recorder.rs
+3 -22
View File
@@ -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<ThreadId>,
source: SessionSource,
originator: String,
thread_source: Option<ThreadSource>,
base_instructions: BaseInstructions,
dynamic_tools: Vec<DynamicToolSpec>,
@@ -165,26 +165,6 @@ fn clone_io_error(err: &IoError) -> IoError {
}
impl RolloutRecorderParams {
pub fn new(
conversation_id: ThreadId,
forked_from_id: Option<ThreadId>,
source: SessionSource,
thread_source: Option<ThreadSource>,
base_instructions: BaseInstructions,
dynamic_tools: Vec<DynamicToolSpec>,
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(),
codex-rs/rollout/src/recorder_tests.rs
+30 -27
View File
@@ -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,
)
codex-rs/rollout/src/session_index.rs
+7 -3
View File
@@ -103,9 +103,13 @@ pub async fn find_thread_names_by_ids(
let Ok(entry) = serde_json::from_str::<SessionIndexEntry>(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());
}
}
}
codex-rs/rollout/src/session_index_tests.rs
+27
View File
@@ -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()?;
codex-rs/state/src/runtime/threads.rs
+45
View File
@@ -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,
codex-rs/thread-store/README.md
+28
View File
@@ -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.
codex-rs/thread-store/src/in_memory.rs
+327 -22
View File
@@ -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<HashMap<String, Arc<InMemoryThreadStore>>> {
#[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<String, Arc<InMemoryThreadStore>>> {
@@ -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<ThreadId, CreateThreadParams>,
histories: HashMap<ThreadId, Vec<RolloutItem>>,
metadata_updates: HashMap<ThreadId, ThreadMetadataUpdate>,
names: HashMap<ThreadId, Option<String>>,
git_infos: HashMap<ThreadId, Option<GitInfo>>,
rollout_paths: HashMap<PathBuf, ThreadId>,
}
@@ -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(&params.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;
}
}
codex-rs/thread-store/src/lib.rs
+2
View File
@@ -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;
codex-rs/thread-store/src/local/apply_thread_metadata.rs
+558
View File
@@ -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<ThreadMetadata>,
update: ThreadMetadataUpdate,
memory_mode: Option<ThreadMemoryMode>,
) -> ThreadStoreResult<ThreadMetadata> {
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<Option<GitInfo>>,
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<i64> {
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<T>(update: Option<Option<T>>, existing: Option<T>) -> Option<T> {
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<T: serde::Serialize>(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);
}
}
codex-rs/thread-store/src/local/create_thread.rs
+10 -9
View File
@@ -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,
)
codex-rs/thread-store/src/local/helpers.rs
+7
View File
@@ -170,6 +170,9 @@ pub(super) fn git_info_from_parts(
branch: Option<String>,
origin_url: Option<String>,
) -> Option<GitInfo> {
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<String>) -> Option<String> {
value.filter(|value| !value.is_empty())
}
fn thread_id_from_rollout_path(path: &Path) -> Option<ThreadId> {
let file_name = path.file_name()?.to_str()?;
let stem = file_name.strip_suffix(".jsonl")?;
codex-rs/thread-store/src/local/mod.rs
+27
View File
@@ -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<Mutex<HashMap<ThreadId, RolloutRecorder>>>,
metadata_apply_semaphore: Arc<Semaphore>,
state_db: Option<StateDbHandle>,
}
@@ -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<OwnedSemaphorePermit> {
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(),
codex-rs/thread-store/src/local/read_thread.rs
+118 -15
View File
@@ -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<codex_protocol::protocol::GitInfo>,
) -> Option<codex_protocol::protocol::GitInfo> {
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<String>, rollout_value: Option<String>) -> Option<String> {
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");
codex-rs/thread-store/src/store.rs
+15 -1
View File
@@ -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<()>;
codex-rs/thread-store/src/types.rs
+64 -1
View File
@@ -49,6 +49,8 @@ pub struct CreateThreadParams {
pub forked_from_id: Option<ThreadId>,
/// 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<ThreadSource>,
/// Base instructions persisted in session metadata.
@@ -87,6 +89,67 @@ pub struct AppendThreadItemsParams {
pub items: Vec<RolloutItem>,
}
/// 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<PathBuf>,
/// Best available user-facing preview, usually the first user message.
pub preview: Option<String>,
/// Optional user-facing thread name/title.
pub name: Option<Option<String>>,
/// Model provider id associated with the thread.
pub model_provider: Option<String>,
/// Latest observed model, if known.
pub model: Option<Option<String>>,
/// Latest observed reasoning effort, if known.
pub reasoning_effort: Option<Option<ReasoningEffort>>,
/// Thread creation timestamp.
pub created_at: Option<DateTime<Utc>>,
/// Thread last-update timestamp.
pub updated_at: Option<DateTime<Utc>>,
/// Runtime source for the thread.
pub source: Option<SessionSource>,
/// Optional analytics source classification for this thread.
pub thread_source: Option<Option<ThreadSource>>,
/// Optional random nickname for thread-spawn sub-agents.
pub agent_nickname: Option<Option<String>>,
/// Optional role for thread-spawn sub-agents.
pub agent_role: Option<Option<String>>,
/// Optional canonical path for thread-spawn sub-agents.
pub agent_path: Option<Option<String>>,
/// Working directory captured for the thread.
pub cwd: Option<PathBuf>,
/// CLI version captured for the thread.
pub cli_version: Option<String>,
/// Approval mode captured for the thread.
pub approval_mode: Option<AskForApproval>,
/// Sandbox policy captured for the thread.
pub sandbox_policy: Option<SandboxPolicy>,
/// Last observed token usage.
pub token_usage: Option<Option<TokenUsage>>,
/// First user message observed for this thread, if any.
pub first_user_message: Option<Option<String>>,
/// Optional Git metadata captured for the thread.
pub git_info: Option<Option<GitInfo>>,
/// Explicit thread memory behavior.
pub memory_mode: Option<MemoryMode>,
/// Dynamic tools available to the thread.
pub dynamic_tools: Option<Vec<DynamicToolSpec>>,
}
/// 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<RolloutItem>,
/// 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<StoredTurnError>,