repair stale rollout paths

This commit is contained in:
anais
2026-06-15 21:59:35 -07:00
parent d6fcb78f07
commit b9decc0a0c
7 changed files with 737 additions and 147 deletions
+1
View File
@@ -48,6 +48,7 @@ pub use list::ThreadListConfig;
pub use list::ThreadListLayout;
pub use list::ThreadSortKey;
pub use list::ThreadsPage;
pub use list::find_any_thread_path_by_id_str;
pub use list::find_archived_thread_path_by_id_str;
pub use list::find_thread_path_by_id_str;
#[deprecated(note = "use find_thread_path_by_id_str")]
+349 -77
View File
@@ -1766,9 +1766,9 @@ async fn find_thread_path_by_id_str_in_subdir(
state_db_ctx: Option<&codex_state::StateRuntime>,
) -> io::Result<Option<PathBuf>> {
// Validate UUID format early.
if Uuid::parse_str(id_str).is_err() {
let Ok(target_uuid) = Uuid::parse_str(id_str) else {
return Ok(None);
}
};
// Prefer DB lookup, then fall back to rollout file search.
// TODO(jif): sqlite migration phase 1
@@ -1778,19 +1778,23 @@ async fn find_thread_path_by_id_str_in_subdir(
_ => None,
};
let thread_id = ThreadId::from_string(id_str).ok();
let mut state_metadata_path = None;
let mut unverified_db_path = None;
let mut fallback_reason = state_db_ctx.is_none().then_some("db_unavailable");
if let Some(state_db_ctx) = state_db_ctx
&& let Some(thread_id) = thread_id
{
match state_db_ctx
.find_rollout_path_by_id(thread_id, archived_only)
.find_rollout_path_and_created_at_by_id(thread_id, archived_only)
.await
{
Ok(Some(db_path)) => {
Ok(Some((db_path, created_at))) => {
if let Some(existing_db_path) =
compression::existing_rollout_path(db_path.as_path()).await
compression::existing_rollout_path_blocking(db_path.as_path())
{
if rollout_file_name_matches_uuid(existing_db_path.as_path(), target_uuid) {
return Ok(Some(existing_db_path));
}
match read_session_meta_line(&existing_db_path).await {
Ok(meta_line) if meta_line.meta.id == thread_id => {
return Ok(Some(existing_db_path));
@@ -1809,6 +1813,17 @@ async fn find_thread_path_by_id_str_in_subdir(
"mismatch",
/*telemetry_override*/ None,
);
state_metadata_path = find_rollout_path_by_id_from_state_created_at(
codex_home,
subdir,
target_uuid,
thread_id,
archived_only,
state_db_ctx,
db_path.as_path(),
created_at,
)
.await;
}
Err(err) => {
tracing::debug!(
@@ -1816,6 +1831,7 @@ async fn find_thread_path_by_id_str_in_subdir(
existing_db_path.display()
);
unverified_db_path = Some(existing_db_path);
fallback_reason = Some("unverified_path");
}
}
} else {
@@ -1831,6 +1847,17 @@ async fn find_thread_path_by_id_str_in_subdir(
"stale_path",
/*telemetry_override*/ None,
);
state_metadata_path = find_rollout_path_by_id_from_state_created_at(
codex_home,
subdir,
target_uuid,
thread_id,
archived_only,
state_db_ctx,
db_path.as_path(),
created_at,
)
.await;
}
}
Ok(None) => fallback_reason = Some("missing_row"),
@@ -1843,61 +1870,13 @@ async fn find_thread_path_by_id_str_in_subdir(
}
}
let mut root = codex_home.to_path_buf();
root.push(subdir);
if !root.exists() {
return Ok(unverified_db_path);
}
let (filename_match, filename_scan_error) = match find_rollout_path_by_id_from_filenames(
root.as_path(),
id_str,
)
.await
{
Ok(path) => (path, None),
Err(err) => {
tracing::warn!(
"rollout filename lookup failed during find_thread_path_by_id_str_in_subdir: {err}"
);
(None, Some(err))
}
};
let found = match filename_match {
Some(path) => Some(path),
None => {
// This is safe because we know the values are valid.
#[allow(clippy::unwrap_used)]
let limit = NonZero::new(1).unwrap();
let options = file_search::FileSearchOptions {
limit,
compute_indices: false,
respect_gitignore: false,
..Default::default()
};
let results = file_search::run(
id_str,
vec![root.clone()],
options,
/*cancel_flag*/ None,
)
.map_err(|e| io::Error::other(format!("file search failed: {e}")))?;
let found = results
.matches
.into_iter()
.map(|m| m.full_path())
.find_map(compression::RolloutFile::from_path)
.map(compression::RolloutFile::into_path);
if found.is_none()
&& let Some(err) = filename_scan_error
{
return Err(err);
}
found
let mut found_path_repaired = false;
let found = match state_metadata_path {
Some((path, repaired)) => {
found_path_repaired = repaired;
Some(path)
}
None => find_thread_path_by_id_str_in_subdir_without_db(codex_home, subdir, id_str).await?,
};
if let Some(found_path) = found.as_ref() {
tracing::debug!("state db missing rollout path for thread {id_str}");
@@ -1911,18 +1890,294 @@ async fn find_thread_path_by_id_str_in_subdir(
/*telemetry_override*/ None,
);
}
state_db::read_repair_rollout_path(
state_db_ctx,
thread_id,
archived_only,
found_path.as_path(),
)
.await;
if !found_path_repaired {
state_db::read_repair_rollout_path(
state_db_ctx,
thread_id,
archived_only,
found_path.as_path(),
)
.await;
}
}
Ok(found.or(unverified_db_path))
}
async fn find_thread_path_by_id_str_in_any_subdir(
codex_home: &Path,
id_str: &str,
state_db_ctx: Option<&codex_state::StateRuntime>,
) -> io::Result<Option<PathBuf>> {
let Ok(target_uuid) = Uuid::parse_str(id_str) else {
return Ok(None);
};
let thread_id = ThreadId::from_string(id_str).ok();
let mut preferred_subdir = None;
let mut unverified_db_path = None;
let mut fallback_reason = state_db_ctx.is_none().then_some("db_unavailable");
if let Some(state_db_ctx) = state_db_ctx
&& let Some(thread_id) = thread_id
{
match state_db_ctx
.find_rollout_path_and_created_at_by_id(thread_id, /*archived_only*/ None)
.await
{
Ok(Some((db_path, created_at))) => {
preferred_subdir = rollout_subdir_hint(codex_home, db_path.as_path());
if let Some(existing_db_path) =
compression::existing_rollout_path_blocking(db_path.as_path())
{
if rollout_file_name_matches_uuid(existing_db_path.as_path(), target_uuid) {
return Ok(Some(existing_db_path));
}
match read_session_meta_line(&existing_db_path).await {
Ok(meta_line) if meta_line.meta.id == thread_id => {
return Ok(Some(existing_db_path));
}
Ok(meta_line) => {
tracing::error!(
"state db returned rollout path for thread {id_str} but file belongs to thread {}: {}",
meta_line.meta.id,
existing_db_path.display()
);
tracing::warn!(
"state db discrepancy during find_thread_path_by_id_str_in_any_subdir: mismatched_db_path"
);
codex_state::record_fallback(
"find_thread_path",
"mismatch",
/*telemetry_override*/ None,
);
fallback_reason = Some("mismatch");
}
Err(err) => {
tracing::debug!(
"state db returned rollout path for thread {id_str} that could not be verified: {}: {err}",
existing_db_path.display()
);
unverified_db_path = Some(existing_db_path);
fallback_reason = Some("unverified_path");
}
}
} else {
tracing::error!(
"state db returned stale rollout path for thread {id_str}: {}",
db_path.display()
);
tracing::warn!(
"state db discrepancy during find_thread_path_by_id_str_in_any_subdir: stale_db_path"
);
codex_state::record_fallback(
"find_thread_path",
"stale_path",
/*telemetry_override*/ None,
);
fallback_reason = Some("stale_path");
}
if let Some(subdir) = preferred_subdir
&& let Some((path, _repaired)) = find_rollout_path_by_id_from_state_created_at(
codex_home,
subdir,
target_uuid,
thread_id,
archived_only_for_subdir(subdir),
state_db_ctx,
db_path.as_path(),
created_at,
)
.await
{
return Ok(Some(path));
}
}
Ok(None) => fallback_reason = Some("missing_row"),
Err(err) => {
tracing::warn!(
"state db find_rollout_path_by_id failed during find_any_path_query: {err}"
);
fallback_reason = Some("db_error");
}
}
}
for (subdir, archived_only) in ordered_session_subdirs(preferred_subdir) {
if let Some(path) =
find_thread_path_by_id_str_in_subdir_without_db(codex_home, subdir, id_str).await?
{
tracing::debug!("state db missing rollout path for thread {id_str}");
tracing::warn!(
"state db discrepancy during find_thread_path_by_id_str_in_any_subdir: falling_back"
);
if let Some(reason) = fallback_reason {
codex_state::record_fallback(
"find_thread_path",
reason,
/*telemetry_override*/ None,
);
}
state_db::read_repair_rollout_path(
state_db_ctx,
thread_id,
archived_only,
path.as_path(),
)
.await;
return Ok(Some(path));
}
}
Ok(unverified_db_path)
}
pub(crate) async fn find_thread_path_by_id_str_in_subdir_without_db(
codex_home: &Path,
subdir: &str,
id_str: &str,
) -> io::Result<Option<PathBuf>> {
let mut root = codex_home.to_path_buf();
root.push(subdir);
if !root.exists() {
return Ok(None);
}
let (filename_match, filename_scan_error) = match find_rollout_path_by_id_from_filenames(
root.as_path(),
id_str,
)
.await
{
Ok(path) => (path, None),
Err(err) => {
tracing::warn!(
"rollout filename lookup failed during find_thread_path_by_id_str fallback scan: {err}"
);
(None, Some(err))
}
};
match filename_match {
Some(path) => Ok(Some(path)),
None => {
#[allow(clippy::unwrap_used)]
let limit = NonZero::new(1).unwrap();
let options = file_search::FileSearchOptions {
limit,
compute_indices: false,
respect_gitignore: false,
..Default::default()
};
let results = file_search::run(id_str, vec![root], options, /*cancel_flag*/ None)
.map_err(|e| io::Error::other(format!("file search failed: {e}")))?;
let found = results
.matches
.into_iter()
.map(|m| m.full_path())
.find_map(compression::RolloutFile::from_path)
.map(compression::RolloutFile::into_path);
if found.is_none()
&& let Some(err) = filename_scan_error
{
return Err(err);
}
Ok(found)
}
}
}
fn archived_only_for_subdir(subdir: &str) -> Option<bool> {
match subdir {
SESSIONS_SUBDIR => Some(false),
ARCHIVED_SESSIONS_SUBDIR => Some(true),
_ => None,
}
}
fn rollout_subdir_hint(codex_home: &Path, path: &Path) -> Option<&'static str> {
if path.starts_with(codex_home.join(ARCHIVED_SESSIONS_SUBDIR)) {
return Some(ARCHIVED_SESSIONS_SUBDIR);
}
if path.starts_with(codex_home.join(SESSIONS_SUBDIR)) {
return Some(SESSIONS_SUBDIR);
}
None
}
fn ordered_session_subdirs(
preferred_subdir: Option<&'static str>,
) -> [(&'static str, Option<bool>); 2] {
match preferred_subdir {
Some(ARCHIVED_SESSIONS_SUBDIR) => [
(ARCHIVED_SESSIONS_SUBDIR, Some(true)),
(SESSIONS_SUBDIR, Some(false)),
],
_ => [
(SESSIONS_SUBDIR, Some(false)),
(ARCHIVED_SESSIONS_SUBDIR, Some(true)),
],
}
}
async fn find_rollout_path_by_id_from_state_created_at(
codex_home: &Path,
subdir: &str,
target_uuid: Uuid,
thread_id: ThreadId,
archived_only: Option<bool>,
state_db_ctx: &codex_state::StateRuntime,
stale_path: &Path,
created_at: DateTime<Utc>,
) -> Option<(PathBuf, bool)> {
let created_at = created_at.with_timezone(&chrono::Local);
let file_name = format!(
"rollout-{:04}-{:02}-{:02}T{:02}-{:02}-{:02}-{target_uuid}.jsonl",
created_at.year(),
created_at.month(),
created_at.day(),
created_at.hour(),
created_at.minute(),
created_at.second(),
);
let candidate = if subdir == ARCHIVED_SESSIONS_SUBDIR {
codex_home.join(subdir).join(file_name)
} else {
codex_home
.join(subdir)
.join(format!("{:04}", created_at.year()))
.join(format!("{:02}", created_at.month()))
.join(format!("{:02}", created_at.day()))
.join(file_name)
};
let existing_path = compression::existing_rollout_path_blocking(candidate.as_path())?;
if !rollout_file_name_matches_uuid(existing_path.as_path(), target_uuid) {
return None;
}
let repaired = state_db_ctx
.update_thread_rollout_path_if_current(
thread_id,
stale_path,
existing_path.as_path(),
archived_only,
)
.await
.unwrap_or_else(|err| {
tracing::warn!(
"state db stale path direct repair failed for thread {thread_id}: {err}"
);
false
});
Some((existing_path, repaired))
}
fn rollout_file_name_matches_uuid(path: &Path, target: Uuid) -> bool {
path.file_name()
.and_then(OsStr::to_str)
.and_then(parse_timestamp_uuid_from_filename)
.is_some_and(|(_ts, id)| id == target)
}
async fn find_rollout_path_by_id_from_filenames(
root: &Path,
id_str: &str,
@@ -1930,33 +2185,41 @@ async fn find_rollout_path_by_id_from_filenames(
let Ok(target) = Uuid::parse_str(id_str) else {
return Ok(None);
};
let root = root.to_path_buf();
tokio::task::spawn_blocking(move || {
find_rollout_path_by_id_from_filenames_blocking(root.as_path(), target)
})
.await
.map_err(io::Error::other)?
}
fn find_rollout_path_by_id_from_filenames_blocking(
root: &Path,
target: Uuid,
) -> io::Result<Option<PathBuf>> {
let mut stack = vec![root.to_path_buf()];
while let Some(dir) = stack.pop() {
let mut read_dir = match tokio::fs::read_dir(dir.as_path()).await {
let read_dir = match std::fs::read_dir(dir.as_path()) {
Ok(read_dir) => read_dir,
Err(err) if err.kind() == io::ErrorKind::NotFound => continue,
Err(err) => return Err(err),
};
while let Some(entry) = read_dir.next_entry().await? {
let path = entry.path();
let file_type = entry.file_type().await?;
for entry in read_dir {
let entry = entry?;
let file_type = entry.file_type()?;
if file_type.is_dir() {
let path = entry.path();
stack.push(path);
continue;
}
if !file_type.is_file() {
continue;
}
let Some(rollout_file) = compression::RolloutFile::from_path(path) else {
continue;
};
let Some((_ts, id)) =
parse_timestamp_uuid_from_filename(rollout_file.plain_file_name())
else {
let Some((_ts, id, path)) = rollout_parts_from_blocking_dir_entry(&entry) else {
continue;
};
if id == target {
return Ok(Some(rollout_file.into_path()));
return Ok(Some(path));
}
}
}
@@ -1984,6 +2247,15 @@ pub async fn find_archived_thread_path_by_id_str(
.await
}
/// Locate a recorded thread rollout file by UUID across active and archived session roots.
pub async fn find_any_thread_path_by_id_str(
codex_home: &Path,
id_str: &str,
state_db_ctx: Option<&codex_state::StateRuntime>,
) -> io::Result<Option<PathBuf>> {
find_thread_path_by_id_str_in_any_subdir(codex_home, id_str, state_db_ctx).await
}
/// Extract the `YYYY/MM/DD` directory components from a rollout filename.
pub fn rollout_date_parts(file_name: &OsStr) -> Option<(String, String, String)> {
let name = file_name.to_string_lossy();
+67
View File
@@ -706,6 +706,73 @@ async fn list_threads_db_enabled_repairs_stale_rollout_paths() -> std::io::Resul
Ok(())
}
#[tokio::test]
async fn state_db_only_list_repairs_stale_rollout_path() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let uuid = Uuid::from_u128(9016);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let real_path = write_session_file(home.path(), "2025-01-03T17-00-00", uuid)?;
let stale_path = home.path().join(format!(
"sessions/2099/01/01/rollout-2099-01-01T00-00-00-{uuid}.jsonl"
));
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, 17, 0, 0)
.single()
.expect("valid datetime");
let mut builder = codex_state::ThreadMetadataBuilder::new(
thread_id,
stale_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.first_user_message = Some("Hello from user".to_string());
metadata.preview = metadata.first_user_message.clone();
runtime
.upsert_thread(&metadata)
.await
.expect("state db upsert should succeed");
let page = RolloutRecorder::list_threads_from_state_db(
Some(runtime.clone()),
&config,
/*page_size*/ 1,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
config.model_provider_id.as_str(),
/*search_term*/ None,
)
.await?;
assert_eq!(page.items.len(), 1);
assert_eq!(page.items[0].path, real_path);
let repaired_path = runtime
.find_rollout_path_by_id(thread_id, Some(false))
.await
.expect("state db lookup should succeed");
assert_eq!(repaired_path, Some(real_path));
Ok(())
}
#[tokio::test]
async fn list_threads_state_db_only_skips_jsonl_repair_scan() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
+83 -22
View File
@@ -438,6 +438,10 @@ pub async fn list_threads_db(
let mut item = item;
item.rollout_path = existing_path;
valid_items.push(item);
} else if let Some(repaired_item) =
repair_stale_rollout_path_from_metadata(ctx, codex_home, &item, archived).await
{
valid_items.push(repaired_item);
} else {
warn!(
"state db list_threads returned stale rollout path for thread {}: {}",
@@ -458,6 +462,44 @@ pub async fn list_threads_db(
}
}
async fn repair_stale_rollout_path_from_metadata(
ctx: &codex_state::StateRuntime,
codex_home: &Path,
item: &codex_state::ThreadMetadata,
archived: bool,
) -> Option<codex_state::ThreadMetadata> {
let subdir = if archived {
crate::ARCHIVED_SESSIONS_SUBDIR
} else {
crate::SESSIONS_SUBDIR
};
let candidate = crate::list::find_thread_path_by_id_str_in_subdir_without_db(
codex_home,
subdir,
&item.id.to_string(),
)
.await
.ok()
.flatten()?;
let existing_path = crate::compression::existing_rollout_path_blocking(candidate.as_path())?;
warn!(
"state db list_threads repaired stale rollout path for thread {}: {} -> {}",
item.id,
item.rollout_path.display(),
existing_path.display()
);
let mut item = item.clone();
item.rollout_path = existing_path;
if let Err(err) = ctx.upsert_thread(&item).await {
warn!(
"state db list_threads stale path repair upsert failed for thread {}: {err}",
item.id
);
}
Some(item)
}
/// Look up the rollout path for a thread id using SQLite.
pub async fn find_rollout_path_by_id(
context: Option<&codex_state::StateRuntime>,
@@ -581,28 +623,9 @@ pub async fn read_repair_rollout_path(
&& let Ok(Some(metadata)) = ctx.get_thread(thread_id).await
{
saw_existing_metadata = true;
let mut repaired = metadata.clone();
repaired.rollout_path = rollout_path.to_path_buf();
repaired.cwd = normalize_cwd_for_state_db(&repaired.cwd);
match archived_only {
Some(true) if repaired.archived_at.is_none() => {
repaired.archived_at = Some(repaired.updated_at);
}
Some(false) => {
repaired.archived_at = None;
}
Some(true) | None => {}
}
if repaired == metadata {
return;
}
warn!("state db discrepancy during read_repair_rollout_path: upsert_needed (fast path)");
if let Err(err) = ctx.upsert_thread(&repaired).await {
warn!(
"state db read-repair upsert failed for {}: {err}",
rollout_path.display()
);
} else {
if read_repair_rollout_path_from_metadata(Some(ctx), metadata, archived_only, rollout_path)
.await
{
return;
}
}
@@ -629,6 +652,44 @@ pub async fn read_repair_rollout_path(
.await;
}
/// Repair a rollout path when the caller already has the metadata row.
pub async fn read_repair_rollout_path_from_metadata(
context: Option<&codex_state::StateRuntime>,
mut metadata: codex_state::ThreadMetadata,
archived_only: Option<bool>,
rollout_path: &Path,
) -> bool {
let Some(ctx) = context else {
return true;
};
let original = metadata.clone();
metadata.rollout_path = rollout_path.to_path_buf();
metadata.cwd = normalize_cwd_for_state_db(&metadata.cwd);
match archived_only {
Some(true) if metadata.archived_at.is_none() => {
metadata.archived_at = Some(metadata.updated_at);
}
Some(false) => {
metadata.archived_at = None;
}
Some(true) | None => {}
}
if metadata == original {
return true;
}
warn!("state db discrepancy during read_repair_rollout_path: upsert_needed (fast path)");
if let Err(err) = ctx.upsert_thread(&metadata).await {
warn!(
"state db read-repair upsert failed for {}: {err}",
rollout_path.display()
);
return false;
}
true
}
/// Apply rollout items incrementally to SQLite.
#[allow(clippy::too_many_arguments)]
pub async fn apply_rollout_items(
+143 -1
View File
@@ -7,6 +7,7 @@ use std::fs::File;
use std::fs::FileTimes;
use std::io::Write;
use std::path::Path;
use std::path::PathBuf;
use chrono::TimeZone;
use pretty_assertions::assert_eq;
@@ -19,6 +20,7 @@ use time::macros::format_description;
use uuid::Uuid;
use crate::INTERACTIVE_SESSION_SOURCES;
use crate::find_any_thread_path_by_id_str;
use crate::find_thread_path_by_id_str;
use crate::list::Cursor;
use crate::list::ThreadItem;
@@ -230,6 +232,85 @@ async fn find_thread_path_accepts_existing_state_db_path_without_canonical_filen
assert_eq!(found, Some(db_rollout_path));
}
#[tokio::test]
async fn find_any_thread_path_prefers_archived_state_db_path() {
let temp = TempDir::new().unwrap();
let home = temp.path();
let uuid = Uuid::from_u128(306);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let active_ts = "2025-01-03T13-00-00";
write_session_file(
home,
active_ts,
uuid,
/*num_records*/ 1,
Some(SessionSource::Cli),
)
.unwrap();
let archived_ts = "2025-01-04T13-00-00";
let archived_rollout_path = write_archived_session_file(
home,
archived_ts,
uuid,
/*num_records*/ 1,
Some(SessionSource::Cli),
)
.unwrap();
let runtime = insert_state_db_thread(
home,
thread_id,
archived_rollout_path.as_path(),
/*archived*/ true,
)
.await;
let found = find_any_thread_path_by_id_str(home, &uuid.to_string(), Some(runtime.as_ref()))
.await
.expect("lookup should succeed");
assert_eq!(found, Some(archived_rollout_path));
}
#[tokio::test]
async fn find_any_thread_path_repairs_stale_archived_db_path() {
let temp = TempDir::new().unwrap();
let home = temp.path();
let uuid = Uuid::from_u128(307);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let ts = "2025-01-03T12-00-00";
let archived_rollout_path = write_archived_session_file(
home,
ts,
uuid,
/*num_records*/ 1,
Some(SessionSource::Cli),
)
.unwrap();
let stale_db_path = home.join(format!(
"archived_sessions/rollout-2099-01-01T00-00-00-{uuid}.jsonl"
));
let runtime = insert_state_db_thread(
home,
thread_id,
stale_db_path.as_path(),
/*archived*/ true,
)
.await;
let found = find_any_thread_path_by_id_str(home, &uuid.to_string(), Some(runtime.as_ref()))
.await
.expect("lookup should succeed");
assert_eq!(found, Some(archived_rollout_path.clone()));
assert_state_db_rollout_path_with_archive(
home,
thread_id,
Some(true),
Some(archived_rollout_path.as_path()),
)
.await;
}
#[test]
fn rollout_date_parts_extracts_directory_components() {
let file_name = OsStr::new("rollout-2025-03-01T09-00-00-123.jsonl");
@@ -244,12 +325,21 @@ async fn assert_state_db_rollout_path(
home: &Path,
thread_id: ThreadId,
expected_path: Option<&Path>,
) {
assert_state_db_rollout_path_with_archive(home, thread_id, Some(false), expected_path).await;
}
async fn assert_state_db_rollout_path_with_archive(
home: &Path,
thread_id: ThreadId,
archived_only: Option<bool>,
expected_path: Option<&Path>,
) {
let runtime = codex_state::StateRuntime::init(home.to_path_buf(), TEST_PROVIDER.to_string())
.await
.expect("state db should initialize");
let path = runtime
.find_rollout_path_by_id(thread_id, Some(false))
.find_rollout_path_by_id(thread_id, archived_only)
.await
.expect("state db lookup should succeed");
assert_eq!(path.as_deref(), expected_path);
@@ -343,6 +433,58 @@ fn write_session_file_with_provider(
Ok((dt, uuid))
}
fn write_archived_session_file(
root: &Path,
ts_str: &str,
uuid: Uuid,
num_records: usize,
source: Option<SessionSource>,
) -> std::io::Result<PathBuf> {
let format: &[FormatItem] =
format_description!("[year]-[month]-[day]T[hour]-[minute]-[second]");
let dt = PrimitiveDateTime::parse(ts_str, format)
.unwrap()
.assume_utc();
let dir = root.join("archived_sessions");
fs::create_dir_all(&dir)?;
let filename = format!("rollout-{ts_str}-{uuid}.jsonl");
let file_path = dir.join(filename);
let mut file = File::create(&file_path)?;
let mut payload = serde_json::json!({
"id": uuid,
"timestamp": ts_str,
"cwd": ".",
"originator": "test_originator",
"cli_version": "test_version",
"base_instructions": null,
});
if let Some(source) = source {
payload["source"] = serde_json::to_value(source).unwrap();
}
payload["model_provider"] = serde_json::Value::String(TEST_PROVIDER.to_string());
let meta = serde_json::json!({
"timestamp": ts_str,
"type": "session_meta",
"payload": payload,
});
writeln!(file, "{meta}")?;
for i in 0..num_records {
let rec = serde_json::json!({
"record_type": "response",
"index": i
});
writeln!(file, "{rec}")?;
}
let times = FileTimes::new().set_modified(dt.into());
file.set_times(times)?;
Ok(file_path)
}
fn write_goal_started_session_file(
root: &Path,
ts_str: &str,
+58
View File
@@ -348,6 +348,64 @@ ON CONFLICT(child_thread_id) DO NOTHING
.map(PathBuf::from))
}
/// Find a rollout path and creation timestamp by thread id using the underlying database.
pub async fn find_rollout_path_and_created_at_by_id(
&self,
id: ThreadId,
archived_only: Option<bool>,
) -> anyhow::Result<Option<(PathBuf, DateTime<Utc>)>> {
let mut builder = QueryBuilder::<Sqlite>::new(
"SELECT rollout_path, created_at_ms AS created_at FROM threads WHERE id = ",
);
builder.push_bind(id.to_string());
match archived_only {
Some(true) => {
builder.push(" AND archived = 1");
}
Some(false) => {
builder.push(" AND archived = 0");
}
None => {}
}
let row = builder.build().fetch_optional(self.pool.as_ref()).await?;
row.map(|row| {
let rollout_path: String = row.try_get("rollout_path")?;
let created_at: i64 = row.try_get("created_at")?;
Ok((
PathBuf::from(rollout_path),
epoch_millis_to_datetime(created_at)?,
))
})
.transpose()
}
/// Repair a stale rollout path in place when the row still points at the expected path.
pub async fn update_thread_rollout_path_if_current(
&self,
id: ThreadId,
expected_rollout_path: &Path,
repaired_rollout_path: &Path,
archived_only: Option<bool>,
) -> anyhow::Result<bool> {
let mut builder = QueryBuilder::<Sqlite>::new("UPDATE threads SET rollout_path = ");
builder.push_bind(repaired_rollout_path.display().to_string());
builder.push(" WHERE id = ");
builder.push_bind(id.to_string());
builder.push(" AND rollout_path = ");
builder.push_bind(expected_rollout_path.display().to_string());
match archived_only {
Some(true) => {
builder.push(" AND archived = 1");
}
Some(false) => {
builder.push(" AND archived = 0");
}
None => {}
}
let result = builder.build().execute(self.pool.as_ref()).await?;
Ok(result.rows_affected() > 0)
}
/// Find the newest thread whose user-facing title exactly matches `title`.
#[allow(clippy::too_many_arguments)]
pub async fn find_thread_by_exact_title(
+36 -47
View File
@@ -5,7 +5,7 @@ use codex_protocol::protocol::AskForApproval;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use codex_rollout::RolloutRecorder;
use codex_rollout::find_archived_thread_path_by_id_str;
use codex_rollout::find_any_thread_path_by_id_str;
use codex_rollout::find_thread_name_by_id;
use codex_rollout::find_thread_path_by_id_str;
use codex_rollout::read_session_meta_line;
@@ -38,35 +38,34 @@ pub(super) async fn read_thread(
store.config.codex_home.as_path(),
metadata.rollout_path.as_path(),
)))
&& (!params.include_history
|| sqlite_rollout_path_can_load_history_for_thread(
store,
&metadata.rollout_path,
thread_id,
)
.await)
{
let metadata_sandbox_policy = metadata.sandbox_policy.clone();
let rollout_thread = matching_rollout_thread_from_sqlite_path(
store,
metadata.rollout_path.as_path(),
thread_id,
params.include_archived,
)
.await;
let mut thread = stored_thread_from_sqlite_metadata(store, metadata).await;
if !params.include_history
&& let Some(rollout_path) = thread.rollout_path.clone()
&& let Ok(mut rollout_thread) = read_thread_from_rollout_path(store, rollout_path).await
&& rollout_thread.thread_id == thread_id
&& (params.include_archived || rollout_thread.archived_at.is_none())
&& !rollout_thread.preview.is_empty()
{
if thread.name.is_some() {
rollout_thread.name = thread.name;
if !params.include_history || rollout_thread.is_some() {
if !params.include_history
&& let Some(mut rollout_thread) = rollout_thread
&& !rollout_thread.preview.is_empty()
{
if thread.name.is_some() {
rollout_thread.name = thread.name;
}
rollout_thread.git_info = thread.git_info;
rollout_thread.permission_profile = permission_profile_from_metadata_value(
&metadata_sandbox_policy,
rollout_thread.cwd.as_path(),
);
thread = rollout_thread;
}
rollout_thread.git_info = thread.git_info;
rollout_thread.permission_profile = permission_profile_from_metadata_value(
&metadata_sandbox_policy,
rollout_thread.cwd.as_path(),
);
thread = rollout_thread;
attach_history_if_requested(&mut thread, params.include_history).await?;
return Ok(thread);
}
attach_history_if_requested(&mut thread, params.include_history).await?;
return Ok(thread);
}
let path = resolve_rollout_path(store, thread_id, params.include_archived)
@@ -85,20 +84,21 @@ pub(super) async fn read_thread(
Ok(thread)
}
async fn sqlite_rollout_path_can_load_history_for_thread(
async fn matching_rollout_thread_from_sqlite_path(
store: &LocalThreadStore,
path: &std::path::Path,
thread_id: codex_protocol::ThreadId,
) -> bool {
if codex_rollout::existing_rollout_path(path).await.is_none() {
return false;
}
// SQLite metadata can outlive a moved/recreated rollout path. When history is
// requested, verify the path still resolves to the requested thread before
// trusting it as the source replay.
include_archived: bool,
) -> Option<StoredThread> {
codex_rollout::existing_rollout_path(path).await.as_ref()?;
// SQLite metadata can outlive a moved or recreated rollout path. Verify the
// path still resolves to the requested thread before trusting its summary or history.
read_thread_from_rollout_path(store, path.to_path_buf())
.await
.is_ok_and(|thread| thread.thread_id == thread_id)
.ok()
.filter(|thread| {
thread.thread_id == thread_id && (include_archived || thread.archived_at.is_none())
})
}
pub(super) async fn read_thread_by_rollout_path(
@@ -209,7 +209,7 @@ async fn resolve_rollout_path(
let state_db_ctx = store.state_db().await;
if include_archived {
match find_thread_path_by_id_str(
find_any_thread_path_by_id_str(
store.config.codex_home.as_path(),
&thread_id.to_string(),
state_db_ctx.as_deref(),
@@ -217,18 +217,7 @@ async fn resolve_rollout_path(
.await
.map_err(|err| ThreadStoreError::InvalidRequest {
message: format!("failed to locate thread id {thread_id}: {err}"),
})? {
Some(path) => Ok(Some(path)),
None => find_archived_thread_path_by_id_str(
store.config.codex_home.as_path(),
&thread_id.to_string(),
state_db_ctx.as_deref(),
)
.await
.map_err(|err| ThreadStoreError::InvalidRequest {
message: format!("failed to locate archived thread id {thread_id}: {err}"),
}),
}
})
} else {
find_thread_path_by_id_str(
store.config.codex_home.as_path(),