github/codex
4
0 Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-13 14:46:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Support multiple cwd filters for thread list (#18502)

Browse Source 
## Summary

- Teach app-server `thread/list` to accept either a single `cwd` or an
array of cwd filters, returning threads whose recorded session cwd
matches any requested path
- Add `useStateDbOnly` as an explicit opt-in fast path for callers that
want to answer `thread/list` from SQLite without scanning JSONL rollout
files
- Preserve backwards compatibility: by default, `thread/list` still
scans JSONL rollouts and repairs SQLite state
- Wire the new cwd array and SQLite-only options through app-server,
local/remote thread-store, rollout listing, generated TypeScript/schema
fixtures, proto output, and docs

## Test Plan

- `cargo test -p codex-app-server-protocol`
- `cargo test -p codex-rollout`
- `cargo test -p codex-thread-store`
- `cargo test -p codex-app-server thread_list`
- `just fmt`
- `just fix -p codex-app-server-protocol -p codex-rollout -p
codex-thread-store -p codex-app-server`
- `cargo build -p codex-cli --bin codex`
This commit is contained in:
acrognale-oai authored and GitHub committed 2026-04-22 06:10:09 -04:00
1 parent b04ffeee4c
commit 4f8c58f737
32 files changed
+1183 -133
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

codex-rs/rollout/src/list.rs
+58 -10
View File
@@ -1,6 +1,7 @@
#![allow(warnings, clippy::all)]
use async_trait::async_trait;
use codex_utils_path as path_utils;
use std::cmp::Reverse;
use std::ffi::OsStr;
use std::io;
@@ -126,6 +127,7 @@ pub enum ThreadListLayout {
pub struct ThreadListConfig<'a> {
pub allowed_sources: &'a [SessionSource],
pub model_providers: Option<&'a [String]>,
pub cwd_filters: Option<&'a [PathBuf]>,
pub default_provider: &'a str,
pub layout: ThreadListLayout,
}
@@ -207,6 +209,7 @@ struct FilesByCreatedAtVisitor<'a> {
more_matches_available: bool,
allowed_sources: &'a [SessionSource],
provider_matcher: Option<&'a ProviderMatcher<'a>>,
cwd_filters: Option<&'a [PathBuf]>,
}
#[async_trait]
@@ -237,6 +240,7 @@ impl<'a> RolloutFileVisitor for FilesByCreatedAtVisitor<'a> {
path,
self.allowed_sources,
self.provider_matcher,
self.cwd_filters,
updated_at,
)
.await
@@ -317,6 +321,7 @@ pub async fn get_threads(
sort_key: ThreadSortKey,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
cwd_filters: Option<&[PathBuf]>,
default_provider: &str,
) -> io::Result<ThreadsPage> {
let root = codex_home.join(SESSIONS_SUBDIR);
@@ -328,6 +333,7 @@ pub async fn get_threads(
ThreadListConfig {
allowed_sources,
model_providers,
cwd_filters,
default_provider,
layout: ThreadListLayout::NestedByDate,
},
@@ -366,6 +372,7 @@ pub async fn get_threads_in_root(
sort_key,
config.allowed_sources,
provider_matcher.as_ref(),
config.cwd_filters,
)
.await?
}
@@ -377,6 +384,7 @@ pub async fn get_threads_in_root(
sort_key,
config.allowed_sources,
provider_matcher.as_ref(),
config.cwd_filters,
)
.await?
}
@@ -395,6 +403,7 @@ async fn traverse_directories_for_paths(
sort_key: ThreadSortKey,
allowed_sources: &[SessionSource],
provider_matcher: Option<&ProviderMatcher<'_>>,
cwd_filters: Option<&[PathBuf]>,
) -> io::Result<ThreadsPage> {
match sort_key {
ThreadSortKey::CreatedAt => {
@@ -404,6 +413,7 @@ async fn traverse_directories_for_paths(
anchor,
allowed_sources,
provider_matcher,
cwd_filters,
)
.await
}
@@ -414,6 +424,7 @@ async fn traverse_directories_for_paths(
anchor,
allowed_sources,
provider_matcher,
cwd_filters,
)
.await
}
@@ -427,15 +438,30 @@ async fn traverse_flat_paths(
sort_key: ThreadSortKey,
allowed_sources: &[SessionSource],
provider_matcher: Option<&ProviderMatcher<'_>>,
cwd_filters: Option<&[PathBuf]>,
) -> io::Result<ThreadsPage> {
match sort_key {
ThreadSortKey::CreatedAt => {
traverse_flat_paths_created(root, page_size, anchor, allowed_sources, provider_matcher)
.await
traverse_flat_paths_created(
root,
page_size,
anchor,
allowed_sources,
provider_matcher,
cwd_filters,
)
.await
}
ThreadSortKey::UpdatedAt => {
traverse_flat_paths_updated(root, page_size, anchor, allowed_sources, provider_matcher)
.await
traverse_flat_paths_updated(
root,
page_size,
anchor,
allowed_sources,
provider_matcher,
cwd_filters,
)
.await
}
}
}
@@ -452,6 +478,7 @@ async fn traverse_directories_for_paths_created(
anchor: Option<Cursor>,
allowed_sources: &[SessionSource],
provider_matcher: Option<&ProviderMatcher<'_>>,
cwd_filters: Option<&[PathBuf]>,
) -> io::Result<ThreadsPage> {
let mut items: Vec<ThreadItem> = Vec::with_capacity(page_size);
let mut scanned_files = 0usize;
@@ -463,6 +490,7 @@ async fn traverse_directories_for_paths_created(
more_matches_available,
allowed_sources,
provider_matcher,
cwd_filters,
};
walk_rollout_files(&root, &mut scanned_files, &mut visitor).await?;
more_matches_available = visitor.more_matches_available;
@@ -499,14 +527,14 @@ async fn traverse_directories_for_paths_updated(
anchor: Option<Cursor>,
allowed_sources: &[SessionSource],
provider_matcher: Option<&ProviderMatcher<'_>>,
cwd_filters: Option<&[PathBuf]>,
) -> io::Result<ThreadsPage> {
let mut items: Vec<ThreadItem> = Vec::with_capacity(page_size);
let mut scanned_files = 0usize;
let mut anchor_state = AnchorState::new(anchor);
let mut more_matches_available = false;
let candidates = collect_files_by_updated_at(&root, &mut scanned_files).await?;
let mut candidates = candidates;
let mut candidates = collect_files_by_updated_at(&root, &mut scanned_files).await?;
candidates.sort_by_key(|candidate| {
let ts = candidate.updated_at.unwrap_or(OffsetDateTime::UNIX_EPOCH);
(Reverse(ts), Reverse(candidate.id))
@@ -527,6 +555,7 @@ async fn traverse_directories_for_paths_updated(
candidate.path,
allowed_sources,
provider_matcher,
cwd_filters,
updated_at_fallback,
)
.await
@@ -559,6 +588,7 @@ async fn traverse_flat_paths_created(
anchor: Option<Cursor>,
allowed_sources: &[SessionSource],
provider_matcher: Option<&ProviderMatcher<'_>>,
cwd_filters: Option<&[PathBuf]>,
) -> io::Result<ThreadsPage> {
let mut items: Vec<ThreadItem> = Vec::with_capacity(page_size);
let mut scanned_files = 0usize;
@@ -578,8 +608,14 @@ async fn traverse_flat_paths_created(
.await
.unwrap_or(None)
.and_then(format_rfc3339);
if let Some(item) =
build_thread_item(path, allowed_sources, provider_matcher, updated_at).await
if let Some(item) = build_thread_item(
path,
allowed_sources,
provider_matcher,
cwd_filters,
updated_at,
)
.await
{
items.push(item);
}
@@ -609,14 +645,14 @@ async fn traverse_flat_paths_updated(
anchor: Option<Cursor>,
allowed_sources: &[SessionSource],
provider_matcher: Option<&ProviderMatcher<'_>>,
cwd_filters: Option<&[PathBuf]>,
) -> io::Result<ThreadsPage> {
let mut items: Vec<ThreadItem> = Vec::with_capacity(page_size);
let mut scanned_files = 0usize;
let mut anchor_state = AnchorState::new(anchor);
let mut more_matches_available = false;
let candidates = collect_flat_files_by_updated_at(&root, &mut scanned_files).await?;
let mut candidates = candidates;
let mut candidates = collect_flat_files_by_updated_at(&root, &mut scanned_files).await?;
candidates.sort_by_key(|candidate| {
let ts = candidate.updated_at.unwrap_or(OffsetDateTime::UNIX_EPOCH);
(Reverse(ts), Reverse(candidate.id))
@@ -637,6 +673,7 @@ async fn traverse_flat_paths_updated(
candidate.path,
allowed_sources,
provider_matcher,
cwd_filters,
updated_at_fallback,
)
.await
@@ -698,6 +735,7 @@ async fn build_thread_item(
path: PathBuf,
allowed_sources: &[SessionSource],
provider_matcher: Option<&ProviderMatcher<'_>>,
cwd_filters: Option<&[PathBuf]>,
updated_at: Option<String>,
) -> Option<ThreadItem> {
// Read head and detect message events; stop once meta + user are found.
@@ -717,6 +755,15 @@ async fn build_thread_item(
{
return None;
}
if let Some(cwd_filters) = cwd_filters
&& !summary.cwd.as_ref().is_some_and(|cwd| {
cwd_filters
.iter()
.any(|filter| path_utils::paths_match_after_normalization(cwd, filter))
})
{
return None;
}
// Apply filters: must have session meta and at least one user message event
if summary.saw_session_meta && summary.saw_user_event {
let HeadTailSummary {
@@ -768,6 +815,7 @@ pub async fn read_thread_item_from_rollout(path: PathBuf) -> Option<ThreadItem>
path,
&[],
/*provider_matcher*/ None,
/*cwd_filters*/ None,
/*updated_at*/ None,
)
.await
codex-rs/rollout/src/recorder.rs
+223 -33
View File
@@ -214,6 +214,18 @@ fn sanitize_rollout_item_for_persistence(
}
}
#[derive(Clone, Copy)]
enum ThreadListArchiveFilter {
Active,
Archived,
}
#[derive(Clone, Copy)]
enum ThreadListRepairMode {
ScanAndRepair,
StateDbOnly,
}
impl RolloutRecorder {
/// List threads (rollout files) under the provided Codex home directory.
#[allow(clippy::too_many_arguments)]
@@ -225,6 +237,7 @@ impl RolloutRecorder {
sort_direction: SortDirection,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
cwd_filters: Option<&[PathBuf]>,
default_provider: &str,
search_term: Option<&str>,
) -> std::io::Result<ThreadsPage> {
@@ -236,8 +249,40 @@ impl RolloutRecorder {
sort_direction,
allowed_sources,
model_providers,
cwd_filters,
default_provider,
/*archived*/ false,
ThreadListArchiveFilter::Active,
ThreadListRepairMode::ScanAndRepair,
search_term,
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn list_threads_from_state_db(
config: &impl RolloutConfigView,
page_size: usize,
cursor: Option<&Cursor>,
sort_key: ThreadSortKey,
sort_direction: SortDirection,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
cwd_filters: Option<&[PathBuf]>,
default_provider: &str,
search_term: Option<&str>,
) -> std::io::Result<ThreadsPage> {
Self::list_threads_with_db_fallback(
config,
page_size,
cursor,
sort_key,
sort_direction,
allowed_sources,
model_providers,
cwd_filters,
default_provider,
ThreadListArchiveFilter::Active,
ThreadListRepairMode::StateDbOnly,
search_term,
)
.await
@@ -253,6 +298,7 @@ impl RolloutRecorder {
sort_direction: SortDirection,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
cwd_filters: Option<&[PathBuf]>,
default_provider: &str,
search_term: Option<&str>,
) -> std::io::Result<ThreadsPage> {
@@ -264,8 +310,40 @@ impl RolloutRecorder {
sort_direction,
allowed_sources,
model_providers,
cwd_filters,
default_provider,
/*archived*/ true,
ThreadListArchiveFilter::Archived,
ThreadListRepairMode::ScanAndRepair,
search_term,
)
.await
}
#[allow(clippy::too_many_arguments)]
pub async fn list_archived_threads_from_state_db(
config: &impl RolloutConfigView,
page_size: usize,
cursor: Option<&Cursor>,
sort_key: ThreadSortKey,
sort_direction: SortDirection,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
cwd_filters: Option<&[PathBuf]>,
default_provider: &str,
search_term: Option<&str>,
) -> std::io::Result<ThreadsPage> {
Self::list_threads_with_db_fallback(
config,
page_size,
cursor,
sort_key,
sort_direction,
allowed_sources,
model_providers,
cwd_filters,
default_provider,
ThreadListArchiveFilter::Archived,
ThreadListRepairMode::StateDbOnly,
search_term,
)
.await
@@ -280,19 +358,24 @@ impl RolloutRecorder {
sort_direction: SortDirection,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
cwd_filters: Option<&[PathBuf]>,
default_provider: &str,
archived: bool,
archive_filter: ThreadListArchiveFilter,
repair_mode: ThreadListRepairMode,
search_term: Option<&str>,
) -> std::io::Result<ThreadsPage> {
let codex_home = config.codex_home();
let state_db_ctx = state_db::get_state_db(config).await;
let archived = match archive_filter {
ThreadListArchiveFilter::Active => false,
ThreadListArchiveFilter::Archived => true,
};
if cwd_filters.is_some_and(<[std::path::PathBuf]>::is_empty) {
return Ok(ThreadsPage::default());
}
// Search is the SQLite-optimized path and assumes a DB marked backfill-complete is
// actually populated enough to answer the query. If unmigrated rollout files still exist
// on disk, the repair path below may or may not run and catch them depending on whether
// SQLite already has another matching search hit.
if search_term.is_some()
&& let Some(db_page) = state_db::list_threads_db(
if matches!(repair_mode, ThreadListRepairMode::StateDbOnly) {
return Ok(state_db::list_threads_db(
state_db_ctx.as_deref(),
codex_home,
page_size,
@@ -301,18 +384,22 @@ impl RolloutRecorder {
sort_direction,
allowed_sources,
model_providers,
cwd_filters,
archived,
search_term,
)
.await
&& (!db_page.items.is_empty() || cursor.is_some())
{
return Ok(db_page.into());
.map(Into::into)
.unwrap_or_default());
}
let listing_has_metadata_filters = !allowed_sources.is_empty()
|| model_providers.is_some()
|| cwd_filters.is_some()
|| search_term.is_some();
// Filesystem-first listing intentionally overfetches so we can repair stale/missing
// SQLite rollout paths before the final DB-backed page is returned.
let fs_page_size = page_size.saturating_mul(2).max(page_size);
// SQLite rows before returning the scan page for filtered listings or the DB page for
// unfiltered listings.
let fs_page = match sort_direction {
SortDirection::Asc => {
list_threads_from_files_asc(
@@ -322,6 +409,7 @@ impl RolloutRecorder {
sort_key,
allowed_sources,
model_providers,
cwd_filters,
default_provider,
archived,
search_term,
@@ -331,11 +419,12 @@ impl RolloutRecorder {
SortDirection::Desc => {
list_threads_from_files_desc(
codex_home,
fs_page_size,
page_size.saturating_mul(2),
cursor,
sort_key,
allowed_sources,
model_providers,
cwd_filters,
default_provider,
archived,
search_term,
@@ -347,24 +436,39 @@ impl RolloutRecorder {
if state_db_ctx.is_none() {
// Keep legacy behavior when SQLite is unavailable: return filesystem results
// at the requested page size.
return Ok(match sort_direction {
SortDirection::Asc => fs_page,
SortDirection::Desc => truncate_fs_page(fs_page, page_size, sort_key),
});
return Ok(page_from_filesystem_scan(
fs_page,
sort_direction,
page_size,
sort_key,
));
}
// Warm the DB by repairing every filesystem hit before querying SQLite.
for item in &fs_page.items {
state_db::read_repair_rollout_path(
state_db_ctx.as_deref(),
item.thread_id,
Some(archived),
item.path.as_path(),
)
.await;
if listing_has_metadata_filters {
state_db::reconcile_rollout(
state_db_ctx.as_deref(),
item.path.as_path(),
default_provider,
/*builder*/ None,
&[],
Some(archived),
/*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;
}
}
if let Some(db_page) = state_db::list_threads_db(
let db_page = state_db::list_threads_db(
state_db_ctx.as_deref(),
codex_home,
page_size,
@@ -373,20 +477,83 @@ impl RolloutRecorder {
sort_direction,
allowed_sources,
model_providers,
cwd_filters,
archived,
search_term,
)
.await
{
.await;
if let Some(db_page) = db_page {
if search_term.is_some() && (!db_page.items.is_empty() || cursor.is_some()) {
for item in &db_page.items {
state_db::reconcile_rollout(
state_db_ctx.as_deref(),
item.rollout_path.as_path(),
default_provider,
/*builder*/ None,
&[],
Some(archived),
/*new_thread_memory_mode*/ None,
)
.await;
}
if let Some(repaired_db_page) = state_db::list_threads_db(
state_db_ctx.as_deref(),
codex_home,
page_size,
cursor,
sort_key,
sort_direction,
allowed_sources,
model_providers,
cwd_filters,
archived,
search_term,
)
.await
{
return Ok(repaired_db_page.into());
}
return Ok(db_page.into());
}
if listing_has_metadata_filters {
for item in &db_page.items {
state_db::reconcile_rollout(
state_db_ctx.as_deref(),
item.rollout_path.as_path(),
default_provider,
/*builder*/ None,
&[],
Some(archived),
/*new_thread_memory_mode*/ None,
)
.await;
}
return Ok(page_from_filesystem_scan(
fs_page,
sort_direction,
page_size,
sort_key,
));
}
return Ok(db_page.into());
}
if listing_has_metadata_filters {
return Ok(page_from_filesystem_scan(
fs_page,
sort_direction,
page_size,
sort_key,
));
}
// If SQLite listing still fails, return the filesystem page rather than failing the list.
tracing::error!("Falling back on rollout system");
tracing::warn!("state db discrepancy during list_threads_with_db_fallback: falling_back");
Ok(match sort_direction {
SortDirection::Asc => fs_page,
SortDirection::Desc => truncate_fs_page(fs_page, page_size, sort_key),
})
Ok(page_from_filesystem_scan(
fs_page,
sort_direction,
page_size,
sort_key,
))
}
/// Find the newest recorded thread path, optionally filtering to a matching cwd.
@@ -403,6 +570,7 @@ impl RolloutRecorder {
) -> std::io::Result<Option<PathBuf>> {
let codex_home = config.codex_home();
let state_db_ctx = state_db::get_state_db(config).await;
let cwd_filter = filter_cwd.map(Path::to_path_buf);
if state_db_ctx.is_some() {
let mut db_cursor = cursor.cloned();
loop {
@@ -415,6 +583,7 @@ impl RolloutRecorder {
SortDirection::Desc,
allowed_sources,
model_providers,
cwd_filter.as_ref().map(std::slice::from_ref),
/*archived*/ false,
/*search_term*/ None,
)
@@ -443,6 +612,7 @@ impl RolloutRecorder {
sort_key,
allowed_sources,
model_providers,
cwd_filter.as_ref().map(std::slice::from_ref),
default_provider,
)
.await?;
@@ -796,6 +966,18 @@ fn truncate_fs_page(
page
}
fn page_from_filesystem_scan(
page: ThreadsPage,
sort_direction: SortDirection,
page_size: usize,
sort_key: ThreadSortKey,
) -> ThreadsPage {
match sort_direction {
SortDirection::Asc => page,
SortDirection::Desc => truncate_fs_page(page, page_size, sort_key),
}
}
#[allow(clippy::too_many_arguments)]
async fn list_threads_from_files_desc(
codex_home: &Path,
@@ -804,6 +986,7 @@ async fn list_threads_from_files_desc(
sort_key: ThreadSortKey,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
cwd_filters: Option<&[PathBuf]>,
default_provider: &str,
archived: bool,
search_term: Option<&str>,
@@ -823,6 +1006,7 @@ async fn list_threads_from_files_desc(
sort_key,
allowed_sources,
model_providers,
cwd_filters,
default_provider,
archived,
)
@@ -864,6 +1048,7 @@ async fn list_threads_from_files_desc(
sort_key,
allowed_sources,
model_providers,
cwd_filters,
default_provider,
archived,
)
@@ -878,6 +1063,7 @@ async fn list_threads_from_files_desc_unfiltered(
sort_key: ThreadSortKey,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
cwd_filters: Option<&[PathBuf]>,
default_provider: &str,
archived: bool,
) -> std::io::Result<ThreadsPage> {
@@ -891,6 +1077,7 @@ async fn list_threads_from_files_desc_unfiltered(
ThreadListConfig {
allowed_sources,
model_providers,
cwd_filters,
default_provider,
layout: ThreadListLayout::Flat,
},
@@ -904,6 +1091,7 @@ async fn list_threads_from_files_desc_unfiltered(
sort_key,
allowed_sources,
model_providers,
cwd_filters,
default_provider,
)
.await
@@ -918,6 +1106,7 @@ async fn list_threads_from_files_asc(
sort_key: ThreadSortKey,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
cwd_filters: Option<&[PathBuf]>,
default_provider: &str,
archived: bool,
search_term: Option<&str>,
@@ -935,6 +1124,7 @@ async fn list_threads_from_files_asc(
sort_key,
allowed_sources,
model_providers,
cwd_filters,
default_provider,
archived,
/*search_term*/ None,
codex-rs/rollout/src/recorder_tests.rs
+272
View File
@@ -375,6 +375,7 @@ async fn list_threads_db_disabled_does_not_skip_paginated_items() -> std::io::Re
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
default_provider.as_str(),
/*search_term*/ None,
)
@@ -391,6 +392,7 @@ async fn list_threads_db_disabled_does_not_skip_paginated_items() -> std::io::Re
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
default_provider.as_str(),
/*search_term*/ None,
)
@@ -449,6 +451,7 @@ async fn list_threads_db_enabled_drops_missing_rollout_paths() -> std::io::Resul
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
default_provider.as_str(),
/*search_term*/ None,
)
@@ -512,6 +515,7 @@ async fn list_threads_db_enabled_repairs_stale_rollout_paths() -> std::io::Resul
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
default_provider.as_str(),
/*search_term*/ None,
)
@@ -527,6 +531,274 @@ async fn list_threads_db_enabled_repairs_stale_rollout_paths() -> std::io::Resul
Ok(())
}
#[tokio::test]
async fn list_threads_state_db_only_skips_jsonl_repair_scan() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init(
home.path().to_path_buf(),
config.model_provider_id.clone(),
)
.await
.expect("state db should initialize");
runtime
.mark_backfill_complete(/*last_watermark*/ None)
.await
.expect("backfill should be complete");
let uuid = Uuid::from_u128(9012);
let ts = "2025-01-03T14-00-00";
let day_dir = home.path().join("sessions/2025/01/03");
fs::create_dir_all(&day_dir)?;
let path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl"));
let mut file = File::create(&path)?;
let meta = serde_json::json!({
"timestamp": ts,
"type": "session_meta",
"payload": {
"id": uuid,
"timestamp": ts,
"cwd": home.path().display().to_string(),
"originator": "test_originator",
"cli_version": "test_version",
"source": "cli",
"model_provider": "test-provider",
},
});
writeln!(file, "{meta}")?;
let user_event = serde_json::json!({
"timestamp": ts,
"type": "event_msg",
"payload": {
"type": "user_message",
"message": "Hello from user",
"kind": "plain",
},
});
writeln!(file, "{user_event}")?;
let cwd_filters = [home.path().to_path_buf()];
let state_db_only_page = RolloutRecorder::list_threads_from_state_db(
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ Some(cwd_filters.as_slice()),
config.model_provider_id.as_str(),
/*search_term*/ None,
)
.await?;
assert_eq!(state_db_only_page.items.len(), 0);
let repaired_page = RolloutRecorder::list_threads(
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ Some(cwd_filters.as_slice()),
config.model_provider_id.as_str(),
/*search_term*/ None,
)
.await?;
assert_eq!(repaired_page.items.len(), 1);
let repaired_state_db_only_page = RolloutRecorder::list_threads_from_state_db(
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ Some(cwd_filters.as_slice()),
config.model_provider_id.as_str(),
/*search_term*/ None,
)
.await?;
assert_eq!(repaired_state_db_only_page.items.len(), 1);
Ok(())
}
#[tokio::test]
async fn list_threads_default_filter_returns_filesystem_scan_results() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let uuid = Uuid::from_u128(9013);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let real_path = write_session_file(home.path(), "2025-01-03T13-00-00", uuid)?;
let stale_cwd = home.path().join("stale-cwd");
let runtime = codex_state::StateRuntime::init(
home.path().to_path_buf(),
config.model_provider_id.clone(),
)
.await
.expect("state db should initialize");
runtime
.mark_backfill_complete(/*last_watermark*/ None)
.await
.expect("backfill should be complete");
let created_at = chrono::Utc
.with_ymd_and_hms(2025, 1, 3, 13, 0, 0)
.single()
.expect("valid datetime");
let mut builder = codex_state::ThreadMetadataBuilder::new(
thread_id,
real_path,
created_at,
SessionSource::Cli,
);
builder.model_provider = Some(config.model_provider_id.clone());
builder.cwd = stale_cwd.clone();
let mut metadata = builder.build(config.model_provider_id.as_str());
metadata.first_user_message = Some("Hello from user".to_string());
runtime
.upsert_thread(&metadata)
.await
.expect("state db upsert should succeed");
let cwd_filters = [stale_cwd];
let state_db_only_page = RolloutRecorder::list_threads_from_state_db(
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ Some(cwd_filters.as_slice()),
config.model_provider_id.as_str(),
/*search_term*/ None,
)
.await?;
assert_eq!(state_db_only_page.items.len(), 1);
let scanned_page = RolloutRecorder::list_threads(
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ Some(cwd_filters.as_slice()),
config.model_provider_id.as_str(),
/*search_term*/ None,
)
.await?;
assert_eq!(scanned_page.items.len(), 0);
let repaired_state_db_only_page = RolloutRecorder::list_threads_from_state_db(
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ Some(cwd_filters.as_slice()),
config.model_provider_id.as_str(),
/*search_term*/ None,
)
.await?;
assert_eq!(repaired_state_db_only_page.items.len(), 0);
Ok(())
}
#[tokio::test]
async fn list_threads_search_repairs_stale_state_db_hits_before_returning() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let uuid = Uuid::from_u128(9014);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let real_path = write_session_file(home.path(), "2025-01-03T15-00-00", uuid)?;
let runtime = codex_state::StateRuntime::init(
home.path().to_path_buf(),
config.model_provider_id.clone(),
)
.await
.expect("state db should initialize");
runtime
.mark_backfill_complete(/*last_watermark*/ None)
.await
.expect("backfill should be complete");
let created_at = chrono::Utc
.with_ymd_and_hms(2025, 1, 3, 15, 0, 0)
.single()
.expect("valid datetime");
let mut builder = codex_state::ThreadMetadataBuilder::new(
thread_id,
real_path,
created_at,
SessionSource::Cli,
);
builder.model_provider = Some(config.model_provider_id.clone());
builder.cwd = home.path().to_path_buf();
let mut metadata = builder.build(config.model_provider_id.as_str());
metadata.title = "needle stale title".to_string();
metadata.first_user_message = Some("stale first user".to_string());
runtime
.upsert_thread(&metadata)
.await
.expect("state db upsert should succeed");
let stale_state_db_only_page = RolloutRecorder::list_threads_from_state_db(
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
config.model_provider_id.as_str(),
Some("needle"),
)
.await?;
assert_eq!(stale_state_db_only_page.items.len(), 1);
let scanned_page = RolloutRecorder::list_threads(
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
config.model_provider_id.as_str(),
Some("needle"),
)
.await?;
assert_eq!(scanned_page.items.len(), 0);
let repaired_state_db_only_page = RolloutRecorder::list_threads_from_state_db(
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
config.model_provider_id.as_str(),
Some("needle"),
)
.await?;
assert_eq!(repaired_state_db_only_page.items.len(), 0);
Ok(())
}
#[tokio::test]
async fn resume_candidate_matches_cwd_reads_latest_turn_context() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
codex-rs/rollout/src/state_db.rs
+8
View File
@@ -191,6 +191,7 @@ pub async fn list_threads_db(
sort_direction: SortDirection,
allowed_sources: &[SessionSource],
model_providers: Option<&[String]>,
cwd_filters: Option<&[PathBuf]>,
archived: bool,
search_term: Option<&str>,
) -> Option<codex_state::ThreadsPage> {
@@ -213,6 +214,12 @@ pub async fn list_threads_db(
})
.collect();
let model_providers = model_providers.map(<[String]>::to_vec);
let normalized_cwd_filters = cwd_filters.map(|filters| {
filters
.iter()
.map(|cwd| normalize_cwd_for_state_db(cwd))
.collect::<Vec<_>>()
});
match ctx
.list_threads(
page_size,
@@ -220,6 +227,7 @@ pub async fn list_threads_db(
archived_only: archived,
allowed_sources: allowed_sources.as_slice(),
model_providers: model_providers.as_deref(),
cwd_filters: normalized_cwd_filters.as_deref(),
anchor: anchor.as_ref(),
sort_key: match sort_key {
ThreadSortKey::CreatedAt => codex_state::SortKey::CreatedAt,
codex-rs/rollout/src/tests.rs
+18
View File
@@ -540,6 +540,7 @@ async fn test_list_conversations_latest_first() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -689,6 +690,7 @@ async fn test_pagination_cursor() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -756,6 +758,7 @@ async fn test_pagination_cursor() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -823,6 +826,7 @@ async fn test_pagination_cursor() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -877,6 +881,7 @@ async fn test_list_threads_scans_past_head_for_user_event() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -910,6 +915,7 @@ async fn test_get_thread_contents() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -1000,6 +1006,7 @@ async fn test_base_instructions_missing_in_meta_defaults_to_null() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -1043,6 +1050,7 @@ async fn test_base_instructions_present_in_meta_is_preserved() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -1101,6 +1109,7 @@ async fn test_created_at_sort_uses_file_mtime_for_updated_at() -> Result<()> {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await?;
@@ -1186,6 +1195,7 @@ async fn test_updated_at_uses_file_mtime() -> Result<()> {
ThreadSortKey::UpdatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await?;
@@ -1247,6 +1257,7 @@ async fn test_timestamp_only_cursor_skips_same_second_filesystem_ties() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -1315,6 +1326,7 @@ async fn test_timestamp_only_cursor_skips_same_second_filesystem_ties() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -1363,6 +1375,7 @@ async fn test_source_filter_excludes_non_matching_sessions() {
ThreadSortKey::CreatedAt,
INTERACTIVE_SESSION_SOURCES.as_slice(),
Some(provider_filter.as_slice()),
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -1385,6 +1398,7 @@ async fn test_source_filter_excludes_non_matching_sessions() {
ThreadSortKey::CreatedAt,
NO_SOURCE_FILTER,
/*model_providers*/ None,
/*cwd_filters*/ None,
TEST_PROVIDER,
)
.await
@@ -1447,6 +1461,7 @@ async fn test_model_provider_filter_selects_only_matching_sessions() -> Result<(
ThreadSortKey::CreatedAt,
NO_SOURCE_FILTER,
Some(openai_filter.as_slice()),
/*cwd_filters*/ None,
"openai",
)
.await?;
@@ -1467,6 +1482,7 @@ async fn test_model_provider_filter_selects_only_matching_sessions() -> Result<(
ThreadSortKey::CreatedAt,
NO_SOURCE_FILTER,
Some(beta_filter.as_slice()),
/*cwd_filters*/ None,
"openai",
)
.await?;
@@ -1486,6 +1502,7 @@ async fn test_model_provider_filter_selects_only_matching_sessions() -> Result<(
ThreadSortKey::CreatedAt,
NO_SOURCE_FILTER,
Some(unknown_filter.as_slice()),
/*cwd_filters*/ None,
"openai",
)
.await?;
@@ -1498,6 +1515,7 @@ async fn test_model_provider_filter_selects_only_matching_sessions() -> Result<(
ThreadSortKey::CreatedAt,
NO_SOURCE_FILTER,
/*model_providers*/ None,
/*cwd_filters*/ None,
"openai",
)
.await?;