feat: compressor

codex-rs/thread-store/src/local/helpers.rs

@@ -72,10 +72,11 @@ pub(super) fn matching_rollout_file_name(
             ),
         });
     };
-    let required_suffix = format!("{thread_id}.jsonl");
-    if file_name
-        .to_string_lossy()
-        .ends_with(required_suffix.as_str())
+    let required_plain_suffix = format!("{thread_id}.jsonl");
+    let required_compressed_suffix = format!("{required_plain_suffix}.zst");
+    let file_name_str = file_name.to_string_lossy();
+    if file_name_str.ends_with(required_plain_suffix.as_str())
+        || file_name_str.ends_with(required_compressed_suffix.as_str())
     {
         Ok(file_name)
     } else {
@@ -186,6 +187,7 @@ pub(super) fn git_info_from_parts(
 
 fn thread_id_from_rollout_path(path: &Path) -> Option<ThreadId> {
     let file_name = path.file_name()?.to_str()?;
+    let file_name = file_name.strip_suffix(".zst").unwrap_or(file_name);
     let stem = file_name.strip_suffix(".jsonl")?;
     if stem.len() < 37 {
         return None;

codex-rs/thread-store/src/local/live_writer.rs

@@ -156,9 +156,9 @@ async fn sync_materialized_rollout_path(
     thread_id: ThreadId,
 ) -> ThreadStoreResult<()> {
     let rollout_path = rollout_path(store, thread_id).await?;
-    if !tokio::fs::try_exists(rollout_path.as_path())
+    if codex_rollout::existing_rollout_path(rollout_path.as_path())
         .await
-        .unwrap_or(false)
+        .is_none()
     {
         return Ok(());
     }

codex-rs/thread-store/src/local/read_thread.rs

@@ -84,7 +84,7 @@ async fn sqlite_rollout_path_can_load_history_for_thread(
     path: &std::path::Path,
     thread_id: codex_protocol::ThreadId,
 ) -> bool {
-    if !tokio::fs::try_exists(path).await.unwrap_or(false) {
+    if codex_rollout::existing_rollout_path(path).await.is_none() {
         return false;
     }
     // SQLite metadata can outlive a moved/recreated rollout path. When history is
@@ -101,7 +101,7 @@ pub(super) async fn read_thread_by_rollout_path(
     include_archived: bool,
     include_history: bool,
 ) -> ThreadStoreResult<StoredThread> {
-    let path = resolve_requested_rollout_path(store, rollout_path)?;
+    let path = resolve_requested_rollout_path(store, rollout_path).await?;
     let mut thread = read_thread_from_rollout_path(store, path).await?;
     if !include_archived && thread.archived_at.is_some() {
         return Err(ThreadStoreError::InvalidRequest {
@@ -128,7 +128,7 @@ pub(super) async fn read_thread_by_rollout_path(
     Ok(thread)
 }
 
-fn resolve_requested_rollout_path(
+async fn resolve_requested_rollout_path(
     store: &LocalThreadStore,
     rollout_path: std::path::PathBuf,
 ) -> ThreadStoreResult<std::path::PathBuf> {
@@ -137,7 +137,15 @@ fn resolve_requested_rollout_path(
     } else {
         rollout_path
     };
-    std::fs::canonicalize(&path).map_err(|err| ThreadStoreError::InvalidRequest {
+    let Some(path) = codex_rollout::existing_rollout_path(path.as_path()).await else {
+        return Err(ThreadStoreError::InvalidRequest {
+            message: format!(
+                "failed to resolve rollout path `{}`: file does not exist",
+                path.display()
+            ),
+        });
+    };
+    std::fs::canonicalize(path.as_path()).map_err(|err| ThreadStoreError::InvalidRequest {
         message: format!("failed to resolve rollout path `{}`: {err}", path.display()),
     })
 }
@@ -166,11 +174,9 @@ async fn resolve_rollout_path(
     include_archived: bool,
 ) -> ThreadStoreResult<Option<std::path::PathBuf>> {
     if let Ok(path) = live_writer::rollout_path(store, thread_id).await
-        && tokio::fs::try_exists(path.as_path()).await.map_err(|err| {
-            ThreadStoreError::InvalidRequest {
-                message: format!("failed to check rollout path for thread id {thread_id}: {err}"),
-            }
-        })?
+        && codex_rollout::existing_rollout_path(path.as_path())
+            .await
+            .is_some()
         && (include_archived || !rollout_path_is_archived(store.config.codex_home.as_path(), &path))
     {
         return Ok(Some(path));
@@ -280,6 +286,9 @@ async fn stored_thread_from_sqlite_metadata(
         .await
         .ok()
         .map(|meta_line| meta_line.meta);
+    let rollout_path = codex_rollout::existing_rollout_path(metadata.rollout_path.as_path())
+        .await
+        .unwrap_or_else(|| metadata.rollout_path.clone());
     let forked_from_id = session_meta.as_ref().and_then(|meta| meta.forked_from_id);
     let preview = metadata
         .preview
@@ -288,7 +297,7 @@ async fn stored_thread_from_sqlite_metadata(
         .unwrap_or_default();
     StoredThread {
         thread_id: metadata.id,
-        rollout_path: Some(metadata.rollout_path),
+        rollout_path: Some(rollout_path),
         forked_from_id,
         preview,
         name,

codex-rs/thread-store/src/local/update_thread_metadata.rs

@@ -68,13 +68,14 @@ pub(super) async fn update_thread_metadata(
     if live_writer::rollout_path(store, thread_id).await.is_ok() {
         live_writer::persist_thread(store, thread_id).await?;
     }
-    let resolved_rollout_path =
+    let mut resolved_rollout_path =
         resolve_rollout_path(store, thread_id, params.include_archived).await?;
     let name = patch.name;
     let git_info = patch.git_info;
     if let Some(memory_mode) = patch.memory_mode {
         apply_thread_memory_mode(resolved_rollout_path.path.as_path(), thread_id, memory_mode)
             .await?;
+        refresh_resolved_rollout_path(&mut resolved_rollout_path).await;
     }
 
     let state_db_ctx = store.state_db().await;
@@ -142,6 +143,7 @@ pub(super) async fn update_thread_metadata(
             memory_mode.as_deref(),
         )
         .await?;
+        refresh_resolved_rollout_path(&mut resolved_rollout_path).await;
         apply_thread_git_info(store, thread_id, sha, branch, origin_url).await?;
     }
 
@@ -172,6 +174,12 @@ pub(super) async fn update_thread_metadata(
     Ok(thread)
 }
 
+async fn refresh_resolved_rollout_path(resolved: &mut ResolvedRolloutPath) {
+    if let Some(path) = codex_rollout::existing_rollout_path(resolved.path.as_path()).await {
+        resolved.path = path;
+    }
+}
+
 async fn apply_metadata_update(
     store: &LocalThreadStore,
     thread_id: ThreadId,