diff --git a/codex-rs/app-server/src/request_processors/thread_delete.rs b/codex-rs/app-server/src/request_processors/thread_delete.rs index 06fb2e2c30..7b014daa21 100644 --- a/codex-rs/app-server/src/request_processors/thread_delete.rs +++ b/codex-rs/app-server/src/request_processors/thread_delete.rs @@ -54,17 +54,6 @@ impl ThreadRequestProcessor { .await .map_err(thread_store_delete_error)?; - if let Some(state_db) = self.state_db.as_ref() { - state_db - .delete_threads_strict(thread_ids.as_slice()) - .await - .map_err(|err| { - internal_error(format!( - "failed to delete app-server state for {thread_id}: {err}" - )) - })?; - } - deleted_thread_ids.extend( delete_order .into_iter() diff --git a/codex-rs/app-server/tests/suite/v2/thread_delete.rs b/codex-rs/app-server/tests/suite/v2/thread_delete.rs index 7f3741e8b1..ceebbbf3db 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_delete.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_delete.rs @@ -149,6 +149,7 @@ async fn thread_delete_deletes_spawned_descendants() -> Result<()> { assert_eq!(deleted_ids, vec![grandchild_id, child_id, parent_id]); for thread_id in [parent_thread_id, child_thread_id, grandchild_thread_id] { + assert_eq!(state_db.get_thread(thread_id).await?, None); let rollout_path = find_thread_path_by_id_str( codex_home.path(), &thread_id.to_string(), diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 56b4df5f94..1ed26b95e4 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -422,7 +422,9 @@ pub fn thread_store_from_config( } store } - ThreadStoreConfig::InMemory { id } => InMemoryThreadStore::for_id(id), + ThreadStoreConfig::InMemory { id } => { + Arc::new(InMemoryThreadStore::for_id(id).with_state_db(state_db)) + } } } diff --git a/codex-rs/thread-store/src/in_memory.rs b/codex-rs/thread-store/src/in_memory.rs index 924a0fe788..955f874156 100644 --- a/codex-rs/thread-store/src/in_memory.rs +++ b/codex-rs/thread-store/src/in_memory.rs @@ -67,6 +67,42 @@ mod tests { use codex_protocol::models::BaseInstructions; use codex_protocol::protocol::SessionSource; + #[tokio::test] + async fn deletion_cleans_associated_sqlite_and_shared_memory() + -> Result<(), Box> { + use codex_utils_absolute_path::test_support::PathExt; + use pretty_assertions::assert_eq; + + let home = tempfile::TempDir::new()?; + let state_db = codex_state::StateRuntime::init( + codex_state::SqliteConfig::new_for_testing(home.path().abs()), + "test".to_string(), + ) + .await?; + let shared = InMemoryThreadStore::default(); + let store = shared.with_state_db(Some(state_db.clone())); + let thread_id = ThreadId::new(); + shared + .create_thread(create_thread_params(thread_id, ThreadHistoryMode::Legacy)) + .await?; + let metadata = codex_state::ThreadMetadataBuilder::new( + thread_id, + home.path().join("thread.jsonl"), + Utc::now(), + SessionSource::Cli, + ) + .build("test"); + state_db.upsert_thread(&metadata).await?; + + store + .delete_thread(DeleteThreadParams { thread_id }) + .await?; + + assert_eq!(state_db.get_thread(thread_id).await?, None); + assert!(!shared.state.lock().await.histories.contains_key(&thread_id)); + Ok(()) + } + #[tokio::test] async fn default_turn_pagination_methods_return_unsupported() { let store = InMemoryThreadStore::default(); @@ -498,8 +534,9 @@ pub struct InMemoryThreadStoreCalls { /// service. #[derive(Default)] pub struct InMemoryThreadStore { - state: tokio::sync::Mutex, - omit_metadata_update_result: AtomicBool, + state: Arc>, + omit_metadata_update_result: Arc, + state_db: Option, } #[derive(Default)] @@ -526,6 +563,15 @@ impl InMemoryThreadStore { .clone() } + /// Shares this debug store's thread data while owning cleanup of the caller's SQLite state. + pub fn with_state_db(&self, state_db: Option) -> Self { + Self { + state: Arc::clone(&self.state), + omit_metadata_update_result: Arc::clone(&self.omit_metadata_update_result), + state_db, + } + } + /// Removes a shared in-memory store for `id`. pub fn remove_id(id: &str) -> Option> { stores_guard().remove(id) @@ -830,8 +876,18 @@ impl InMemoryThreadStore { } async fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreResult<()> { + self.state.lock().await.calls.delete_thread += 1; + let deleted_state_rows = if let Some(state_db) = &self.state_db { + state_db + .delete_threads_strict(&[params.thread_id]) + .await + .map_err(|error| ThreadStoreError::Internal { + message: format!("failed to delete thread state: {error}"), + })? + } else { + 0 + }; let mut state = self.state.lock().await; - state.calls.delete_thread += 1; let existed = state.histories.remove(¶ms.thread_id).is_some(); state.created_threads.remove(¶ms.thread_id); state.names.remove(¶ms.thread_id); @@ -842,7 +898,7 @@ impl InMemoryThreadStore { state .rollout_paths .retain(|_, thread_id| *thread_id != params.thread_id); - if existed { + if existed || deleted_state_rows > 0 { Ok(()) } else { Err(ThreadStoreError::ThreadNotFound { diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index a8b6a36650..2650a56f85 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -11,13 +11,19 @@ mod local; mod projects; mod queue_store; mod store; +mod thread_attachments; mod thread_metadata_sync; mod thread_sections; mod types; +pub use codex_state::AddThreadAttachmentOutcome; pub use codex_state::MAX_QUEUE_ITEMS; +pub use codex_state::MAX_THREAD_ATTACHMENT_PAYLOAD_BYTES; pub use codex_state::ProjectSortKey; pub use codex_state::QueuedUserSubmissionRecord; +pub use codex_state::RemoveThreadAttachmentOutcome; +pub use codex_state::ThreadAttachment; +pub use codex_state::ThreadAttachmentPage; pub use error::ThreadStoreError; pub use error::ThreadStoreResult; pub use in_memory::InMemoryThreadStore; @@ -49,6 +55,9 @@ pub use queue_store::QueueStore; pub use store::PersistContext; pub use store::ThreadStore; pub use store::ThreadStoreFuture; +pub use thread_attachments::AddThreadAttachmentParams; +pub use thread_attachments::ListThreadAttachmentsParams; +pub use thread_attachments::RemoveThreadAttachmentParams; pub use thread_sections::CreateThreadSectionParams; pub use thread_sections::DeleteThreadSectionParams; pub use thread_sections::ListThreadSectionsParams; diff --git a/codex-rs/thread-store/src/local/delete_thread.rs b/codex-rs/thread-store/src/local/delete_thread.rs index f089ba6434..775344a783 100644 --- a/codex-rs/thread-store/src/local/delete_thread.rs +++ b/codex-rs/thread-store/src/local/delete_thread.rs @@ -1,8 +1,9 @@ //! 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. The app-server deletes main state DB rows -//! after every associated rollout is removed; this module deletes local history projection rows. +//! vanishes after discovery counts as already deleted. Main state DB rows are deleted after every +//! associated rollout is removed, under the same lifecycle lock, so queued artifact mutations cannot +//! use deleted thread metadata. use std::collections::HashMap; use std::collections::HashSet; @@ -72,7 +73,19 @@ pub(super) async fn delete_thread( let thread_rollouts = ThreadRollouts::from_index(&reference_index, thread_id); ensure_no_external_references(&reference_index, std::slice::from_ref(&thread_rollouts))?; let mut writer_guards = store.acquire_writer_locks(&[thread_id]).await?; - delete_thread_after_reference_check(store, thread_rollouts, &mut writer_guards).await + let found_rollout = + match delete_thread_after_reference_check(store, thread_rollouts, &mut writer_guards).await + { + Ok(()) => true, + Err(ThreadStoreError::ThreadNotFound { .. }) => false, + Err(err) => return Err(err), + }; + let deleted_state_rows = delete_state_rows(store, &[thread_id]).await?; + if found_rollout || deleted_state_rows > 0 { + Ok(()) + } else { + Err(ThreadStoreError::ThreadNotFound { thread_id }) + } } pub(super) async fn delete_threads( @@ -111,9 +124,26 @@ pub(super) async fn delete_threads( Err(err) => return Err(err), } } + // Retain the complete retry graph until every rollout has been removed. + delete_state_rows(store, &thread_ids).await?; Ok(()) } +async fn delete_state_rows( + store: &LocalThreadStore, + thread_ids: &[codex_protocol::ThreadId], +) -> ThreadStoreResult { + let Some(state_db) = store.state_db.as_ref() else { + return Ok(0); + }; + state_db + .delete_threads_strict(thread_ids) + .await + .map_err(|err| ThreadStoreError::Internal { + message: format!("failed to delete thread state: {err}"), + }) +} + fn ensure_no_external_references( reference_index: &RolloutReferenceIndex, thread_rollouts: &[ThreadRollouts], diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 0c2c36d413..37467db3e1 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -16,6 +16,7 @@ mod rollout_migration; #[allow(dead_code)] mod rollout_lineage; mod search_threads; +mod thread_attachments; mod thread_history; mod thread_history_materialization; mod thread_rollout_resolver; @@ -52,6 +53,8 @@ use tokio::sync::OwnedRwLockReadGuard; use tokio::sync::OwnedRwLockWriteGuard; use tokio::sync::RwLock; +use crate::AddThreadAttachmentOutcome; +use crate::AddThreadAttachmentParams; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::ArchiveThreadsParams; @@ -66,6 +69,7 @@ use crate::DeletedProject; use crate::ItemPage; use crate::ListItemsParams; use crate::ListProjectsParams; +use crate::ListThreadAttachmentsParams; use crate::ListThreadSectionsParams; use crate::ListThreadsParams; use crate::ListTimelineParams; @@ -79,6 +83,8 @@ use crate::PreparedFork; use crate::ProjectMoveOutcome; use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadParams; +use crate::RemoveThreadAttachmentOutcome; +use crate::RemoveThreadAttachmentParams; use crate::RenameThreadSectionParams; use crate::ResumeThreadParams; use crate::RevertThreadParams; @@ -91,6 +97,7 @@ use crate::StoredThread; use crate::StoredThreadHistory; use crate::StoredThreadSection; use crate::StoredThreadSectionsPage; +use crate::ThreadAttachmentPage; use crate::ThreadMetadataPatch; use crate::ThreadOccurrenceSearchPage; use crate::ThreadPage; @@ -580,6 +587,31 @@ impl ThreadStore for LocalThreadStore { Box::pin(async move { thread_sections::delete_thread_section(self, params).await }) } + fn supports_thread_attachments(&self) -> bool { + self.state_db.is_some() + } + + fn add_thread_attachment( + &self, + params: AddThreadAttachmentParams, + ) -> ThreadStoreFuture<'_, AddThreadAttachmentOutcome> { + Box::pin(async move { thread_attachments::add_thread_attachment(self, params).await }) + } + + fn list_thread_attachments( + &self, + params: ListThreadAttachmentsParams, + ) -> ThreadStoreFuture<'_, ThreadAttachmentPage> { + Box::pin(async move { thread_attachments::list_thread_attachments(self, params).await }) + } + + fn remove_thread_attachment( + &self, + params: RemoveThreadAttachmentParams, + ) -> ThreadStoreFuture<'_, RemoveThreadAttachmentOutcome> { + Box::pin(async move { thread_attachments::remove_thread_attachment(self, params).await }) + } + fn supports_projects(&self) -> bool { self.state_db.is_some() } diff --git a/codex-rs/thread-store/src/local/thread_attachments.rs b/codex-rs/thread-store/src/local/thread_attachments.rs new file mode 100644 index 0000000000..95cea009c4 --- /dev/null +++ b/codex-rs/thread-store/src/local/thread_attachments.rs @@ -0,0 +1,98 @@ +//! Local attachment adapter sharing lifecycle exclusion with thread deletion. + +use super::LocalThreadStore; +use crate::AddThreadAttachmentOutcome; +use crate::AddThreadAttachmentParams; +use crate::ListThreadAttachmentsParams; +use crate::RemoveThreadAttachmentOutcome; +use crate::RemoveThreadAttachmentParams; +use crate::ThreadAttachmentPage; +use crate::ThreadStoreError; +use crate::ThreadStoreResult; +use codex_protocol::ThreadId; +use codex_rollout::StateDbHandle; + +pub(super) async fn add_thread_attachment( + store: &LocalThreadStore, + params: AddThreadAttachmentParams, +) -> ThreadStoreResult { + let state_db = state_db(store, "thread/attachment/add")?; + let _lifecycle_reservation = store + .live_writer_locks + .reserve_lifecycle(params.thread_id) + .await; + state_db + .add_thread_attachment( + params.thread_id, + ¶ms.attachment_type, + ¶ms.identity_key, + ¶ms.payload, + ) + .await + .map_err(|error| attachment_error("add", Some(params.thread_id), error)) +} + +pub(super) async fn list_thread_attachments( + store: &LocalThreadStore, + params: ListThreadAttachmentsParams, +) -> ThreadStoreResult { + let state_db = state_db(store, "thread/attachment/list")?; + state_db + .list_thread_attachments(params.thread_id, params.cursor.as_deref(), params.limit) + .await + .map_err(|error| attachment_error("list", /*thread_id*/ None, error)) +} + +pub(super) async fn remove_thread_attachment( + store: &LocalThreadStore, + params: RemoveThreadAttachmentParams, +) -> ThreadStoreResult { + let state_db = state_db(store, "thread/attachment/remove")?; + let _lifecycle_reservation = store + .live_writer_locks + .reserve_lifecycle(params.thread_id) + .await; + state_db + .remove_thread_attachment( + params.thread_id, + ¶ms.attachment_type, + ¶ms.identity_key, + ) + .await + .map_err(|error| attachment_error("remove", Some(params.thread_id), error)) +} + +fn state_db<'store>( + store: &'store LocalThreadStore, + operation: &'static str, +) -> ThreadStoreResult<&'store StateDbHandle> { + store + .state_db + .as_ref() + .ok_or(ThreadStoreError::Unsupported { operation }) +} + +fn attachment_error( + operation: &str, + thread_id: Option, + error: impl std::fmt::Display, +) -> ThreadStoreError { + let message = error.to_string(); + if let Some(message) = message.strip_prefix("invalid thread attachment request: ") { + return ThreadStoreError::InvalidRequest { + message: message.to_string(), + }; + } + if message.starts_with("thread not found: ") + && let Some(thread_id) = thread_id + { + return ThreadStoreError::ThreadNotFound { thread_id }; + } + ThreadStoreError::Internal { + message: format!("failed to {operation} thread attachment: {message}"), + } +} + +#[cfg(test)] +#[path = "thread_attachments_tests.rs"] +mod tests; diff --git a/codex-rs/thread-store/src/local/thread_attachments_tests.rs b/codex-rs/thread-store/src/local/thread_attachments_tests.rs new file mode 100644 index 0000000000..936447a73d --- /dev/null +++ b/codex-rs/thread-store/src/local/thread_attachments_tests.rs @@ -0,0 +1,509 @@ +//! Coverage for unloaded attachment operations and thread deletion exclusion. + +use super::LocalThreadStore; +use crate::AddThreadAttachmentOutcome; +use crate::AddThreadAttachmentParams; +use crate::DeleteThreadParams; +use crate::DeleteThreadsParams; +use crate::InMemoryThreadStore; +use crate::ListThreadAttachmentsParams; +use crate::RemoveThreadAttachmentOutcome; +use crate::RemoveThreadAttachmentParams; +use crate::ThreadAttachmentPage; +use crate::ThreadStore; +use crate::ThreadStoreError; +use crate::local::test_support::test_config; +use crate::local::test_support::write_session_file; +use chrono::Utc; +use codex_protocol::ThreadId; +use codex_protocol::protocol::SessionSource; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::path::Path; +use std::path::PathBuf; +use uuid::Uuid; + +#[tokio::test] +async fn thread_attachments_require_a_supported_sqlite_store() { + let thread_id = ThreadId::new(); + let default_store = InMemoryThreadStore::default(); + assert!(!default_store.supports_thread_attachments()); + assert!(matches!( + default_store + .add_thread_attachment(AddThreadAttachmentParams { + thread_id, + attachment_type: "pull_request".to_string(), + identity_key: "openai/codex#1".to_string(), + payload: json!({"url": "https://github.com/openai/codex/pull/1"}), + }) + .await, + Err(ThreadStoreError::Unsupported { + operation: "thread/attachment/add" + }) + )); + + let home = tempfile::TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + assert!(!store.supports_thread_attachments()); + assert!(matches!( + store + .list_thread_attachments(ListThreadAttachmentsParams { + thread_id, + cursor: None, + limit: 10, + }) + .await, + Err(ThreadStoreError::Unsupported { + operation: "thread/attachment/list" + }) + )); + assert!(matches!( + store + .remove_thread_attachment(RemoveThreadAttachmentParams { + thread_id, + attachment_type: "pull_request".to_string(), + identity_key: "openai/codex#1".to_string(), + }) + .await, + Err(ThreadStoreError::Unsupported { + operation: "thread/attachment/remove" + }) + )); +} + +#[tokio::test] +async fn unloaded_thread_attachments_support_listing_and_removal() { + let home = tempfile::TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let runtime = codex_state::StateRuntime::init( + config.sqlite.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config.clone(), Some(runtime.clone())); + assert!(store.supports_thread_attachments()); + + let first_thread_id = ThreadId::new(); + let second_thread_id = ThreadId::new(); + for thread_id in [first_thread_id, second_thread_id] { + let metadata = codex_state::ThreadMetadataBuilder::new( + thread_id, + home.path().join(format!("{thread_id}.jsonl")), + Utc::now(), + SessionSource::Cli, + ) + .build(&config.default_model_provider_id); + runtime + .upsert_thread(&metadata) + .await + .expect("persist unloaded thread metadata"); + } + + let params = AddThreadAttachmentParams { + thread_id: first_thread_id, + attachment_type: "pull_request".to_string(), + identity_key: "openai/codex#1".to_string(), + payload: json!({"url": "https://github.com/openai/codex/pull/1"}), + }; + let AddThreadAttachmentOutcome::Created(first_attachment) = store + .add_thread_attachment(params.clone()) + .await + .expect("attach attachment to unloaded persisted thread") + else { + panic!("first explicit attachment should create an attachment"); + }; + assert_eq!( + store + .add_thread_attachment(params.clone()) + .await + .expect("repeat attachment attachment"), + AddThreadAttachmentOutcome::Existing(first_attachment.clone()) + ); + + let second_params = AddThreadAttachmentParams { + thread_id: first_thread_id, + identity_key: "openai/codex#2".to_string(), + payload: json!({"url": "https://github.com/openai/codex/pull/2"}), + ..params.clone() + }; + let AddThreadAttachmentOutcome::Created(second_attachment) = store + .add_thread_attachment(second_params) + .await + .expect("attach second thread attachment") + else { + panic!("second thread attachment should create an attachment"); + }; + + let first_page = store + .list_thread_attachments(ListThreadAttachmentsParams { + thread_id: first_thread_id, + cursor: None, + limit: 1, + }) + .await + .expect("list first page"); + assert_eq!(first_page.attachments.len(), 1); + let second_page = store + .list_thread_attachments(ListThreadAttachmentsParams { + thread_id: first_thread_id, + cursor: first_page.next_cursor, + limit: 1, + }) + .await + .expect("list second page"); + assert_eq!(second_page.attachments.len(), 1); + assert_eq!(second_page.next_cursor, None); + let listed_ids = [ + first_page.attachments[0].id.clone(), + second_page.attachments[0].id.clone(), + ]; + assert!(listed_ids.contains(&first_attachment.id)); + assert!(listed_ids.contains(&second_attachment.id)); + + let removal_params = RemoveThreadAttachmentParams { + thread_id: first_thread_id, + attachment_type: params.attachment_type.clone(), + identity_key: params.identity_key.clone(), + }; + assert_eq!( + store + .remove_thread_attachment(removal_params.clone()) + .await + .expect("remove existing attachment"), + RemoveThreadAttachmentOutcome::Removed(first_attachment) + ); + assert_eq!( + store + .remove_thread_attachment(removal_params) + .await + .expect("repeat attachment removal"), + RemoveThreadAttachmentOutcome::NotFound + ); + assert!(matches!( + store + .add_thread_attachment(params) + .await + .expect("reattach removed attachment"), + AddThreadAttachmentOutcome::Created(_) + )); +} + +#[tokio::test] +async fn attachment_state_errors_preserve_missing_thread_and_invalid_request_categories() { + let home = tempfile::TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let runtime = codex_state::StateRuntime::init( + config.sqlite.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config, Some(runtime)); + let thread_id = ThreadId::new(); + + assert!(matches!( + store + .add_thread_attachment(AddThreadAttachmentParams { + thread_id, + attachment_type: "pull_request".to_string(), + identity_key: "openai/codex#1".to_string(), + payload: json!({}), + }) + .await, + Err(ThreadStoreError::ThreadNotFound { thread_id: missing }) if missing == thread_id + )); + assert!(matches!( + store + .add_thread_attachment(AddThreadAttachmentParams { + thread_id, + attachment_type: " ".to_string(), + identity_key: "openai/codex#1".to_string(), + payload: json!({}), + }) + .await, + Err(ThreadStoreError::InvalidRequest { message }) + if message == "attachment type must not be empty" + )); + assert!(matches!( + store + .list_thread_attachments(ListThreadAttachmentsParams { + thread_id, + cursor: None, + limit: 0, + }) + .await, + Err(ThreadStoreError::InvalidRequest { message }) + if message == "page limit must be between 1 and 100" + )); +} + +#[tokio::test] +async fn attachment_mutations_wait_for_exclusive_thread_lifecycle_operations() { + let home = tempfile::TempDir::new().expect("temp dir"); + let config = test_config(home.path()); + let runtime = codex_state::StateRuntime::init( + config.sqlite.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config.clone(), Some(runtime.clone())); + let thread_id = ThreadId::new(); + let metadata = codex_state::ThreadMetadataBuilder::new( + thread_id, + home.path().join(format!("{thread_id}.jsonl")), + Utc::now(), + SessionSource::Cli, + ) + .build(&config.default_model_provider_id); + runtime + .upsert_thread(&metadata) + .await + .expect("persist attachment owner"); + + let lifecycle_guard = store.live_writer_locks.lock_lifecycle(thread_id).await; + let mut attachment = Box::pin(store.add_thread_attachment(AddThreadAttachmentParams { + thread_id, + attachment_type: "pull_request".to_string(), + identity_key: "openai/codex#1".to_string(), + payload: json!({"url": "https://github.com/openai/codex/pull/1"}), + })); + tokio::select! { + biased; + result = &mut attachment => panic!("attachment attached during lifecycle mutation: {result:?}"), + _ = tokio::task::yield_now() => {} + } + drop(lifecycle_guard); + assert!(matches!( + attachment.await, + Ok(AddThreadAttachmentOutcome::Created(_)) + )); + + let lifecycle_guard = store.live_writer_locks.lock_lifecycle(thread_id).await; + let mut removal = Box::pin( + store.remove_thread_attachment(RemoveThreadAttachmentParams { + thread_id, + attachment_type: "pull_request".to_string(), + identity_key: "openai/codex#1".to_string(), + }), + ); + tokio::select! { + biased; + result = &mut removal => panic!("attachment removed during lifecycle mutation: {result:?}"), + _ = tokio::task::yield_now() => {} + } + drop(lifecycle_guard); + assert!(matches!( + removal.await, + Ok(RemoveThreadAttachmentOutcome::Removed(_)) + )); +} + +#[tokio::test] +async fn attachment_mutations_queued_behind_deletion_reject_deleted_threads() { + enum DeleteMode { + Single, + Batch, + } + + for mode in [DeleteMode::Single, DeleteMode::Batch] { + let home = tempfile::TempDir::new().expect("temp dir"); + let thread_id = ThreadId::new(); + let thread_ids = match mode { + DeleteMode::Single => vec![thread_id], + DeleteMode::Batch => vec![thread_id, ThreadId::new()], + }; + let (store, _) = store_with_attachment_owners(home.path(), &thread_ids).await; + let runtime = store.state_db().await.expect("state db"); + + let reservation = store.live_writer_locks.reserve_lifecycle(thread_id).await; + let mut deletion = match mode { + DeleteMode::Single => store.delete_thread(DeleteThreadParams { thread_id }), + DeleteMode::Batch => store.delete_threads(DeleteThreadsParams { + thread_ids: thread_ids.clone(), + }), + }; + assert!(futures::poll!(&mut deletion).is_pending()); + + // The queued exclusive delete must precede subsequent shared attachment reservations. + let mut attachment = store.add_thread_attachment(AddThreadAttachmentParams { + thread_id, + attachment_type: "pull_request".to_string(), + identity_key: "openai/codex#2".to_string(), + payload: json!({"url": "https://github.com/openai/codex/pull/2"}), + }); + let mut removal = store.remove_thread_attachment(RemoveThreadAttachmentParams { + thread_id, + attachment_type: "pull_request".to_string(), + identity_key: "openai/codex#1".to_string(), + }); + assert!(futures::poll!(&mut attachment).is_pending()); + assert!(futures::poll!(&mut removal).is_pending()); + drop(reservation); + + let (deletion, attachment, removal) = tokio::join!(deletion, attachment, removal); + deletion.expect("delete attachment owners"); + assert!( + matches!(attachment, Err(ThreadStoreError::ThreadNotFound { thread_id: missing }) if missing == thread_id), + "attachment resumed after deletion: {attachment:?}" + ); + assert!( + matches!(removal, Err(ThreadStoreError::ThreadNotFound { thread_id: missing }) if missing == thread_id), + "removal resumed after deletion: {removal:?}" + ); + for &owner_id in &thread_ids { + assert_eq!( + runtime + .get_thread(owner_id) + .await + .expect("read deleted owner"), + None + ); + } + for &owner_id in &thread_ids { + assert_eq!( + store + .list_thread_attachments(ListThreadAttachmentsParams { + thread_id: owner_id, + cursor: None, + limit: 10, + }) + .await + .expect("list attachments after deletion"), + ThreadAttachmentPage { + attachments: Vec::new(), + next_cursor: None, + } + ); + } + } +} + +#[tokio::test] +async fn attachment_owner_deletion_can_retry_after_state_cleanup_fails() { + let home = tempfile::TempDir::new().expect("temp dir"); + let parent_id = ThreadId::new(); + let child_id = ThreadId::new(); + let (store, rollout_paths) = + store_with_attachment_owners(home.path(), &[parent_id, child_id]).await; + let runtime = store.state_db().await.expect("state db"); + runtime + .upsert_thread_spawn_edge( + parent_id, + child_id, + codex_state::DirectionalThreadSpawnEdgeStatus::Closed, + ) + .await + .expect("persist child relation for deletion retry"); + let pool = runtime + .sqlite() + .open_read_write_pool(&runtime.sqlite().state_db_path()) + .await + .expect("open state db for failure injection"); + sqlx::query( + "CREATE TRIGGER fail_thread_delete BEFORE DELETE ON threads BEGIN SELECT RAISE(ABORT, 'state cleanup failed'); END", + ) + .execute(&pool) + .await + .expect("inject state cleanup failure"); + let params = DeleteThreadsParams { + thread_ids: vec![child_id, parent_id], + }; + let error = store + .delete_threads(params.clone()) + .await + .expect_err("state cleanup failure must fail the delete"); + assert!(matches!(error, ThreadStoreError::Internal { .. })); + assert!(rollout_paths.iter().all(|path| !path.exists())); + assert_eq!( + runtime + .list_thread_spawn_descendants(parent_id) + .await + .expect("read child relation after failed deletion"), + vec![child_id] + ); + + sqlx::query("DROP TRIGGER fail_thread_delete") + .execute(&pool) + .await + .expect("restore state cleanup"); + store + .delete_thread(DeleteThreadParams { + thread_id: child_id, + }) + .await + .expect("single-thread retry must remove the owner even without a rollout"); + store + .delete_threads(params.clone()) + .await + .expect("batch retry must tolerate missing rollouts and already-deleted members"); + for &owner_id in ¶ms.thread_ids { + assert_eq!( + store + .list_thread_attachments(ListThreadAttachmentsParams { + thread_id: owner_id, + cursor: None, + limit: 10, + }) + .await + .expect("list attachments after deletion retry"), + ThreadAttachmentPage { + attachments: Vec::new(), + next_cursor: None, + } + ); + } + assert_eq!( + runtime + .delete_threads_strict(¶ms.thread_ids) + .await + .expect("repeated app-server state cleanup is idempotent"), + 0 + ); +} + +async fn store_with_attachment_owners( + home: &Path, + thread_ids: &[ThreadId], +) -> (LocalThreadStore, Vec) { + let config = test_config(home); + let runtime = codex_state::StateRuntime::init( + config.sqlite.clone(), + config.default_model_provider_id.clone(), + ) + .await + .expect("state db should initialize"); + let store = LocalThreadStore::new(config.clone(), Some(runtime.clone())); + let mut rollout_paths = Vec::new(); + for &thread_id in thread_ids { + let rollout_path = write_session_file( + home, + "2025-01-03T12-00-00", + Uuid::parse_str(&thread_id.to_string()).expect("thread UUID"), + ) + .expect("session file"); + let metadata = codex_state::ThreadMetadataBuilder::new( + thread_id, + rollout_path.clone(), + Utc::now(), + SessionSource::Cli, + ) + .build(&config.default_model_provider_id); + runtime + .upsert_thread(&metadata) + .await + .expect("persist attachment owner"); + store + .add_thread_attachment(AddThreadAttachmentParams { + thread_id, + attachment_type: "pull_request".to_string(), + identity_key: "openai/codex#1".to_string(), + payload: json!({"url": "https://github.com/openai/codex/pull/1"}), + }) + .await + .expect("attach attachment before deleting its owner"); + rollout_paths.push(rollout_path); + } + (store, rollout_paths) +} diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index 1e7c0637a8..64672c43f7 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -4,6 +4,8 @@ use std::any::Any; use std::future::Future; use std::pin::Pin; +use crate::AddThreadAttachmentOutcome; +use crate::AddThreadAttachmentParams; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::ArchiveThreadsParams; @@ -18,6 +20,7 @@ use crate::DeletedProject; use crate::ItemPage; use crate::ListItemsParams; use crate::ListProjectsParams; +use crate::ListThreadAttachmentsParams; use crate::ListThreadSectionsParams; use crate::ListThreadsParams; use crate::ListTurnsParams; @@ -29,6 +32,8 @@ use crate::PreparedFork; use crate::ProjectMoveOutcome; use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadParams; +use crate::RemoveThreadAttachmentOutcome; +use crate::RemoveThreadAttachmentParams; use crate::RenameThreadSectionParams; use crate::ResumeThreadParams; use crate::RevertThreadParams; @@ -41,6 +46,7 @@ use crate::StoredThread; use crate::StoredThreadHistory; use crate::StoredThreadSection; use crate::StoredThreadSectionsPage; +use crate::ThreadAttachmentPage; use crate::ThreadMetadataPatch; use crate::ThreadOccurrenceSearchPage; use crate::ThreadPage; @@ -251,6 +257,47 @@ pub trait ThreadStore: Any + Send + Sync { }) } + /// Whether this store can persist and discover thread-owned attachments. + fn supports_thread_attachments(&self) -> bool { + false + } + + /// Attaches an attachment, returning an existing attachment for repeated requests. + fn add_thread_attachment( + &self, + _params: AddThreadAttachmentParams, + ) -> ThreadStoreFuture<'_, AddThreadAttachmentOutcome> { + Box::pin(async { + Err(ThreadStoreError::Unsupported { + operation: "thread/attachment/add", + }) + }) + } + + /// Lists attachments belonging to one persisted thread. + fn list_thread_attachments( + &self, + _params: ListThreadAttachmentsParams, + ) -> ThreadStoreFuture<'_, ThreadAttachmentPage> { + Box::pin(async { + Err(ThreadStoreError::Unsupported { + operation: "thread/attachment/list", + }) + }) + } + + /// Removes an attachment and reports whether it existed. + fn remove_thread_attachment( + &self, + _params: RemoveThreadAttachmentParams, + ) -> ThreadStoreFuture<'_, RemoveThreadAttachmentOutcome> { + Box::pin(async { + Err(ThreadStoreError::Unsupported { + operation: "thread/attachment/remove", + }) + }) + } + /// Whether this store supports durable host-owned projects. fn supports_projects(&self) -> bool { false @@ -427,9 +474,11 @@ pub trait ThreadStore: Any + Send + Sync { fn unarchive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, StoredThread>; /// Deletes a thread's persisted rollout data and associated metadata. + /// Success includes cleanup of associated persisted state; callers must not repeat it. fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreFuture<'_, ()>; - /// Deletes threads in order, treating already-missing members as deleted. + /// Deletes threads and their associated persisted state in order, treating already-missing + /// members as deleted. /// /// Stores with request-scoped delete preflight should override this instead of repeating /// that work through [`ThreadStore::delete_thread`]. diff --git a/codex-rs/thread-store/src/thread_attachments.rs b/codex-rs/thread-store/src/thread_attachments.rs new file mode 100644 index 0000000000..4abe2d17c2 --- /dev/null +++ b/codex-rs/thread-store/src/thread_attachments.rs @@ -0,0 +1,39 @@ +//! Typed parameters for attachment operations on stored threads. + +use codex_protocol::ThreadId; +use serde_json::Value; + +/// Parameters for attaching a thread-owned attachment. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct AddThreadAttachmentParams { + /// Thread that owns the attachment. + pub thread_id: ThreadId, + /// Client-defined attachment type. + pub attachment_type: String, + /// Stable attachment identity within its thread and type. + pub identity_key: String, + /// Bounded, client-defined attachment metadata. + pub payload: Value, +} + +/// Parameters for listing attachments owned by one thread. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ListThreadAttachmentsParams { + /// Thread whose attachments should be returned. + pub thread_id: ThreadId, + /// Opaque cursor returned by a previous attachment listing. + pub cursor: Option, + /// Maximum number of attachments to return. + pub limit: usize, +} + +/// Parameters for removing a thread-owned attachment. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RemoveThreadAttachmentParams { + /// Thread that owns the attachment. + pub thread_id: ThreadId, + /// Client-defined attachment type. + pub attachment_type: String, + /// Stable attachment identity within its thread and type. + pub identity_key: String, +}