Enforce writer ownership for thread archive and deletion (#35031)

## Why

Paginated threads allow only one app-server process to write at a time. Archive and delete operations must not mutate a thread or its spawned descendants while another process owns any of them.

## What changed

- Acquire paginated writer locks for the full archive or deletion set before changing rollout files, including descendants whose rollout has not materialized yet.
- Add a batch archive store operation so ownership conflicts are detected before any thread in the subtree is archived.
- Return ownership conflicts as JSON-RPC invalid-request errors and document the behavior.

## Testing

- Cover archive and deletion conflicts for owned threads and descendants, including unmaterialized rollouts.
- Verify deletion still removes rollouts with unreadable metadata.

GitOrigin-RevId: f62a3e0c1ee2631cea48dd9145b588a1486e2425

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<Vec<codex_protocol::ThreadId>> {
+    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() {