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

use lightweight sqlite rows for thread list

Browse Source
This commit is contained in:
anais
2026-06-21 15:25:47 -07:00
parent 7ddcf58cd9
commit e45d157a55
13 changed files with 610 additions and 178 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/thread_processor.rs
+24 -21
View File
@@ -1942,28 +1942,25 @@ impl ThreadRequestProcessor {
let backwards_cursor = stored_threads.first().and_then(|thread| {
thread_backwards_cursor_for_sort_key(thread, store_sort_key, sort_direction)
});
let mut threads = Vec::with_capacity(stored_threads.len());
let mut status_ids = Vec::with_capacity(stored_threads.len());
let status_ids = stored_threads
.iter()
.map(|thread| thread.thread_id.to_string())
.collect::<Vec<_>>();
let fallback_provider = self.config.model_provider_id.clone();
for stored_thread in stored_threads {
let (thread, _) = thread_from_stored_thread(
stored_thread,
fallback_provider.as_str(),
&self.config.cwd,
);
status_ids.push(thread.id.clone());
threads.push(thread);
}
let statuses = self
.thread_watch_manager
.loaded_statuses_for_threads(status_ids)
.loaded_status_overrides_for_threads(status_ids)
.await;
let data: Vec<_> = threads
let data: Vec<_> = stored_threads
.into_iter()
.map(|mut thread| {
.map(|stored_thread| {
let (mut thread, _) = thread_from_stored_thread(
stored_thread,
fallback_provider.as_str(),
&self.config.cwd,
);
if let Some(status) = statuses.get(&thread.id) {
thread.status = status.clone();
}
@@ -2084,7 +2081,7 @@ impl ThreadRequestProcessor {
}
let statuses = self
.thread_watch_manager
.loaded_statuses_for_threads(status_ids)
.loaded_status_overrides_for_threads(status_ids)
.await;
let data = results
.into_iter()
@@ -3635,6 +3632,11 @@ impl ThreadRequestProcessor {
SortDirection::Asc => StoreSortDirection::Asc,
SortDirection::Desc => StoreSortDirection::Desc,
};
let store_applies_cwd_filters = self
.thread_store
.as_any()
.downcast_ref::<LocalThreadStore>()
.is_some();
while remaining > 0 {
let page_size = remaining.min(THREAD_LIST_MAX_LIMIT);
@@ -3666,11 +3668,12 @@ impl ThreadRequestProcessor {
if source_kind_filter
.as_ref()
.is_none_or(|filter| source_kind_matches(&source, filter))
&& cwd_filters.as_ref().is_none_or(|expected_cwds| {
expected_cwds.iter().any(|expected_cwd| {
path_utils::paths_match_after_normalization(&it.cwd, expected_cwd)
})
})
&& (store_applies_cwd_filters
|| cwd_filters.as_ref().is_none_or(|expected_cwds| {
expected_cwds.iter().any(|expected_cwd| {
path_utils::paths_match_after_normalization(&it.cwd, expected_cwd)
})
}))
{
filtered.push(it);
if filtered.len() >= remaining {
codex-rs/app-server/src/thread_status.rs
+16
View File
@@ -113,6 +113,7 @@ impl ThreadWatchManager {
self.state.lock().await.loaded_status_for_thread(thread_id)
}
#[cfg(test)]
pub(crate) async fn loaded_statuses_for_threads(
&self,
thread_ids: Vec<String>,
@@ -127,6 +128,21 @@ impl ThreadWatchManager {
.collect()
}
pub(crate) async fn loaded_status_overrides_for_threads(
&self,
thread_ids: Vec<String>,
) -> HashMap<String, ThreadStatus> {
let state = self.state.lock().await;
thread_ids
.into_iter()
.filter_map(|thread_id| {
state.status_for(&thread_id).and_then(|status| {
(!matches!(status, ThreadStatus::NotLoaded)).then_some((thread_id, status))
})
})
.collect()
}
#[cfg(test)]
pub(crate) async fn running_turn_count(&self) -> usize {
self.state
codex-rs/app-server/tests/suite/v2/thread_list.rs
+4 -6
View File
@@ -1014,12 +1014,10 @@ async fn thread_list_parent_filter_reads_direct_children_from_state_db() -> Resu
),
] {
let created_at = DateTime::parse_from_rfc3339(created_at)?.with_timezone(&Utc);
let mut builder = codex_state::ThreadMetadataBuilder::new(
thread_id,
codex_home.path().join(format!("{thread_id}.jsonl")),
created_at,
source,
);
let rollout_path = codex_home.path().join(format!("{thread_id}.jsonl"));
std::fs::write(&rollout_path, "")?;
let mut builder =
codex_state::ThreadMetadataBuilder::new(thread_id, rollout_path, created_at, source);
builder.model_provider = Some(model_provider.to_string());
builder.cwd = codex_home.path().to_path_buf();
builder.cli_version = Some("0.0.0".to_string());
codex-rs/rollout/src/list.rs
+134 -47
View File
@@ -52,6 +52,8 @@ pub struct ThreadItem {
pub first_user_message: Option<String>,
/// Best available user-facing preview for discovery and list display.
pub preview: Option<String>,
/// Explicit user-assigned name from persisted metadata, when available.
pub name: Option<String>,
/// Working directory from session metadata.
pub cwd: Option<PathBuf>,
/// Git branch from session metadata.
@@ -94,6 +96,7 @@ struct HeadTailSummary {
thread_id: Option<ThreadId>,
first_user_message: Option<String>,
preview: Option<String>,
name: Option<String>,
cwd: Option<PathBuf>,
git_branch: Option<String>,
git_sha: Option<String>,
@@ -266,30 +269,6 @@ impl<'a> RolloutFileVisitor for FilesByCreatedAtVisitor<'a> {
}
}
/// Collects lightweight file candidates (path + id + mtime).
/// Sorting after mtime happens after all files are collected.
struct FilesByUpdatedAtVisitor<'a> {
candidates: &'a mut Vec<ThreadCandidate>,
}
impl<'a> RolloutFileVisitor for FilesByUpdatedAtVisitor<'a> {
async fn visit(
&mut self,
_ts: OffsetDateTime,
id: Uuid,
path: PathBuf,
_scanned: usize,
) -> ControlFlow<()> {
let updated_at = file_modified_time(&path).await.unwrap_or(None);
self.candidates.push(ThreadCandidate {
path,
id,
updated_at,
});
ControlFlow::Continue(())
}
}
impl serde::Serialize for Cursor {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
@@ -801,6 +780,7 @@ async fn build_thread_item(
thread_id,
first_user_message,
preview,
name,
cwd,
git_branch,
git_sha,
@@ -823,6 +803,7 @@ async fn build_thread_item(
thread_id,
first_user_message,
preview,
name,
cwd,
git_branch,
git_sha,
@@ -981,37 +962,89 @@ async fn collect_files_by_updated_at(
root: &Path,
scanned_files: &mut usize,
) -> io::Result<Vec<ThreadCandidate>> {
let mut candidates = Vec::new();
let mut visitor = FilesByUpdatedAtVisitor {
candidates: &mut candidates,
};
walk_rollout_files(root, scanned_files, &mut visitor).await?;
let root = root.to_path_buf();
let initial_scanned_files = *scanned_files;
let (candidates, scanned) = tokio::task::spawn_blocking(move || {
let mut scanned_files = initial_scanned_files;
let candidates = collect_files_by_updated_at_blocking(root.as_path(), &mut scanned_files)?;
Ok::<_, io::Error>((candidates, scanned_files))
})
.await
.map_err(io::Error::other)??;
*scanned_files = scanned;
Ok(candidates)
}
async fn collect_flat_files_by_updated_at(
root: &Path,
scanned_files: &mut usize,
) -> io::Result<Vec<ThreadCandidate>> {
let root = root.to_path_buf();
let initial_scanned_files = *scanned_files;
let (candidates, scanned) = tokio::task::spawn_blocking(move || {
let mut scanned_files = initial_scanned_files;
let candidates =
collect_flat_files_by_updated_at_blocking(root.as_path(), &mut scanned_files)?;
Ok::<_, io::Error>((candidates, scanned_files))
})
.await
.map_err(io::Error::other)??;
*scanned_files = scanned;
Ok(candidates)
}
fn collect_files_by_updated_at_blocking(
root: &Path,
scanned_files: &mut usize,
) -> io::Result<Vec<ThreadCandidate>> {
let mut candidates = Vec::new();
let mut dir = tokio::fs::read_dir(root).await?;
while let Some(entry) = dir.next_entry().await? {
let year_dirs = collect_dirs_desc_blocking(root, |value| value.parse::<u16>().ok())?;
'outer: for (_year, year_path) in year_dirs.iter() {
if *scanned_files >= MAX_SCAN_FILES {
break;
}
if !entry
.file_type()
.await
.map(|ft| ft.is_file())
.unwrap_or(false)
{
continue;
let month_dirs = collect_dirs_desc_blocking(year_path, |value| value.parse::<u8>().ok())?;
for (_month, month_path) in month_dirs.iter() {
if *scanned_files >= MAX_SCAN_FILES {
break 'outer;
}
let day_dirs =
collect_dirs_desc_blocking(month_path, |value| value.parse::<u8>().ok())?;
for (_day, day_path) in day_dirs.iter() {
if *scanned_files >= MAX_SCAN_FILES {
break 'outer;
}
let day_files = collect_rollout_day_file_candidates_blocking(day_path)?;
for (_timestamp, id, path, updated_at) in day_files {
*scanned_files += 1;
if *scanned_files > MAX_SCAN_FILES {
break 'outer;
}
candidates.push(ThreadCandidate {
path,
id,
updated_at,
});
}
}
}
let Some(rollout_file) = compression::RolloutFile::from_path(entry.path()) else {
continue;
};
let Some((_ts, id)) = parse_timestamp_uuid_from_filename(rollout_file.plain_file_name())
}
Ok(candidates)
}
fn collect_flat_files_by_updated_at_blocking(
root: &Path,
scanned_files: &mut usize,
) -> io::Result<Vec<ThreadCandidate>> {
let mut candidates = Vec::new();
for entry in std::fs::read_dir(root)? {
if *scanned_files >= MAX_SCAN_FILES {
break;
}
let entry = entry?;
let Some((_timestamp, id, path, updated_at)) = rollout_candidate_from_dir_entry(&entry)
else {
continue;
};
@@ -1019,11 +1052,8 @@ async fn collect_flat_files_by_updated_at(
if *scanned_files > MAX_SCAN_FILES {
break;
}
let updated_at = file_modified_time(rollout_file.path())
.await
.unwrap_or(None);
candidates.push(ThreadCandidate {
path: rollout_file.into_path(),
path,
id,
updated_at,
});
@@ -1032,6 +1062,63 @@ async fn collect_flat_files_by_updated_at(
Ok(candidates)
}
fn collect_dirs_desc_blocking<T, F>(parent: &Path, parse: F) -> io::Result<Vec<(T, PathBuf)>>
where
T: Ord + Copy,
F: Fn(&str) -> Option<T>,
{
let mut directories = Vec::new();
for entry in std::fs::read_dir(parent)? {
let entry = entry?;
if entry
.file_type()
.map(|file_type| file_type.is_dir())
.unwrap_or(false)
&& let Some(name) = entry.file_name().to_str()
&& let Some(value) = parse(name)
{
directories.push((value, entry.path()));
}
}
directories.sort_by_key(|(value, _)| Reverse(*value));
Ok(directories)
}
fn collect_rollout_day_file_candidates_blocking(
day_path: &Path,
) -> io::Result<Vec<(OffsetDateTime, Uuid, PathBuf, Option<OffsetDateTime>)>> {
let mut day_files = Vec::new();
for entry in std::fs::read_dir(day_path)? {
let entry = entry?;
if let Some(candidate) = rollout_candidate_from_dir_entry(&entry) {
day_files.push(candidate);
}
}
day_files.sort_by_key(|(timestamp, id, _, _)| (Reverse(*timestamp), Reverse(*id)));
Ok(day_files)
}
fn rollout_candidate_from_dir_entry(
entry: &std::fs::DirEntry,
) -> Option<(OffsetDateTime, Uuid, PathBuf, Option<OffsetDateTime>)> {
if !entry
.file_type()
.map(|file_type| file_type.is_file())
.unwrap_or(false)
{
return None;
}
let rollout_file = compression::RolloutFile::from_path(entry.path())?;
let (timestamp, id) = parse_timestamp_uuid_from_filename(rollout_file.plain_file_name())?;
let updated_at = entry
.metadata()
.ok()
.and_then(|metadata| metadata.modified().ok())
.map(OffsetDateTime::from)
.and_then(truncate_to_millis);
Some((timestamp, id, rollout_file.into_path(), updated_at))
}
async fn walk_rollout_files(
root: &Path,
scanned_files: &mut usize,
codex-rs/rollout/src/recorder.rs
+95 -20
View File
@@ -452,8 +452,8 @@ impl RolloutRecorder {
// Warm the DB by repairing every filesystem hit before querying SQLite. Source/provider/cwd
// filters are already validated from rollout head metadata, so lightweight read-repair is
// enough there. Search can depend on full title metadata, so keep full reconciliation.
for item in &fs_page.items {
if search_term.is_some() {
if search_term.is_some() {
for item in &fs_page.items {
state_db::reconcile_rollout(
state_db_ctx.as_deref(),
item.path.as_path(),
@@ -464,15 +464,19 @@ impl RolloutRecorder {
/*new_thread_memory_mode*/ None,
)
.await;
} else {
state_db::read_repair_rollout_path(
state_db_ctx.as_deref(),
item.thread_id,
Some(archived),
item.path.as_path(),
)
.await;
}
} else {
let repair_paths = fs_page
.items
.iter()
.map(|item| (item.thread_id, item.path.clone()))
.collect::<Vec<_>>();
state_db::read_repair_rollout_paths(
state_db_ctx.as_deref(),
repair_paths.as_slice(),
Some(archived),
)
.await;
}
let db_page = state_db::list_threads_db(
@@ -1037,19 +1041,43 @@ async fn fill_missing_thread_item_metadata_from_state_db(
return page;
};
let thread_ids = page
.items
.iter()
.filter_map(|item| item.thread_id)
.collect::<Vec<_>>();
let mut metadata_by_id = match state_db_ctx.get_threads(thread_ids.as_slice()).await {
Ok(metadata_by_id) => metadata_by_id,
Err(err) => {
warn!(
"state db get_threads failed while overlaying filesystem scan thread metadata: {err}"
);
for item in &mut page.items {
let Some(thread_id) = item.thread_id else {
continue;
};
let metadata = match state_db_ctx.get_thread(thread_id).await {
Ok(Some(metadata)) => metadata,
Ok(None) => continue,
Err(err) => {
warn!(
"state db get_thread failed while overlaying filesystem scan thread metadata: {err}"
);
continue;
}
};
fill_missing_thread_item_metadata(item, thread_item_from_state_metadata(metadata));
}
return page;
}
};
for item in &mut page.items {
let Some(thread_id) = item.thread_id else {
continue;
};
let metadata = match state_db_ctx.get_thread(thread_id).await {
Ok(Some(metadata)) => metadata,
Ok(None) => continue,
Err(err) => {
warn!(
"state db get_thread failed while overlaying filesystem scan thread metadata: {err}"
);
continue;
}
let Some(metadata) = metadata_by_id.remove(&thread_id) else {
continue;
};
fill_missing_thread_item_metadata(item, thread_item_from_state_metadata(metadata));
}
@@ -1063,6 +1091,7 @@ fn fill_missing_thread_item_metadata(item: &mut ThreadItem, state_item: ThreadIt
thread_id: _state_thread_id,
first_user_message,
preview,
name,
cwd,
git_branch,
git_sha,
@@ -1084,6 +1113,9 @@ fn fill_missing_thread_item_metadata(item: &mut ThreadItem, state_item: ThreadIt
if item.preview.is_none() {
item.preview = preview;
}
if item.name.is_none() {
item.name = name;
}
if item.cwd.is_none() {
item.cwd = cwd;
}
@@ -1757,12 +1789,55 @@ impl From<codex_state::ThreadsPage> for ThreadsPage {
}
}
impl From<codex_state::ThreadListPage> for ThreadsPage {
fn from(db_page: codex_state::ThreadListPage) -> Self {
let items = db_page
.items
.into_iter()
.map(thread_item_from_state_list_item)
.collect();
Self {
items,
next_cursor: db_page.next_anchor.map(Into::into),
num_scanned_files: db_page.num_scanned_rows,
reached_scan_cap: false,
}
}
}
fn thread_item_from_state_list_item(item: codex_state::ThreadListItem) -> ThreadItem {
let source = serde_json::from_str(item.source.as_str())
.or_else(|_| serde_json::from_value(Value::String(item.source)))
.unwrap_or(SessionSource::Unknown);
ThreadItem {
path: item.rollout_path,
thread_id: Some(item.id),
first_user_message: item.first_user_message,
preview: item.preview,
name: item.name,
cwd: Some(item.cwd),
git_branch: item.git_branch,
git_sha: item.git_sha,
git_origin_url: item.git_origin_url,
source: Some(source),
parent_thread_id: item.parent_thread_id,
agent_nickname: item.agent_nickname,
agent_role: item.agent_role,
model_provider: Some(item.model_provider),
cli_version: Some(item.cli_version),
created_at: Some(item.created_at.to_rfc3339_opts(SecondsFormat::Secs, true)),
updated_at: Some(item.updated_at.to_rfc3339_opts(SecondsFormat::Millis, true)),
recency_at: Some(item.recency_at.to_rfc3339_opts(SecondsFormat::Millis, true)),
}
}
fn thread_item_from_state_metadata(item: codex_state::ThreadMetadata) -> ThreadItem {
ThreadItem {
path: item.rollout_path,
thread_id: Some(item.id),
first_user_message: item.first_user_message,
preview: item.preview,
name: item.name,
cwd: Some(item.cwd),
git_branch: item.git_branch,
git_sha: item.git_sha,
@@ -1837,7 +1912,7 @@ async fn resume_candidate_matches_cwd(
}
async fn select_resume_path_from_db_page(
page: &codex_state::ThreadsPage,
page: &codex_state::ThreadListPage,
filter_cwd: Option<&Path>,
default_provider: &str,
) -> Option<PathBuf> {
codex-rs/rollout/src/recorder_tests.rs
+2
View File
@@ -973,6 +973,7 @@ fn fill_missing_thread_item_metadata_preserves_identity_and_prefers_state_git_fi
thread_id: Some(filesystem_thread_id),
first_user_message: Some("filesystem message".to_string()),
preview: Some("filesystem preview".to_string()),
name: None,
cwd: None,
git_branch: Some("filesystem-branch".to_string()),
git_sha: Some("filesystem-sha".to_string()),
@@ -992,6 +993,7 @@ fn fill_missing_thread_item_metadata_preserves_identity_and_prefers_state_git_fi
thread_id: Some(state_thread_id),
first_user_message: Some("state message".to_string()),
preview: Some("state preview".to_string()),
name: None,
cwd: Some(PathBuf::from("/tmp/state-cwd")),
git_branch: Some("state-branch".to_string()),
git_sha: Some("state-sha".to_string()),
codex-rs/rollout/src/session_index.rs
+22 -21
View File
@@ -15,7 +15,6 @@ use codex_protocol::ThreadId;
use codex_protocol::protocol::SessionMetaLine;
use serde::Deserialize;
use serde::Serialize;
use tokio::io::AsyncBufReadExt;
const SESSION_INDEX_FILE: &str = "session_index.jsonl";
const READ_CHUNK_SIZE: usize = 8192;
@@ -130,26 +129,10 @@ pub async fn find_thread_names_by_ids(
return Ok(HashMap::new());
}
let file = tokio::fs::File::open(&path).await?;
let reader = tokio::io::BufReader::new(file);
let mut lines = reader.lines();
let mut names = HashMap::with_capacity(thread_ids.len());
while let Some(line) = lines.next_line().await? {
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
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());
}
}
Ok(names)
let thread_ids = thread_ids.clone();
tokio::task::spawn_blocking(move || scan_index_from_end_by_ids(&path, &thread_ids))
.await
.map_err(std::io::Error::other)?
}
/// Locate a recorded thread rollout and read its session metadata by thread name.
@@ -205,6 +188,24 @@ fn scan_index_from_end_by_id(
scan_index_from_end(path, |entry| entry.id == *thread_id)
}
fn scan_index_from_end_by_ids(
path: &Path,
thread_ids: &HashSet<ThreadId>,
) -> std::io::Result<HashMap<ThreadId, String>> {
let mut names = HashMap::with_capacity(thread_ids.len());
scan_index_from_end_for_each(path, |entry| {
let name = entry.thread_name.trim();
if !name.is_empty() && thread_ids.contains(&entry.id) && !names.contains_key(&entry.id) {
names.insert(entry.id, name.to_string());
if names.len() == thread_ids.len() {
return Ok(Some(entry.clone()));
}
}
Ok(None)
})?;
Ok(names)
}
fn stream_thread_ids_from_end_by_name(
path: &Path,
name: &str,
codex-rs/rollout/src/state_db.rs
+89 -26
View File
@@ -12,6 +12,7 @@ use codex_protocol::ThreadId;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
pub use codex_state::LogEntry;
use codex_state::ThreadMetadata;
use codex_state::ThreadMetadataBuilder;
use codex_utils_path::normalize_for_path_comparison;
use serde_json::Value;
@@ -355,7 +356,7 @@ pub async fn list_thread_ids_db(
}
}
/// List thread metadata from SQLite without rollout directory traversal.
/// List thread summaries from SQLite without rollout directory traversal.
#[allow(clippy::too_many_arguments)]
pub async fn list_threads_db(
context: Option<&codex_state::StateRuntime>,
@@ -370,7 +371,7 @@ pub async fn list_threads_db(
parent_thread_id: Option<ThreadId>,
archived: bool,
search_term: Option<&str>,
) -> Option<codex_state::ThreadsPage> {
) -> Option<codex_state::ThreadListPage> {
let ctx = context?;
if ctx.codex_home() != codex_home {
warn!(
@@ -415,17 +416,13 @@ pub async fn list_threads_db(
};
let page = match parent_thread_id {
Some(parent_thread_id) => {
ctx.list_threads_by_parent(page_size, parent_thread_id, filters)
ctx.list_thread_summaries_by_parent(page_size, parent_thread_id, filters)
.await
}
None => ctx.list_threads(page_size, filters).await,
None => ctx.list_thread_summaries(page_size, filters).await,
};
match page {
Ok(mut page) => {
// Parent-filtered listings intentionally treat persisted state as authoritative.
if parent_thread_id.is_some() {
return Some(page);
}
let mut valid_items = Vec::with_capacity(page.items.len());
for item in page.items {
if let Some(existing_path) =
@@ -558,6 +555,52 @@ pub async fn reconcile_rollout(
}
}
/// Repair a batch of rollout locations after filesystem fallback succeeds.
pub async fn read_repair_rollout_paths(
context: Option<&codex_state::StateRuntime>,
rollout_paths: &[(Option<ThreadId>, PathBuf)],
archived_only: Option<bool>,
) {
let Some(ctx) = context else {
return;
};
if rollout_paths.is_empty() {
return;
}
let thread_ids = rollout_paths
.iter()
.filter_map(|(thread_id, _)| *thread_id)
.collect::<Vec<_>>();
let metadata_by_id = match ctx.get_threads(thread_ids.as_slice()).await {
Ok(metadata_by_id) => metadata_by_id,
Err(err) => {
warn!("state db batch read-repair lookup failed: {err}");
for (thread_id, rollout_path) in rollout_paths {
read_repair_rollout_path(Some(ctx), *thread_id, archived_only, rollout_path).await;
}
return;
}
};
for (thread_id, rollout_path) in rollout_paths {
if let Some(thread_id) = thread_id
&& let Some(metadata) = metadata_by_id.get(thread_id)
&& repair_existing_rollout_location(
ctx,
metadata,
archived_only,
rollout_path.as_path(),
)
.await
{
continue;
}
warn!("state db discrepancy during read_repair_rollout_path: upsert_needed (slow path)");
reconcile_rollout_path_from_contents(ctx, rollout_path.as_path(), archived_only).await;
}
}
/// Repair a thread's rollout path after filesystem fallback succeeds.
pub async fn read_repair_rollout_path(
context: Option<&codex_state::StateRuntime>,
@@ -576,24 +619,8 @@ pub async fn read_repair_rollout_path(
&& let Ok(Some(metadata)) = ctx.get_thread(thread_id).await
{
saw_existing_metadata = true;
let cwd = normalize_cwd_for_state_db(&metadata.cwd);
let archive_state = match archived_only {
Some(true) => codex_state::ThreadArchiveState::Archived,
Some(false) => codex_state::ThreadArchiveState::Active,
None => codex_state::ThreadArchiveState::Preserve,
};
match ctx
.repair_thread_rollout_location(thread_id, rollout_path, cwd.as_path(), archive_state)
.await
{
Ok(true) => return,
Ok(false) => saw_existing_metadata = false,
Err(err) => {
warn!(
"state db read-repair update failed for {}: {err}",
rollout_path.display()
);
}
if repair_existing_rollout_location(ctx, &metadata, archived_only, rollout_path).await {
return;
}
}
@@ -602,6 +629,42 @@ pub async fn read_repair_rollout_path(
if !saw_existing_metadata {
warn!("state db discrepancy during read_repair_rollout_path: upsert_needed (slow path)");
}
reconcile_rollout_path_from_contents(ctx, rollout_path, archived_only).await;
}
async fn repair_existing_rollout_location(
ctx: &codex_state::StateRuntime,
metadata: &ThreadMetadata,
archived_only: Option<bool>,
rollout_path: &Path,
) -> bool {
let cwd = normalize_cwd_for_state_db(&metadata.cwd);
let archive_state = match archived_only {
Some(true) => codex_state::ThreadArchiveState::Archived,
Some(false) => codex_state::ThreadArchiveState::Active,
None => codex_state::ThreadArchiveState::Preserve,
};
match ctx
.repair_thread_rollout_location(metadata.id, rollout_path, cwd.as_path(), archive_state)
.await
{
Ok(true) => true,
Ok(false) => false,
Err(err) => {
warn!(
"state db read-repair update failed for {}: {err}",
rollout_path.display()
);
false
}
}
}
async fn reconcile_rollout_path_from_contents(
ctx: &codex_state::StateRuntime,
rollout_path: &Path,
archived_only: Option<bool>,
) {
let default_provider = crate::list::read_session_meta_line(rollout_path)
.await
.ok()
codex-rs/rollout/src/tests.rs
+11
View File
@@ -601,6 +601,7 @@ async fn test_list_conversations_latest_first() {
thread_id: Some(thread_id_from_uuid(u3)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
@@ -620,6 +621,7 @@ async fn test_list_conversations_latest_first() {
thread_id: Some(thread_id_from_uuid(u2)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
@@ -639,6 +641,7 @@ async fn test_list_conversations_latest_first() {
thread_id: Some(thread_id_from_uuid(u1)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
@@ -751,6 +754,7 @@ async fn test_pagination_cursor() {
thread_id: Some(thread_id_from_uuid(u5)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
@@ -770,6 +774,7 @@ async fn test_pagination_cursor() {
thread_id: Some(thread_id_from_uuid(u4)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
@@ -825,6 +830,7 @@ async fn test_pagination_cursor() {
thread_id: Some(thread_id_from_uuid(u3)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
@@ -844,6 +850,7 @@ async fn test_pagination_cursor() {
thread_id: Some(thread_id_from_uuid(u2)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
@@ -891,6 +898,7 @@ async fn test_pagination_cursor() {
thread_id: Some(thread_id_from_uuid(u1)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
@@ -1063,6 +1071,7 @@ async fn test_get_thread_contents() {
thread_id: Some(thread_id_from_uuid(uuid)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
@@ -1419,6 +1428,7 @@ async fn test_timestamp_only_cursor_skips_same_second_filesystem_ties() {
thread_id: Some(thread_id_from_uuid(u3)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
@@ -1438,6 +1448,7 @@ async fn test_timestamp_only_cursor_skips_same_second_filesystem_ties() {
thread_id: Some(thread_id_from_uuid(u2)),
first_user_message: Some("Hello from user".to_string()),
preview: Some("Hello from user".to_string()),
name: None,
cwd: Some(Path::new(".").to_path_buf()),
git_branch: None,
git_sha: None,
codex-rs/state/migrations/0041_thread_spawn_edges_parent_child.sql
+2
View File
@@ -0,0 +1,2 @@
CREATE INDEX idx_thread_spawn_edges_parent_child
ON thread_spawn_edges(parent_thread_id, child_thread_id);
codex-rs/state/src/runtime/threads.rs
+30
View File
@@ -46,6 +46,36 @@ WHERE threads.id = ?
.transpose()
}
pub async fn get_threads(
&self,
ids: &[ThreadId],
) -> anyhow::Result<std::collections::HashMap<ThreadId, crate::ThreadMetadata>> {
if ids.is_empty() {
return Ok(std::collections::HashMap::new());
}
const MAX_SQL_BINDINGS: usize = 900;
let mut metadata_by_id = std::collections::HashMap::with_capacity(ids.len());
for chunk in ids.chunks(MAX_SQL_BINDINGS) {
let mut builder = QueryBuilder::<Sqlite>::new("");
push_thread_select_columns(&mut builder);
builder.push(" FROM threads WHERE threads.id IN (");
let mut separated = builder.separated(", ");
for id in chunk {
separated.push_bind(id.to_string());
}
separated.push_unseparated(")");
let rows = builder.build().fetch_all(self.pool.as_ref()).await?;
for row in rows {
let metadata = ThreadRow::try_from_row(&row).and_then(ThreadMetadata::try_from)?;
metadata_by_id.insert(metadata.id, metadata);
}
}
Ok(metadata_by_id)
}
pub async fn get_thread_memory_mode(&self, id: ThreadId) -> anyhow::Result<Option<String>> {
let row = sqlx::query("SELECT memory_mode FROM threads WHERE id = ?")
.bind(id.to_string())
codex-rs/thread-store/src/local/helpers.rs
+88 -1
View File
@@ -6,6 +6,7 @@ use std::path::PathBuf;
use std::time::SystemTime;
use chrono::DateTime;
use chrono::Timelike;
use chrono::Utc;
use codex_git_utils::GitSha;
use codex_protocol::ThreadId;
@@ -17,6 +18,7 @@ use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionSource;
use codex_rollout::ARCHIVED_SESSIONS_SUBDIR;
use codex_rollout::ThreadItem;
use codex_state::ThreadListItem;
use codex_state::ThreadMetadata;
use crate::StoredThread;
@@ -119,6 +121,7 @@ pub(super) fn stored_thread_from_rollout_item(
.clone()
.or_else(|| item.first_user_message.clone())
.unwrap_or_default();
let name = item.name.clone();
let rollout_path = codex_rollout::plain_rollout_path(item.path.as_path());
Some(StoredThread {
@@ -128,7 +131,7 @@ pub(super) fn stored_thread_from_rollout_item(
forked_from_id: None,
parent_thread_id: item.parent_thread_id,
preview,
name: None,
name,
model_provider: item
.model_provider
.filter(|provider| !provider.is_empty())
@@ -155,6 +158,75 @@ pub(super) fn stored_thread_from_rollout_item(
})
}
pub(super) fn stored_thread_from_state_list_item(
item: ThreadListItem,
archived: bool,
default_provider: &str,
) -> StoredThread {
let ThreadListItem {
id,
rollout_path,
created_at,
updated_at,
recency_at,
source,
parent_thread_id,
agent_nickname,
agent_role,
model_provider,
cwd,
cli_version,
title: _,
name,
preview,
first_user_message,
git_sha,
git_branch,
git_origin_url,
} = item;
let preview = preview
.or_else(|| first_user_message.clone())
.unwrap_or_default();
let git_info = git_info_from_parts(git_sha, git_branch, git_origin_url);
let rollout_path = codex_rollout::plain_rollout_path(rollout_path.as_path());
let model_provider = if model_provider.is_empty() {
default_provider.to_string()
} else {
model_provider
};
StoredThread {
thread_id: id,
extra_config: None,
rollout_path: Some(rollout_path),
forked_from_id: None,
parent_thread_id,
preview,
name,
model_provider,
model: None,
reasoning_effort: None,
created_at: truncate_datetime_to_seconds(created_at),
updated_at: truncate_datetime_to_millis(updated_at),
recency_at: truncate_datetime_to_millis(recency_at),
archived_at: archived.then_some(updated_at),
cwd,
cli_version,
source: parse_session_source(&source),
thread_source: None,
agent_nickname,
agent_role,
agent_path: None,
git_info,
approval_mode: AskForApproval::OnRequest,
permission_profile: PermissionProfile::read_only(),
token_usage: None,
first_user_message,
history: None,
}
}
pub(super) fn permission_profile_from_metadata_value(value: &str, cwd: &Path) -> PermissionProfile {
serde_json::from_str::<PermissionProfile>(value)
.or_else(|_| {
@@ -219,6 +291,21 @@ pub(super) fn set_thread_name_from_title(thread: &mut StoredThread, title: Strin
thread.name = Some(title);
}
pub(super) fn parse_session_source(source: &str) -> SessionSource {
serde_json::from_str(source)
.or_else(|_| serde_json::from_value(serde_json::Value::String(source.to_string())))
.unwrap_or(SessionSource::Unknown)
}
fn truncate_datetime_to_seconds(timestamp: DateTime<Utc>) -> DateTime<Utc> {
DateTime::<Utc>::from_timestamp(timestamp.timestamp(), 0).unwrap_or(timestamp)
}
fn truncate_datetime_to_millis(timestamp: DateTime<Utc>) -> DateTime<Utc> {
let nanos = timestamp.nanosecond() / 1_000_000 * 1_000_000;
DateTime::<Utc>::from_timestamp(timestamp.timestamp(), nanos).unwrap_or(timestamp)
}
fn parse_rfc3339(value: Option<&str>) -> Option<DateTime<Utc>> {
DateTime::parse_from_rfc3339(value?)
.ok()
codex-rs/thread-store/src/local/list_threads.rs
+93 -36
View File
@@ -1,19 +1,17 @@
use std::collections::HashMap;
use std::collections::HashSet;
use codex_protocol::ThreadId;
use codex_rollout::RolloutConfig;
use codex_rollout::RolloutRecorder;
use codex_rollout::find_thread_names_by_ids;
use codex_rollout::parse_cursor;
use super::LocalThreadStore;
use super::helpers::ThreadMetadataName;
use super::helpers::apply_thread_metadata_name;
use super::helpers::set_thread_name_from_title;
use super::helpers::stored_thread_from_rollout_item;
use super::helpers::thread_metadata_name;
use super::helpers::stored_thread_from_state_list_item;
use crate::ListThreadsParams;
use crate::SortDirection;
use crate::StoredThread;
use crate::ThreadPage;
use crate::ThreadSortKey;
use crate::ThreadStoreError;
@@ -42,6 +40,18 @@ pub(super) async fn list_threads(
SortDirection::Desc => codex_rollout::SortDirection::Desc,
};
let state_db = store.state_db().await;
if params.parent_thread_id.is_some() || params.use_state_db_only {
return list_threads_from_state_db(
store,
state_db.as_deref(),
&params,
cursor.as_ref(),
sort_key,
sort_direction,
)
.await;
}
let rollout_config = RolloutConfig {
codex_home: store.config.codex_home.clone(),
sqlite_home: store.config.sqlite_home.clone(),
@@ -77,38 +87,89 @@ pub(super) async fn list_threads(
})
.collect::<Vec<_>>();
let thread_ids = items
fill_thread_names(store, &mut items).await;
Ok(ThreadPage { items, next_cursor })
}
async fn list_threads_from_state_db(
store: &LocalThreadStore,
state_db: Option<&codex_state::StateRuntime>,
params: &ListThreadsParams,
cursor: Option<&codex_rollout::Cursor>,
sort_key: codex_rollout::ThreadSortKey,
sort_direction: codex_rollout::SortDirection,
) -> ThreadStoreResult<ThreadPage> {
let page = codex_rollout::state_db::list_threads_db(
state_db,
store.config.codex_home.as_path(),
params.page_size,
cursor,
sort_key,
sort_direction,
params.allowed_sources.as_slice(),
params.model_providers.as_deref(),
params.cwd_filters.as_deref(),
params.parent_thread_id,
params.archived,
params.search_term.as_deref(),
)
.await;
let Some(page) = page else {
if params.parent_thread_id.is_some() {
return Err(ThreadStoreError::Internal {
message: "state DB unavailable for parent-filtered thread listing".to_string(),
});
}
return Ok(ThreadPage {
items: Vec::new(),
next_cursor: None,
});
};
let codex_state::ThreadListPage {
items, next_anchor, ..
} = page;
let next_cursor = next_anchor
.map(codex_rollout::Cursor::from)
.as_ref()
.and_then(|cursor| serde_json::to_value(cursor).ok())
.and_then(|value| value.as_str().map(str::to_owned));
let items = items
.into_iter()
.map(|item| {
stored_thread_from_state_list_item(
item,
params.archived,
store.config.default_model_provider_id.as_str(),
)
})
.collect::<Vec<_>>();
Ok(ThreadPage { items, next_cursor })
}
async fn fill_thread_names(store: &LocalThreadStore, items: &mut [StoredThread]) {
let thread_ids_missing_names = items
.iter()
.filter(|thread| thread.name.is_none())
.map(|thread| thread.thread_id)
.collect::<HashSet<_>>();
let mut names = HashMap::<ThreadId, ThreadMetadataName>::with_capacity(thread_ids.len());
if let Some(state_db_ctx) = store.state_db().await {
for &thread_id in &thread_ids {
let Ok(Some(metadata)) = state_db_ctx.get_thread(thread_id).await else {
continue;
};
if let Some(name) = thread_metadata_name(&metadata) {
names.insert(thread_id, name);
if thread_ids_missing_names.is_empty() {
return;
}
if let Ok(legacy_names) =
find_thread_names_by_ids(store.config.codex_home.as_path(), &thread_ids_missing_names).await
{
for thread in items {
if thread.name.is_none()
&& let Some(title) = legacy_names.get(&thread.thread_id).cloned()
{
set_thread_name_from_title(thread, title);
}
}
}
if names.len() < thread_ids.len()
&& let Ok(legacy_names) =
find_thread_names_by_ids(store.config.codex_home.as_path(), &thread_ids).await
{
for (thread_id, title) in legacy_names {
names
.entry(thread_id)
.or_insert(ThreadMetadataName::Legacy(title));
}
}
for thread in &mut items {
if let Some(name) = names.get(&thread.thread_id).cloned() {
apply_thread_metadata_name(thread, name);
}
}
Ok(ThreadPage { items, next_cursor })
}
pub(super) async fn list_rollout_threads(
@@ -139,11 +200,7 @@ pub(super) async fn list_rollout_threads(
.ok_or_else(|| ThreadStoreError::Internal {
message: "state DB unavailable for parent-filtered thread listing".to_string(),
})?;
let mut page: codex_rollout::ThreadsPage = page.into();
for item in &mut page.items {
item.parent_thread_id = Some(parent_thread_id);
}
return Ok(page);
return Ok(page.into());
}
let page = if params.use_state_db_only && params.archived {