diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 276f8d73d3..2190a337a8 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -324,7 +324,7 @@ By default, `thread/resume` includes the reconstructed turn history in `thread.t Paginated threads keep the same resume contract as legacy threads. A default resume materializes the full projected history into `thread.turns`; `excludeTurns: true` keeps that array empty and includes `turnsBackwardsCursor` and `itemsBackwardsCursor` for the durable history visible at the resume boundary. Pass each cursor directly to its matching list API with `sortDirection: "desc"`; the first page includes the cursor's head row, while newer records arrive through live notifications. Either cursor is `null` when there is no durable row yet. -Only one app-server process can hold a paginated thread open for writing at a time. If another process already owns the thread, `thread/resume` fails with JSON-RPC error `-32600`. Read-only requests remain available without resuming the thread. +Only one app-server process can hold a paginated thread open for writing at a time. If another process already owns the thread, `thread/resume`, `thread/archive`, and `thread/delete` fail with JSON-RPC error `-32600`. Archive and deletion also fail if another process owns any spawned descendant. Read-only requests remain available without resuming the thread. Experimental clients that want the live resume subscription plus a turns page in one round trip can pass `initialTurnsPage`. It accepts the same `limit`, `sortDirection`, and `itemsView` controls as `thread/turns/list`; omitted controls use its defaults. The response includes `initialTurnsPage` with `nextCursor` and `backwardsCursor` for follow-up pagination. diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index ab48f4daab..13fbf124c5 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -465,6 +465,7 @@ use codex_rollout::state_db::reconcile_rollout; use codex_state::ThreadMetadata; use codex_state::log_db::LogDbLayer; use codex_thread_store::ArchiveThreadParams as StoreArchiveThreadParams; +use codex_thread_store::ArchiveThreadsParams as StoreArchiveThreadsParams; use codex_thread_store::DeleteThreadsParams as StoreDeleteThreadsParams; use codex_thread_store::GitInfoPatch as StoreGitInfoPatch; use codex_thread_store::ItemSortKey as StoreItemSortKey; 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 889d545726..c83140d7ad 100644 --- a/codex-rs/app-server/src/request_processors/thread_delete.rs +++ b/codex-rs/app-server/src/request_processors/thread_delete.rs @@ -149,7 +149,9 @@ fn thread_store_delete_error(err: ThreadStoreError) -> JSONRPCErrorError { ThreadStoreError::ThreadNotFound { thread_id } => { invalid_request(format!("thread not found: {thread_id}")) } - ThreadStoreError::InvalidRequest { message } => invalid_request(message), + ThreadStoreError::InvalidRequest { message } | ThreadStoreError::Conflict { message } => { + invalid_request(message) + } ThreadStoreError::Unsupported { operation } => { unsupported_thread_store_operation(operation) } diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index b77278fb75..80401badad 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -1430,7 +1430,7 @@ impl ThreadRequestProcessor { let thread_id = ThreadId::from_string(¶ms.thread_id) .map_err(|err| invalid_request(format!("invalid session id: {err}")))?; - let thread_ids = self.state_db_spawn_subtree_thread_ids(thread_id).await?; + let subtree_thread_ids = self.state_db_spawn_subtree_thread_ids(thread_id).await?; let mut archive_thread_ids = Vec::new(); match self @@ -1449,7 +1449,7 @@ impl ThreadRequestProcessor { } Err(err) => return Err(thread_store_archive_error("archive", err)), } - for descendant_thread_id in thread_ids.into_iter().skip(1) { + for descendant_thread_id in subtree_thread_ids.iter().copied().skip(1) { match self .thread_store .read_thread(StoreReadThreadParams { @@ -1472,46 +1472,26 @@ impl ThreadRequestProcessor { } } - let mut archived_thread_ids = Vec::new(); - let Some((parent_thread_id, descendant_thread_ids)) = archive_thread_ids.split_first() - else { - return Ok((ThreadArchiveResponse {}, archived_thread_ids)); - }; + if archive_thread_ids.is_empty() { + return Ok((ThreadArchiveResponse {}, Vec::new())); + } - self.prepare_thread_for_archive(*parent_thread_id).await; - match self + archive_thread_ids[1..].reverse(); + for &thread_id_to_archive in &archive_thread_ids { + self.prepare_thread_for_archive(thread_id_to_archive).await; + } + + let archived_thread_ids = self .thread_store - .archive_thread(StoreArchiveThreadParams { - thread_id: *parent_thread_id, + .archive_threads(StoreArchiveThreadsParams { + thread_ids: archive_thread_ids, + writer_lock_thread_ids: subtree_thread_ids, }) .await - { - Ok(()) => { - archived_thread_ids.push(parent_thread_id.to_string()); - } - Err(err) => return Err(thread_store_archive_error("archive", err)), - } - - for descendant_thread_id in descendant_thread_ids.iter().rev().copied() { - self.prepare_thread_for_archive(descendant_thread_id).await; - match self - .thread_store - .archive_thread(StoreArchiveThreadParams { - thread_id: descendant_thread_id, - }) - .await - { - Ok(()) => { - archived_thread_ids.push(descendant_thread_id.to_string()); - } - Err(err) => { - warn!( - "failed to archive spawned descendant thread {descendant_thread_id} while archiving {thread_id}: {err}" - ); - } - } - } - + .map_err(|err| thread_store_archive_error("archive", err))? + .into_iter() + .map(|thread_id| thread_id.to_string()) + .collect(); Ok((ThreadArchiveResponse {}, archived_thread_ids)) } @@ -4953,7 +4933,9 @@ pub(super) fn core_thread_write_error(operation: &str, err: CodexErr) -> JSONRPC fn thread_store_archive_error(operation: &str, err: ThreadStoreError) -> JSONRPCErrorError { match err { - ThreadStoreError::InvalidRequest { message } => invalid_request(message), + ThreadStoreError::InvalidRequest { message } | ThreadStoreError::Conflict { message } => { + invalid_request(message) + } ThreadStoreError::Unsupported { operation: unsupported_operation, } => unsupported_thread_store_operation(unsupported_operation), diff --git a/codex-rs/app-server/tests/suite/v2/thread_archive.rs b/codex-rs/app-server/tests/suite/v2/thread_archive.rs index f850adb384..31630aeff2 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -9,6 +9,7 @@ use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadArchiveParams; use codex_app_server_protocol::ThreadArchiveResponse; use codex_app_server_protocol::ThreadArchivedNotification; +use codex_app_server_protocol::ThreadHistoryMode; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartParams; @@ -33,6 +34,75 @@ use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +#[tokio::test] +async fn thread_archive_rejects_owned_unmaterialized_paginated_descendant() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + let parent_id = create_fake_rollout( + codex_home.path(), + "2025-01-01T00-00-00", + "2025-01-01T00:00:00Z", + "parent", + Some("mock_provider"), + /*git_info*/ None, + )?; + let parent_thread_id = ThreadId::from_string(&parent_id)?; + let mut owner = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let ThreadStartResponse { thread: child, .. } = owner + .start_thread(ThreadStartParams { + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + let child_thread_id = ThreadId::from_string(&child.id)?; + let state_db = StateRuntime::init( + codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()), + "mock_provider".into(), + ) + .await?; + state_db + .upsert_thread_spawn_edge( + parent_thread_id, + child_thread_id, + DirectionalThreadSpawnEdgeStatus::Open, + ) + .await?; + + let mut other = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let request_id = other + .send_thread_archive_request(ThreadArchiveParams { + thread_id: parent_id.clone(), + }) + .await?; + let error: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + other.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!(error.error.code, -32600); + assert_eq!( + error.error.message, + format!("thread {} already has an active writer", child.id) + ); + timeout(DEFAULT_READ_TIMEOUT, owner.shutdown_gracefully()).await??; + let _: ThreadArchiveResponse = other + .request(|request_id| ClientRequest::ThreadArchive { + request_id, + params: ThreadArchiveParams { + thread_id: parent_id, + }, + }) + .await?; + Ok(()) +} + #[tokio::test] async fn thread_archive_requires_materialized_rollout() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; 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 44e5096d1d..7f3741e8b1 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_delete.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_delete.rs @@ -1,6 +1,9 @@ use anyhow::Result; +use app_test_support::MockResponsesConfig; use app_test_support::TestAppServer; +use app_test_support::create_fake_paginated_rollout; use app_test_support::create_fake_rollout; +use app_test_support::create_mock_responses_server_repeating_assistant; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::JSONRPCError; use codex_app_server_protocol::RequestId; @@ -9,6 +12,8 @@ use codex_app_server_protocol::ThreadDeleteResponse; use codex_app_server_protocol::ThreadDeletedNotification; use codex_app_server_protocol::ThreadLoadedListParams; use codex_app_server_protocol::ThreadLoadedListResponse; +use codex_app_server_protocol::ThreadResumeParams; +use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_core::find_thread_path_by_id_str; @@ -25,6 +30,63 @@ use tokio::time::timeout; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); +#[tokio::test] +async fn thread_delete_rejects_paginated_writer_owned_by_another_process() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; + let thread_id = create_fake_paginated_rollout( + codex_home.path(), + "2025-01-01T00-00-00", + "2025-01-01T00:00:00Z", + "owned", + Some("mock_provider"), + /*git_info*/ None, + )?; + let mut owner = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let _: ThreadResumeResponse = owner + .request(|request_id| ClientRequest::ThreadResume { + request_id, + params: ThreadResumeParams { + thread_id: thread_id.clone(), + exclude_turns: true, + ..Default::default() + }, + }) + .await?; + + let mut other = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let request_id = other + .send_thread_delete_request(ThreadDeleteParams { + thread_id: thread_id.clone(), + }) + .await?; + let error: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + other.read_stream_until_error_message(RequestId::Integer(request_id)), + ) + .await??; + assert_eq!(error.error.code, -32600); + assert_eq!( + error.error.message, + format!("thread {thread_id} already has an active writer") + ); + timeout(DEFAULT_READ_TIMEOUT, owner.shutdown_gracefully()).await??; + let _: ThreadDeleteResponse = other + .request(|request_id| ClientRequest::ThreadDelete { + request_id, + params: ThreadDeleteParams { thread_id }, + }) + .await?; + Ok(()) +} + #[tokio::test] async fn thread_delete_deletes_spawned_descendants() -> Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index 18b7312985..8516b68f12 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -24,6 +24,7 @@ pub use store::ThreadStore; pub use store::ThreadStoreFuture; pub use types::AppendThreadItemsParams; pub use types::ArchiveThreadParams; +pub use types::ArchiveThreadsParams; pub use types::ClearableField; pub use types::CreateThreadParams; pub use types::DeleteThreadParams; diff --git a/codex-rs/thread-store/src/local/archive_thread.rs b/codex-rs/thread-store/src/local/archive_thread.rs index 0d77f98219..86bc26d108 100644 --- a/codex-rs/thread-store/src/local/archive_thread.rs +++ b/codex-rs/thread-store/src/local/archive_thread.rs @@ -2,10 +2,58 @@ use super::LocalThreadStore; use super::helpers::matching_rollout_file_name; use super::helpers::scoped_rollout_path; use crate::ArchiveThreadParams; +use crate::ArchiveThreadsParams; use crate::ThreadStoreError; use crate::ThreadStoreResult; use chrono::Utc; use codex_rollout::find_thread_path_by_id_str; +use tracing::warn; + +pub(super) async fn archive_threads( + store: &LocalThreadStore, + params: ArchiveThreadsParams, +) -> ThreadStoreResult> { + let thread_ids = params.thread_ids; + if thread_ids.is_empty() { + return Ok(Vec::new()); + } + + let mut lock_thread_ids = params.writer_lock_thread_ids; + lock_thread_ids.extend(thread_ids.iter().copied()); + lock_thread_ids.sort_unstable_by_key(ToString::to_string); + lock_thread_ids.dedup(); + let mut _live_writer_guards = Vec::with_capacity(lock_thread_ids.len()); + for thread_id in &lock_thread_ids { + _live_writer_guards.push(store.live_writer_locks.lock(*thread_id).await); + if store + .live_recorders + .lock() + .await + .get(thread_id) + .is_some_and(|entry| entry.writer_lock.is_some()) + { + return Err(ThreadStoreError::Conflict { + message: format!("thread {thread_id} already has an active writer"), + }); + } + } + let _writer_guards = store + .acquire_paginated_writer_locks(&lock_thread_ids) + .await?; + + let parent_thread_id = thread_ids[0]; + let mut archived_thread_ids = Vec::new(); + for thread_id in thread_ids { + match archive_thread(store, ArchiveThreadParams { thread_id }).await { + Ok(()) => archived_thread_ids.push(thread_id), + Err(err) if archived_thread_ids.is_empty() => return Err(err), + Err(err) => warn!( + "failed to archive spawned descendant thread {thread_id} while archiving {parent_thread_id}: {err}" + ), + } + } + Ok(archived_thread_ids) +} pub(super) async fn archive_thread( store: &LocalThreadStore, @@ -64,6 +112,7 @@ mod tests { use chrono::Utc; use codex_protocol::ThreadId; use codex_protocol::protocol::SessionSource; + use codex_protocol::protocol::ThreadHistoryMode; use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; use codex_utils_absolute_path::test_support::PathExt; use pretty_assertions::assert_eq; @@ -77,6 +126,50 @@ mod tests { use crate::local::LocalThreadStore; use crate::local::test_support::test_config; use crate::local::test_support::write_session_file; + use crate::local::test_support::write_session_file_with_history_mode; + + #[tokio::test] + async fn archive_threads_rejects_owned_descendants_before_archiving_anything() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let owner = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let parent_uuid = Uuid::from_u128(203); + let parent_thread_id = + ThreadId::from_string(&parent_uuid.to_string()).expect("valid parent thread id"); + let parent_path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T12-00-00", + parent_uuid, + ThreadHistoryMode::Paginated, + ) + .expect("parent session file"); + let child_uuid = Uuid::from_u128(204); + let child_thread_id = + ThreadId::from_string(&child_uuid.to_string()).expect("valid child thread id"); + let child_path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T12-00-01", + child_uuid, + ThreadHistoryMode::Paginated, + ) + .expect("child session file"); + let _owner_guard = owner + .writer_lock_coordinator + .acquire(child_thread_id) + .expect("acquire child writer lock"); + + let error = store + .archive_threads(ArchiveThreadsParams { + thread_ids: vec![parent_thread_id, child_thread_id], + writer_lock_thread_ids: Vec::new(), + }) + .await + .expect_err("owned descendant should block archive"); + + assert!(matches!(error, ThreadStoreError::Conflict { .. })); + assert!(parent_path.exists()); + assert!(child_path.exists()); + } #[tokio::test] async fn archive_thread_moves_rollout_to_archived_collection() { diff --git a/codex-rs/thread-store/src/local/delete_thread.rs b/codex-rs/thread-store/src/local/delete_thread.rs index b30c164844..eb54ba21f3 100644 --- a/codex-rs/thread-store/src/local/delete_thread.rs +++ b/codex-rs/thread-store/src/local/delete_thread.rs @@ -34,7 +34,8 @@ pub(super) async fn delete_thread( if reference_index.reference_count(thread_id) > 0 { return Err(referenced_thread_error(thread_id)); } - delete_thread_after_reference_check(store, thread_id).await + let mut writer_guards = store.acquire_paginated_writer_locks(&[thread_id]).await?; + delete_thread_after_reference_check(store, thread_id, &mut writer_guards).await } pub(super) async fn delete_threads( @@ -50,7 +51,7 @@ pub(super) async fn delete_threads( let mut lock_thread_ids: Vec<_> = deletion_set.iter().copied().collect(); lock_thread_ids.sort_unstable_by_key(ToString::to_string); let mut _live_writer_guards = Vec::with_capacity(lock_thread_ids.len()); - for thread_id in lock_thread_ids { + for &thread_id in &lock_thread_ids { _live_writer_guards.push(store.live_writer_locks.lock(thread_id).await); } @@ -78,8 +79,11 @@ pub(super) async fn delete_threads( } } + let mut writer_guards = store + .acquire_paginated_writer_locks(&lock_thread_ids) + .await?; for thread_id in thread_ids { - match delete_thread_after_reference_check(store, thread_id).await { + match delete_thread_after_reference_check(store, thread_id, &mut writer_guards).await { Ok(()) | Err(ThreadStoreError::ThreadNotFound { .. }) => {} Err(err) => return Err(err), } @@ -106,6 +110,7 @@ fn referenced_thread_error(thread_id: codex_protocol::ThreadId) -> ThreadStoreEr async fn delete_thread_after_reference_check( store: &LocalThreadStore, thread_id: codex_protocol::ThreadId, + writer_guards: &mut Vec, ) -> ThreadStoreResult<()> { let thread_id_str = thread_id.to_string(); let state_db_ctx = store.state_db().await; @@ -145,12 +150,15 @@ async fn delete_thread_after_reference_check( } } // Drop the recorder before removing files, but retain its writer lock until cleanup finishes. - let _writer_lock = store + if let Some(writer_lock) = store .live_recorders .lock() .await .remove(&thread_id) - .and_then(|entry| entry._writer_lock); + .and_then(|entry| entry.writer_lock) + { + writer_guards.push(writer_lock); + } let found_rollout_path = !rollout_paths.is_empty(); for rollout_path in rollout_paths { delete_rollout_file(store, rollout_path.as_path(), thread_id)?; @@ -395,6 +403,89 @@ mod tests { assert!(!child_path.exists()); } + #[tokio::test] + async fn delete_threads_rejects_owned_descendants_before_deleting_anything() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let owner = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let parent_uuid = Uuid::from_u128(309); + let parent_thread_id = + ThreadId::from_string(&parent_uuid.to_string()).expect("valid parent thread id"); + let parent_path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T12-00-00", + parent_uuid, + ThreadHistoryMode::Paginated, + ) + .expect("parent session file"); + let child_uuid = Uuid::from_u128(310); + let child_thread_id = + ThreadId::from_string(&child_uuid.to_string()).expect("valid child thread id"); + let child_path = write_session_file_with_history_mode( + home.path(), + "2025-01-03T12-00-01", + child_uuid, + ThreadHistoryMode::Paginated, + ) + .expect("child session file"); + let _owner_guard = owner + .writer_lock_coordinator + .acquire(child_thread_id) + .expect("acquire child writer lock"); + + let error = store + .delete_threads(DeleteThreadsParams { + thread_ids: vec![parent_thread_id, child_thread_id], + }) + .await + .expect_err("owned descendant should block deletion"); + + assert!(matches!(error, ThreadStoreError::Conflict { .. })); + assert!(parent_path.exists()); + assert!(child_path.exists()); + } + + #[tokio::test] + async fn delete_threads_rejects_owned_thread_before_rollout_materializes() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let owner = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let thread_id = ThreadId::default(); + let _owner_guard = owner + .writer_lock_coordinator + .acquire(thread_id) + .expect("acquire writer lock before rollout exists"); + + let error = store + .delete_threads(DeleteThreadsParams { + thread_ids: vec![thread_id], + }) + .await + .expect_err("owned thread should block deletion before rollout exists"); + + assert!(matches!(error, ThreadStoreError::Conflict { .. })); + } + + #[tokio::test] + async fn delete_threads_removes_rollout_with_unreadable_metadata() { + let home = TempDir::new().expect("temp dir"); + let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); + let uuid = Uuid::from_u128(311); + let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); + let rollout_path = + write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); + std::fs::write(&rollout_path, "{not json}\n").expect("damage rollout metadata"); + + store + .delete_threads(DeleteThreadsParams { + thread_ids: vec![thread_id], + }) + .await + .expect("delete rollout with unreadable metadata"); + + assert!(!rollout_path.exists()); + } + #[tokio::test] async fn delete_rollout_file_treats_vanished_path_as_already_deleted() { let home = TempDir::new().expect("temp dir"); diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 0ebe34f000..fa3ed53b8d 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -34,6 +34,7 @@ use tokio::sync::OwnedMutexGuard; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; +use crate::ArchiveThreadsParams; use crate::CreateThreadParams; use crate::DeleteThreadParams; use crate::DeleteThreadsParams; @@ -91,7 +92,7 @@ struct LiveRecorderEntry { // canonical SessionMeta is durable. Retain the mode captured when live persistence was opened // so missing SQLite rows can still be seeded. history_mode: ThreadHistoryMode, - _writer_lock: Option, + writer_lock: Option, } #[derive(Default)] @@ -207,6 +208,42 @@ impl LocalThreadStore { Ok(()) } + async fn acquire_paginated_writer_locks( + &self, + thread_ids: &[ThreadId], + ) -> ThreadStoreResult> { + let mut writer_locks = Vec::new(); + for &thread_id in thread_ids { + if self + .live_recorders + .lock() + .await + .get(&thread_id) + .is_some_and(|entry| entry.writer_lock.is_some()) + { + continue; + } + + // Only a readable legacy header proves no paginated writer can own this id. Missing + // lazy rollouts and damaged headers must conservatively try the lock. + let history_mode = match read_thread::resolve_rollout_path( + self, thread_id, /*include_archived*/ true, + ) + .await? + { + Some(rollout_path) => codex_rollout::read_session_meta_line(rollout_path.as_path()) + .await + .ok() + .map(|meta_line| meta_line.meta.history_mode), + None => None, + }; + if !matches!(history_mode, Some(ThreadHistoryMode::Legacy)) { + writer_locks.push(self.writer_lock_coordinator.acquire(thread_id)?); + } + } + Ok(writer_locks) + } + async fn insert_live_recorder( &self, thread_id: ThreadId, @@ -222,7 +259,7 @@ impl LocalThreadStore { entry.insert(LiveRecorderEntry { recorder, history_mode, - _writer_lock: writer_lock, + writer_lock, }); Ok(()) } @@ -402,7 +439,24 @@ impl ThreadStore for LocalThreadStore { } fn archive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, ()> { - Box::pin(async move { archive_thread::archive_thread(self, params).await }) + Box::pin(async move { + archive_thread::archive_threads( + self, + ArchiveThreadsParams { + thread_ids: vec![params.thread_id], + writer_lock_thread_ids: Vec::new(), + }, + ) + .await + .map(|_| ()) + }) + } + + fn archive_threads( + &self, + params: ArchiveThreadsParams, + ) -> ThreadStoreFuture<'_, Vec> { + Box::pin(async move { archive_thread::archive_threads(self, params).await }) } fn unarchive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, StoredThread> { diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index f8e798f612..71516c6eaa 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -6,6 +6,7 @@ use std::pin::Pin; use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; +use crate::ArchiveThreadsParams; use crate::CreateThreadParams; use crate::DeleteThreadParams; use crate::DeleteThreadsParams; @@ -167,6 +168,26 @@ pub trait ThreadStore: Any + Send + Sync { /// Archives a thread. fn archive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, ()>; + /// Archives threads in order, returning the successfully archived thread ids. + /// + /// The first thread must archive successfully; later failures are best effort. + fn archive_threads( + &self, + params: ArchiveThreadsParams, + ) -> ThreadStoreFuture<'_, Vec> { + Box::pin(async move { + let mut archived_thread_ids = Vec::new(); + for thread_id in params.thread_ids { + match self.archive_thread(ArchiveThreadParams { thread_id }).await { + Ok(()) => archived_thread_ids.push(thread_id), + Err(err) if archived_thread_ids.is_empty() => return Err(err), + Err(err) => tracing::warn!("failed to archive thread {thread_id}: {err}"), + } + } + Ok(archived_thread_ids) + }) + } + /// Unarchives a thread and returns its updated metadata. fn unarchive_thread(&self, params: ArchiveThreadParams) -> ThreadStoreFuture<'_, StoredThread>; diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index ccfe7c0f71..fac67309b7 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -806,6 +806,17 @@ pub struct ArchiveThreadParams { pub thread_id: ThreadId, } +/// Parameters for archiving a set of threads as one store operation. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct ArchiveThreadsParams { + /// Thread ids to archive, in the order their persisted data should be moved. + pub thread_ids: Vec, + /// Thread ids whose paginated writer ownership must be checked before archiving, including + /// descendants whose rollout has not materialized yet. + #[serde(default)] + pub writer_lock_thread_ids: Vec, +} + /// Parameters for deleting a thread. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct DeleteThreadParams {