openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Move thread metadata into local thread store

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