openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-09 15:58:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

Route app-server thread storage through thread store

Browse Source
This commit is contained in:
Tom Wiltzius
2026-04-10 16:57:50 -07:00
parent bc93ec2b33
commit 4d7b4c3cdf
9 changed files with 408 additions and 763 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
codex-rs/Cargo.lock generated
View File

@@ -1932,6 +1932,7 @@ dependencies = [
"codex-shell-escalation",
"codex-state",
"codex-terminal-detection",
"codex-thread-store",
"codex-tools",
"codex-utils-absolute-path",
"codex-utils-cache",

987
codex-rs/app-server/src/codex_message_processor.rs
View File

File diff suppressed because it is too large Load Diff

1
codex-rs/core/Cargo.toml
View File

@@ -56,6 +56,7 @@ codex-rmcp-client = { workspace = true }
codex-sandboxing = { workspace = true }
codex-state = { workspace = true }
codex-terminal-detection = { workspace = true }
codex-thread-store = { workspace = true }
codex-tools = { workspace = true }
codex-utils-absolute-path = { workspace = true }
codex-utils-cache = { workspace = true }

15
codex-rs/core/src/lib.rs
View File

@@ -153,16 +153,30 @@ pub(crate) mod turn_diff_tracker;
mod turn_metadata;
mod turn_timing;
pub use rollout::ARCHIVED_SESSIONS_SUBDIR;
pub use rollout::ArchiveThreadParams;
pub use rollout::Cursor;
pub use rollout::EventPersistenceMode;
pub use rollout::FindThreadByNameParams;
pub use rollout::GitInfoPatch;
pub use rollout::INTERACTIVE_SESSION_SOURCES;
pub use rollout::ListThreadsParams as StoreListThreadsParams;
pub use rollout::LocalThreadStore;
pub use rollout::ReadThreadParams;
pub use rollout::RolloutRecorder;
pub use rollout::RolloutRecorderParams;
pub use rollout::SESSIONS_SUBDIR;
pub use rollout::SessionMeta;
pub use rollout::SetThreadNameParams;
pub use rollout::StoreThreadSortKey;
pub use rollout::StoredThread;
pub use rollout::ThreadItem;
pub use rollout::ThreadMetadataPatch;
pub use rollout::ThreadOwner;
pub use rollout::ThreadSortKey;
pub use rollout::ThreadStore;
pub use rollout::ThreadStoreError;
pub use rollout::ThreadsPage;
pub use rollout::UpdateThreadMetadataParams;
pub use rollout::append_thread_name;
pub use rollout::find_archived_thread_path_by_id_str;
#[deprecated(note = "use find_thread_path_by_id_str")]
@@ -171,6 +185,7 @@ pub use rollout::find_thread_meta_by_name_str;
pub use rollout::find_thread_name_by_id;
pub use rollout::find_thread_names_by_ids;
pub use rollout::find_thread_path_by_id_str;
pub use rollout::local_thread_store;
pub use rollout::parse_cursor;
pub use rollout::read_head_for_summary;
pub use rollout::read_session_meta_line;

18
codex-rs/core/src/rollout.rs
View File

@@ -22,6 +22,20 @@ pub use codex_rollout::parse_cursor;
pub use codex_rollout::read_head_for_summary;
pub use codex_rollout::read_session_meta_line;
pub use codex_rollout::rollout_date_parts;
pub use codex_thread_store::ArchiveThreadParams;
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::ReadThreadParams;
pub use codex_thread_store::SetThreadNameParams;
pub use codex_thread_store::StoredThread;
pub use codex_thread_store::ThreadMetadataPatch;
pub use codex_thread_store::ThreadOwner;
pub use codex_thread_store::ThreadSortKey as StoreThreadSortKey;
pub use codex_thread_store::ThreadStore;
pub use codex_thread_store::ThreadStoreError;
pub use codex_thread_store::UpdateThreadMetadataParams;
impl codex_rollout::RolloutConfigView for Config {
fn codex_home(&self) -> &std::path::Path {
@@ -45,6 +59,10 @@ impl codex_rollout::RolloutConfigView for Config {
}
}
pub fn local_thread_store(config: &impl codex_rollout::RolloutConfigView) -> LocalThreadStore {
LocalThreadStore::from_config_view(config)
}
pub(crate) mod list {
pub use codex_rollout::ThreadListConfig;
pub use codex_rollout::ThreadListLayout;

2
codex-rs/thread-store/src/local/helpers.rs
View File

@@ -63,6 +63,7 @@ pub(crate) fn metadata_from_items(
pub(crate) fn stored_thread_from_metadata(
metadata: ThreadMetadata,
legacy_path: Option<std::path::PathBuf>,
name: Option<String>,
memory_mode: Option<String>,
history: Option<StoredThreadHistory>,
@@ -72,6 +73,7 @@ pub(crate) fn stored_thread_from_metadata(
StoredThread {
thread_id,
forked_from_id: None,
legacy_path,
owner: Default::default(),
preview: metadata.title.clone(),
name,

137
codex-rs/thread-store/src/local/mod.rs
View File

@@ -1,5 +1,6 @@
use std::path::Path;
use std::path::PathBuf;
use std::time::SystemTime;
use async_trait::async_trait;
use chrono::Utc;
@@ -10,6 +11,7 @@ use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::ThreadNameUpdatedEvent;
use codex_rollout::ARCHIVED_SESSIONS_SUBDIR;
use codex_rollout::RolloutConfig;
use codex_rollout::RolloutConfigView;
use codex_rollout::RolloutRecorder;
use codex_rollout::RolloutRecorderParams;
use codex_rollout::SESSIONS_SUBDIR;
@@ -84,6 +86,11 @@ impl LocalThreadStore {
}
}
/// Create a local store from any rollout configuration view.
pub fn from_config_view(config: &impl RolloutConfigView) -> Self {
Self::new(RolloutConfig::from_view(config))
}
/// Create a local store and initialize the local SQLite state database.
pub async fn with_state_db(config: RolloutConfig) -> Self {
let state_db = codex_rollout::state_db::init(&config).await;
@@ -134,10 +141,10 @@ impl LocalThreadStore {
"local_thread_store_find_path",
)
.await
{
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)),
@@ -275,11 +282,35 @@ impl ThreadStore for LocalThreadStore {
}
async fn read_thread(&self, params: ReadThreadParams) -> ThreadStoreResult<StoredThread> {
let (path, archived) = self
match self
.find_path(params.thread_id, params.include_archived)
.await?;
self.stored_thread_from_path(path.as_path(), archived, params.include_history)
.await
{
Ok((path, archived)) => {
self.stored_thread_from_path(path.as_path(), archived, params.include_history)
.await
}
Err(ThreadStoreError::ThreadNotFound { .. }) if !params.include_history => {
let state_db = self.state_db().await;
let Some(ctx) = state_db.as_deref() else {
return Err(ThreadStoreError::ThreadNotFound {
thread_id: params.thread_id,
});
};
let Some(metadata) = ctx
.get_thread(params.thread_id)
.await
.map_err(display_error)?
else {
return Err(ThreadStoreError::ThreadNotFound {
thread_id: params.thread_id,
});
};
self.stored_thread_from_state_metadata(metadata, false)
.await
}
Err(err) => Err(err),
}
}
async fn list_threads(&self, params: ListThreadsParams) -> ThreadStoreResult<ThreadPage> {
@@ -287,11 +318,7 @@ impl ThreadStore for LocalThreadStore {
let mut last_cursor = cursor.clone();
let requested_page_size = params.page_size.max(1);
let sort_key = rollout_sort_key(params.sort_key);
let allowed_sources = if params.allowed_sources.is_empty() {
codex_rollout::INTERACTIVE_SESSION_SOURCES.clone()
} else {
params.allowed_sources
};
let allowed_sources = params.allowed_sources;
let model_providers = params
.model_providers
.filter(|providers| !providers.is_empty());
@@ -381,21 +408,27 @@ impl ThreadStore for LocalThreadStore {
.iter()
.map(source_to_state_string)
.collect::<Vec<_>>();
let metadata = ctx
.find_thread_by_exact_title(
params.name.as_str(),
allowed_sources.as_slice(),
params.model_providers.as_deref(),
!params.include_archived,
params.cwd.as_deref(),
)
.await
.map_err(display_error)?;
if let Some(metadata) = metadata {
return self
.stored_thread_from_state_metadata(metadata, false)
for archived_only in
[false, true]
.into_iter()
.take(if params.include_archived { 2 } else { 1 })
{
let metadata = ctx
.find_thread_by_exact_title(
params.name.as_str(),
allowed_sources.as_slice(),
params.model_providers.as_deref(),
archived_only,
params.cwd.as_deref(),
)
.await
.map(Some);
.map_err(display_error)?;
if let Some(metadata) = metadata {
return self
.stored_thread_from_state_metadata(metadata, archived_only)
.await
.map(Some);
}
}
}
@@ -485,17 +518,33 @@ impl ThreadStore for LocalThreadStore {
message: "sqlite state db unavailable for git metadata update".to_string(),
});
};
let (path, archived) = self.find_path(params.thread_id, true).await?;
codex_rollout::state_db::reconcile_rollout(
Some(ctx),
path.as_path(),
self.config.model_provider_id.as_str(),
None,
&[],
Some(archived),
None,
)
.await;
match self.find_path(params.thread_id, true).await {
Ok((path, archived)) => {
codex_rollout::state_db::reconcile_rollout(
Some(ctx),
path.as_path(),
self.config.model_provider_id.as_str(),
None,
&[],
Some(archived),
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,
});
}
}
Err(err) => return Err(err),
}
ctx.update_thread_git_info(
params.thread_id,
git_info.sha.as_ref().map(|value| value.as_deref()),
@@ -578,6 +627,20 @@ impl ThreadStore for LocalThreadStore {
tokio::fs::rename(&canonical_path, &restored_path)
.await
.map_err(io_error)?;
tokio::task::spawn_blocking({
let restored_path = restored_path.clone();
move || -> std::io::Result<()> {
let times = std::fs::FileTimes::new().set_modified(SystemTime::now());
std::fs::OpenOptions::new()
.append(true)
.open(&restored_path)?
.set_times(times)?;
Ok(())
}
})
.await
.map_err(display_error)?
.map_err(io_error)?;
if let Some(ctx) = self.state_db().await {
ctx.mark_unarchived(params.thread_id, restored_path.as_path())
.await

4
codex-rs/thread-store/src/local/read.rs
View File

@@ -70,6 +70,7 @@ impl LocalThreadStore {
Ok(stored_thread_from_metadata(
metadata,
Some(path.to_path_buf()),
name,
memory_mode,
include_history.then_some(StoredThreadHistory { thread_id, items }),
@@ -104,7 +105,8 @@ impl LocalThreadStore {
None
};
Ok(stored_thread_from_metadata(
metadata,
metadata.clone(),
Some(metadata.rollout_path.clone()),
name,
memory_mode,
history,

6
codex-rs/thread-store/src/types.rs
View File

@@ -162,7 +162,7 @@ pub struct ListThreadsParams {
pub cursor: Option<String>,
/// Sort order requested by the caller.
pub sort_key: ThreadSortKey,
/// Allowed session sources. Empty means implementation default.
/// Allowed session sources. Empty means all sources.
pub allowed_sources: Vec<SessionSource>,
/// Optional model provider filter. `None` means implementation default, while an empty vector
/// means all providers.
@@ -193,6 +193,8 @@ pub struct StoredThread {
pub thread_id: ThreadId,
/// Source thread id when this thread was forked from another thread.
pub forked_from_id: Option<ThreadId>,
/// Legacy local rollout path, when the backing store has one.
pub legacy_path: Option<PathBuf>,
/// Tenant/owner metadata for multi-tenant stores.
pub owner: ThreadOwner,
/// Best available user-facing preview, usually the first user message.
@@ -306,7 +308,7 @@ pub struct FindThreadByNameParams {
pub include_archived: bool,
/// Optional exact working-directory filter.
pub cwd: Option<PathBuf>,
/// Allowed session sources. Empty means implementation default.
/// Allowed session sources. Empty means all sources.
pub allowed_sources: Vec<SessionSource>,
/// Optional model provider filter. `None` means implementation default, while an empty vector
/// means all providers.