Add thread attachment operations with coordinated deletion (#44350)

## Why

Attachment mutations must coordinate with thread deletion so queued requests cannot use metadata belonging to a deleted thread.

## What changed

- Add typed `ThreadStore` operations for idempotent attachment creation, paginated listing, and removal, with capability detection and unsupported defaults.
- Implement attachment operations for SQLite-backed `LocalThreadStore`, including unloaded threads, and preserve invalid-request and missing-thread errors.
- Coordinate attachment mutations with lifecycle locks and move SQLite cleanup into thread-store deletion. Remove state rows only after rollout deletion, preserving retries after cleanup failures.
- Give the in-memory store ownership of its associated SQLite cleanup and remove duplicate cleanup from the app-server.

## Testing

Add coverage for attachment pagination and idempotency, unsupported stores, error categories, lifecycle exclusion, mutations queued behind single and batch deletion, cleanup failure retries, and in-memory state cleanup. Extend the app-server descendant deletion test to verify SQLite rows are removed.

GitOrigin-RevId: f0157538c7c5c2ee1362ce40eac157101c5d4505
This commit is contained in:
joeytrasatti-openai authored and copyberry committed 2026-09-10 00:30:27 +00:00
1 parent e444aa99d7
commit 2df0b747ba
11 files changed
+834 -20

No files matched your search

@@ -54,17 +54,6 @@ impl ThreadRequestProcessor {
.await .await
.map_err(thread_store_delete_error)?; .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( deleted_thread_ids.extend(
delete_order delete_order
.into_iter() .into_iter()
@@ -149,6 +149,7 @@ async fn thread_delete_deletes_spawned_descendants() -> Result<()> {
assert_eq!(deleted_ids, vec![grandchild_id, child_id, parent_id]); assert_eq!(deleted_ids, vec![grandchild_id, child_id, parent_id]);
for thread_id in [parent_thread_id, child_thread_id, grandchild_thread_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( let rollout_path = find_thread_path_by_id_str(
codex_home.path(), codex_home.path(),
&thread_id.to_string(), &thread_id.to_string(),
+3 -1
View File
@@ -422,7 +422,9 @@ pub fn thread_store_from_config(
} }
store store
} }
ThreadStoreConfig::InMemory { id } => InMemoryThreadStore::for_id(id), ThreadStoreConfig::InMemory { id } => {
Arc::new(InMemoryThreadStore::for_id(id).with_state_db(state_db))
}
} }
} }
+60 -4
View File
@@ -67,6 +67,42 @@ mod tests {
use codex_protocol::models::BaseInstructions; use codex_protocol::models::BaseInstructions;
use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SessionSource;
#[tokio::test]
async fn deletion_cleans_associated_sqlite_and_shared_memory()
-> Result<(), Box<dyn std::error::Error>> {
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] #[tokio::test]
async fn default_turn_pagination_methods_return_unsupported() { async fn default_turn_pagination_methods_return_unsupported() {
let store = InMemoryThreadStore::default(); let store = InMemoryThreadStore::default();
@@ -498,8 +534,9 @@ pub struct InMemoryThreadStoreCalls {
/// service. /// service.
#[derive(Default)] #[derive(Default)]
pub struct InMemoryThreadStore { pub struct InMemoryThreadStore {
state: tokio::sync::Mutex<InMemoryThreadStoreState>, state: Arc<tokio::sync::Mutex<InMemoryThreadStoreState>>,
omit_metadata_update_result: AtomicBool, omit_metadata_update_result: Arc<AtomicBool>,
state_db: Option<codex_rollout::StateDbHandle>,
} }
#[derive(Default)] #[derive(Default)]
@@ -526,6 +563,15 @@ impl InMemoryThreadStore {
.clone() .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<codex_rollout::StateDbHandle>) -> 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`. /// Removes a shared in-memory store for `id`.
pub fn remove_id(id: &str) -> Option<Arc<Self>> { pub fn remove_id(id: &str) -> Option<Arc<Self>> {
stores_guard().remove(id) stores_guard().remove(id)
@@ -830,8 +876,18 @@ impl InMemoryThreadStore {
} }
async fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreResult<()> { 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; let mut state = self.state.lock().await;
state.calls.delete_thread += 1;
let existed = state.histories.remove(&params.thread_id).is_some(); let existed = state.histories.remove(&params.thread_id).is_some();
state.created_threads.remove(&params.thread_id); state.created_threads.remove(&params.thread_id);
state.names.remove(&params.thread_id); state.names.remove(&params.thread_id);
@@ -842,7 +898,7 @@ impl InMemoryThreadStore {
state state
.rollout_paths .rollout_paths
.retain(|_, thread_id| *thread_id != params.thread_id); .retain(|_, thread_id| *thread_id != params.thread_id);
if existed { if existed || deleted_state_rows > 0 {
Ok(()) Ok(())
} else { } else {
Err(ThreadStoreError::ThreadNotFound { Err(ThreadStoreError::ThreadNotFound {
+9
View File
@@ -11,13 +11,19 @@ mod local;
mod projects; mod projects;
mod queue_store; mod queue_store;
mod store; mod store;
mod thread_attachments;
mod thread_metadata_sync; mod thread_metadata_sync;
mod thread_sections; mod thread_sections;
mod types; mod types;
pub use codex_state::AddThreadAttachmentOutcome;
pub use codex_state::MAX_QUEUE_ITEMS; 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::ProjectSortKey;
pub use codex_state::QueuedUserSubmissionRecord; 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::ThreadStoreError;
pub use error::ThreadStoreResult; pub use error::ThreadStoreResult;
pub use in_memory::InMemoryThreadStore; pub use in_memory::InMemoryThreadStore;
@@ -49,6 +55,9 @@ pub use queue_store::QueueStore;
pub use store::PersistContext; pub use store::PersistContext;
pub use store::ThreadStore; pub use store::ThreadStore;
pub use store::ThreadStoreFuture; 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::CreateThreadSectionParams;
pub use thread_sections::DeleteThreadSectionParams; pub use thread_sections::DeleteThreadSectionParams;
pub use thread_sections::ListThreadSectionsParams; pub use thread_sections::ListThreadSectionsParams;
@@ -1,8 +1,9 @@
//! Local hard-delete support for persisted threads. //! Local hard-delete support for persisted threads.
//! //!
//! Existing rollout files are deleted before this operation reports success. A rollout file that //! 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 //! vanishes after discovery counts as already deleted. Main state DB rows are deleted after every
//! after every associated rollout is removed; this module deletes local history projection rows. //! 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::HashMap;
use std::collections::HashSet; use std::collections::HashSet;
@@ -72,7 +73,19 @@ pub(super) async fn delete_thread(
let thread_rollouts = ThreadRollouts::from_index(&reference_index, thread_id); let thread_rollouts = ThreadRollouts::from_index(&reference_index, thread_id);
ensure_no_external_references(&reference_index, std::slice::from_ref(&thread_rollouts))?; ensure_no_external_references(&reference_index, std::slice::from_ref(&thread_rollouts))?;
let mut writer_guards = store.acquire_writer_locks(&[thread_id]).await?; 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( pub(super) async fn delete_threads(
@@ -111,9 +124,26 @@ pub(super) async fn delete_threads(
Err(err) => return Err(err), Err(err) => return Err(err),
} }
} }
// Retain the complete retry graph until every rollout has been removed.
delete_state_rows(store, &thread_ids).await?;
Ok(()) Ok(())
} }
async fn delete_state_rows(
store: &LocalThreadStore,
thread_ids: &[codex_protocol::ThreadId],
) -> ThreadStoreResult<u64> {
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( fn ensure_no_external_references(
reference_index: &RolloutReferenceIndex, reference_index: &RolloutReferenceIndex,
thread_rollouts: &[ThreadRollouts], thread_rollouts: &[ThreadRollouts],
+32
View File
@@ -16,6 +16,7 @@ mod rollout_migration;
#[allow(dead_code)] #[allow(dead_code)]
mod rollout_lineage; mod rollout_lineage;
mod search_threads; mod search_threads;
mod thread_attachments;
mod thread_history; mod thread_history;
mod thread_history_materialization; mod thread_history_materialization;
mod thread_rollout_resolver; mod thread_rollout_resolver;
@@ -52,6 +53,8 @@ use tokio::sync::OwnedRwLockReadGuard;
use tokio::sync::OwnedRwLockWriteGuard; use tokio::sync::OwnedRwLockWriteGuard;
use tokio::sync::RwLock; use tokio::sync::RwLock;
use crate::AddThreadAttachmentOutcome;
use crate::AddThreadAttachmentParams;
use crate::AppendThreadItemsParams; use crate::AppendThreadItemsParams;
use crate::ArchiveThreadParams; use crate::ArchiveThreadParams;
use crate::ArchiveThreadsParams; use crate::ArchiveThreadsParams;
@@ -66,6 +69,7 @@ use crate::DeletedProject;
use crate::ItemPage; use crate::ItemPage;
use crate::ListItemsParams; use crate::ListItemsParams;
use crate::ListProjectsParams; use crate::ListProjectsParams;
use crate::ListThreadAttachmentsParams;
use crate::ListThreadSectionsParams; use crate::ListThreadSectionsParams;
use crate::ListThreadsParams; use crate::ListThreadsParams;
use crate::ListTimelineParams; use crate::ListTimelineParams;
@@ -79,6 +83,8 @@ use crate::PreparedFork;
use crate::ProjectMoveOutcome; use crate::ProjectMoveOutcome;
use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadByRolloutPathParams;
use crate::ReadThreadParams; use crate::ReadThreadParams;
use crate::RemoveThreadAttachmentOutcome;
use crate::RemoveThreadAttachmentParams;
use crate::RenameThreadSectionParams; use crate::RenameThreadSectionParams;
use crate::ResumeThreadParams; use crate::ResumeThreadParams;
use crate::RevertThreadParams; use crate::RevertThreadParams;
@@ -91,6 +97,7 @@ use crate::StoredThread;
use crate::StoredThreadHistory; use crate::StoredThreadHistory;
use crate::StoredThreadSection; use crate::StoredThreadSection;
use crate::StoredThreadSectionsPage; use crate::StoredThreadSectionsPage;
use crate::ThreadAttachmentPage;
use crate::ThreadMetadataPatch; use crate::ThreadMetadataPatch;
use crate::ThreadOccurrenceSearchPage; use crate::ThreadOccurrenceSearchPage;
use crate::ThreadPage; use crate::ThreadPage;
@@ -580,6 +587,31 @@ impl ThreadStore for LocalThreadStore {
Box::pin(async move { thread_sections::delete_thread_section(self, params).await }) 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 { fn supports_projects(&self) -> bool {
self.state_db.is_some() self.state_db.is_some()
} }
@@ -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<AddThreadAttachmentOutcome> {
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,
&params.attachment_type,
&params.identity_key,
&params.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<ThreadAttachmentPage> {
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<RemoveThreadAttachmentOutcome> {
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,
&params.attachment_type,
&params.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<ThreadId>,
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;
@@ -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 &params.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(&params.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<PathBuf>) {
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)
}
+50 -1
View File
@@ -4,6 +4,8 @@ use std::any::Any;
use std::future::Future; use std::future::Future;
use std::pin::Pin; use std::pin::Pin;
use crate::AddThreadAttachmentOutcome;
use crate::AddThreadAttachmentParams;
use crate::AppendThreadItemsParams; use crate::AppendThreadItemsParams;
use crate::ArchiveThreadParams; use crate::ArchiveThreadParams;
use crate::ArchiveThreadsParams; use crate::ArchiveThreadsParams;
@@ -18,6 +20,7 @@ use crate::DeletedProject;
use crate::ItemPage; use crate::ItemPage;
use crate::ListItemsParams; use crate::ListItemsParams;
use crate::ListProjectsParams; use crate::ListProjectsParams;
use crate::ListThreadAttachmentsParams;
use crate::ListThreadSectionsParams; use crate::ListThreadSectionsParams;
use crate::ListThreadsParams; use crate::ListThreadsParams;
use crate::ListTurnsParams; use crate::ListTurnsParams;
@@ -29,6 +32,8 @@ use crate::PreparedFork;
use crate::ProjectMoveOutcome; use crate::ProjectMoveOutcome;
use crate::ReadThreadByRolloutPathParams; use crate::ReadThreadByRolloutPathParams;
use crate::ReadThreadParams; use crate::ReadThreadParams;
use crate::RemoveThreadAttachmentOutcome;
use crate::RemoveThreadAttachmentParams;
use crate::RenameThreadSectionParams; use crate::RenameThreadSectionParams;
use crate::ResumeThreadParams; use crate::ResumeThreadParams;
use crate::RevertThreadParams; use crate::RevertThreadParams;
@@ -41,6 +46,7 @@ use crate::StoredThread;
use crate::StoredThreadHistory; use crate::StoredThreadHistory;
use crate::StoredThreadSection; use crate::StoredThreadSection;
use crate::StoredThreadSectionsPage; use crate::StoredThreadSectionsPage;
use crate::ThreadAttachmentPage;
use crate::ThreadMetadataPatch; use crate::ThreadMetadataPatch;
use crate::ThreadOccurrenceSearchPage; use crate::ThreadOccurrenceSearchPage;
use crate::ThreadPage; 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. /// Whether this store supports durable host-owned projects.
fn supports_projects(&self) -> bool { fn supports_projects(&self) -> bool {
false false
@@ -427,9 +474,11 @@ pub trait ThreadStore: Any + Send + Sync {
fn unarchive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, StoredThread>; fn unarchive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, StoredThread>;
/// Deletes a thread's persisted rollout data and associated metadata. /// 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<'_, ()>; 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 /// Stores with request-scoped delete preflight should override this instead of repeating
/// that work through [`ThreadStore::delete_thread`]. /// that work through [`ThreadStore::delete_thread`].
@@ -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<String>,
/// 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,
}