Codex Handbook
thread-store/src/local/delete_thread.rs 210 lines
//! Local hard-delete support for persisted threads.//!//! Existing rollout files are deleted before this operation reports success. A rollout file that//! vanishes after discovery counts as already deleted. SQLite cleanup happens at the app-server//! layer after every associated rollout has been removed so failed deletes can be retried.use std::io::ErrorKind;use std::path::Path;use codex_rollout::ARCHIVED_SESSIONS_SUBDIR;use codex_rollout::SESSIONS_SUBDIR;use codex_rollout::find_archived_thread_path_by_id_str;use codex_rollout::find_thread_path_by_id_str;use codex_rollout::remove_thread_name_entries;use super::LocalThreadStore;use super::helpers::matching_rollout_file_name;use super::helpers::scoped_rollout_path;use crate::DeleteThreadParams;use crate::ThreadStoreError;use crate::ThreadStoreResult;pub(super) async fn delete_thread(    store: &LocalThreadStore,    params: DeleteThreadParams,) -> ThreadStoreResult<()> {    let thread_id = params.thread_id;    let thread_id_str = thread_id.to_string();    let state_db_ctx = store.state_db().await;    let mut rollout_paths = Vec::new();    match find_thread_path_by_id_str(        store.config.codex_home.as_path(),        thread_id_str.as_str(),        state_db_ctx.as_deref(),    )    .await    {        Ok(Some(path)) => rollout_paths.push(path),        Ok(None) => {}        Err(err) => {            return Err(ThreadStoreError::InvalidRequest {                message: format!("failed to locate thread id {thread_id}: {err}"),            });        }    }    match find_archived_thread_path_by_id_str(        store.config.codex_home.as_path(),        thread_id_str.as_str(),        state_db_ctx.as_deref(),    )    .await    {        Ok(Some(path)) => {            if !rollout_paths.contains(&path) {                rollout_paths.push(path);            }        }        Ok(None) => {}        Err(err) => {            return Err(ThreadStoreError::InvalidRequest {                message: format!("failed to locate archived thread id {thread_id}: {err}"),            });        }    }    let found_rollout_path = !rollout_paths.is_empty();    for rollout_path in rollout_paths {        delete_rollout_file(store, rollout_path.as_path(), thread_id)?;    }    remove_thread_name_entries(store.config.codex_home.as_path(), thread_id)        .await        .map_err(|err| ThreadStoreError::Internal {            message: format!("failed to delete thread name index entries for {thread_id}: {err}"),        })?;    if !found_rollout_path {        return Err(ThreadStoreError::ThreadNotFound { thread_id });    }    store.live_recorders.lock().await.remove(&thread_id);    Ok(())}fn delete_rollout_file(    store: &LocalThreadStore,    rollout_path: &Path,    thread_id: codex_protocol::ThreadId,) -> ThreadStoreResult<bool> {    let plain_path = codex_rollout::plain_rollout_path(rollout_path);    let compressed_path = plain_path.with_extension("jsonl.zst");    let deleted_plain = delete_rollout_path(store, plain_path.as_path(), thread_id)?;    let deleted_compressed = delete_rollout_path(store, compressed_path.as_path(), thread_id)?;    Ok(deleted_plain || deleted_compressed)}fn delete_rollout_path(    store: &LocalThreadStore,    rollout_path: &Path,    thread_id: codex_protocol::ThreadId,) -> ThreadStoreResult<bool> {    let canonical_rollout_path = scoped_rollout_path(        store.config.codex_home.join(SESSIONS_SUBDIR),        rollout_path,        "sessions",    )    .or_else(|_| {        scoped_rollout_path(            store.config.codex_home.join(ARCHIVED_SESSIONS_SUBDIR),            rollout_path,            "archived sessions",        )    })    .or_else(|err| match rollout_path.try_exists() {        Ok(false) => Ok(rollout_path.to_path_buf()),        Ok(true) | Err(_) => Err(err),    })?;    matching_rollout_file_name(&canonical_rollout_path, thread_id, rollout_path)?;    match std::fs::remove_file(&canonical_rollout_path) {        Ok(()) => Ok(true),        Err(err) if err.kind() == ErrorKind::NotFound => Ok(false),        Err(err) => Err(ThreadStoreError::Internal {            message: format!(                "failed to delete rollout file `{}`: {err}",                canonical_rollout_path.display()            ),        }),    }}#[cfg(test)]mod tests {    use codex_protocol::ThreadId;    use pretty_assertions::assert_eq;    use tempfile::TempDir;    use uuid::Uuid;    use super::*;    use crate::ThreadStore;    use crate::local::LocalThreadStore;    use crate::local::test_support::test_config;    use crate::local::test_support::write_archived_session_file;    use crate::local::test_support::write_session_file;    #[tokio::test]    async fn delete_thread_removes_active_and_archived_rollouts() {        let home = TempDir::new().expect("temp dir");        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);        let active_path =            write_session_file(home.path(), "2025-01-03T12-00-00", Uuid::from_u128(301))                .expect("session file");        let compressed_path = active_path.with_extension("jsonl.zst");        std::fs::write(&compressed_path, b"compressed sibling").expect("compressed sibling");        let cases = [            (Uuid::from_u128(301), active_path),            (                Uuid::from_u128(302),                write_archived_session_file(                    home.path(),                    "2025-01-03T12-00-00",                    Uuid::from_u128(302),                )                .expect("archived session file"),            ),        ];        for (uuid, path) in cases {            let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");            store                .delete_thread(DeleteThreadParams { thread_id })                .await                .expect("delete thread");            assert!(!path.exists());        }        assert!(!compressed_path.exists());    }    #[tokio::test]    async fn delete_rollout_file_treats_vanished_path_as_already_deleted() {        let home = TempDir::new().expect("temp dir");        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);        let uuid = Uuid::from_u128(305);        let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");        let path =            write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");        std::fs::remove_file(&path).expect("remove session file");        assert!(!delete_rollout_file(&store, path.as_path(), thread_id).expect("delete rollout"));    }    #[tokio::test]    async fn delete_thread_reports_missing_thread() {        let home = TempDir::new().expect("temp dir");        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);        let thread_id =            ThreadId::from_string("00000000-0000-0000-0000-000000000304").expect("valid thread id");        let err = store            .delete_thread(DeleteThreadParams { thread_id })            .await            .expect_err("missing thread should fail");        assert_eq!(            err.to_string(),            "thread 00000000-0000-0000-0000-000000000304 not found"        );    }}