Protect fork history references during rollout cleanup (#34566)

## Why

Paginated forks can retain a `history_base` pointer to another rollout. Deleting or compressing either side of that relationship can leave the fork without the rollout history it references.

## What changed

- Add a rollout reference index that scans active, archived, and compressed rollout metadata.
- Reject deletion when a rollout is still referenced by a fork outside the requested deletion set.
- Delete spawned thread subtrees as one store operation so internal references do not block deleting the complete subtree, while external references preflight the request before any files are removed.
- Skip referenced rollouts, fork-pointer rollouts, and rollouts with unreadable metadata during background compression.

## Testing

- Cover reference indexing across active, archived, compressed, duplicate, and self-referencing rollouts.
- Verify deletion rejects external references without partially deleting a subtree and permits batch deletion of internally referenced threads.
- Verify compression preserves fork pointer chains and continues past unreadable metadata.

GitOrigin-RevId: a8672727f8706e743d7fb3ca895a1d9ceac98a3c

codex-rs/thread-store/src/local/delete_thread.rs

@@ -4,10 +4,13 @@
 //! 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.
 
+use std::collections::HashMap;
+use std::collections::HashSet;
 use std::io::ErrorKind;
 use std::path::Path;
 
 use codex_rollout::ARCHIVED_SESSIONS_SUBDIR;
+use codex_rollout::RolloutReferenceIndex;
 use codex_rollout::SESSIONS_SUBDIR;
 use codex_rollout::find_archived_thread_path_by_id_str;
 use codex_rollout::find_thread_path_by_id_str;
@@ -17,6 +20,7 @@ use super::LocalThreadStore;
 use super::helpers::matching_rollout_file_name;
 use super::helpers::scoped_rollout_path;
 use crate::DeleteThreadParams;
+use crate::DeleteThreadsParams;
 use crate::ThreadStoreError;
 use crate::ThreadStoreResult;
 
@@ -26,6 +30,83 @@ pub(super) async fn delete_thread(
 ) -> ThreadStoreResult<()> {
     let thread_id = params.thread_id;
     let _live_writer_guard = store.live_writer_locks.lock(thread_id).await;
+    let reference_index = scan_reference_index(store).await?;
+    if reference_index.reference_count(thread_id) > 0 {
+        return Err(referenced_thread_error(thread_id));
+    }
+    delete_thread_after_reference_check(store, thread_id).await
+}
+
+pub(super) async fn delete_threads(
+    store: &LocalThreadStore,
+    params: DeleteThreadsParams,
+) -> ThreadStoreResult<()> {
+    let thread_ids = params.thread_ids;
+    if thread_ids.is_empty() {
+        return Ok(());
+    }
+
+    let deletion_set: HashSet<_> = thread_ids.iter().copied().collect();
+    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 {
+        _live_writer_guards.push(store.live_writer_locks.lock(thread_id).await);
+    }
+
+    let reference_index = scan_reference_index(store).await?;
+    // References from children in this delete set are removed by the same request, so only
+    // references from children outside the set should block it.
+    let mut internal_reference_counts = HashMap::new();
+    for child_thread_id in &deletion_set {
+        if let Some(history_base) = reference_index.history_base(*child_thread_id)
+            && history_base.thread_id != *child_thread_id
+            && deletion_set.contains(&history_base.thread_id)
+        {
+            *internal_reference_counts
+                .entry(history_base.thread_id)
+                .or_default() += 1;
+        }
+    }
+    for thread_id in &thread_ids {
+        let internal_reference_count = internal_reference_counts
+            .get(thread_id)
+            .copied()
+            .unwrap_or_default();
+        if reference_index.reference_count(*thread_id) > internal_reference_count {
+            return Err(referenced_thread_error(*thread_id));
+        }
+    }
+
+    for thread_id in thread_ids {
+        match delete_thread_after_reference_check(store, thread_id).await {
+            Ok(()) | Err(ThreadStoreError::ThreadNotFound { .. }) => {}
+            Err(err) => return Err(err),
+        }
+    }
+    Ok(())
+}
+
+async fn scan_reference_index(
+    store: &LocalThreadStore,
+) -> ThreadStoreResult<RolloutReferenceIndex> {
+    RolloutReferenceIndex::scan(store.config.codex_home.as_path())
+        .await
+        .map_err(|err| ThreadStoreError::Internal {
+            message: format!("failed to scan fork history references: {err}"),
+        })
+}
+
+fn referenced_thread_error(thread_id: codex_protocol::ThreadId) -> ThreadStoreError {
+    ThreadStoreError::InvalidRequest {
+        message: format!("cannot delete thread {thread_id}: forked history still references it"),
+    }
+}
+
+async fn delete_thread_after_reference_check(
+    store: &LocalThreadStore,
+    thread_id: codex_protocol::ThreadId,
+) -> ThreadStoreResult<()> {
     let thread_id_str = thread_id.to_string();
     let state_db_ctx = store.state_db().await;
     let mut rollout_paths = Vec::new();
@@ -135,6 +216,7 @@ fn delete_rollout_path(
 #[cfg(test)]
 mod tests {
     use codex_protocol::ThreadId;
+    use codex_protocol::protocol::HistoryPosition;
     use codex_protocol::protocol::ThreadHistoryMode;
     use pretty_assertions::assert_eq;
     use tempfile::TempDir;
@@ -146,6 +228,7 @@ mod tests {
     use crate::local::test_support::test_config;
     use crate::local::test_support::write_archived_session_file;
     use crate::local::test_support::write_session_file;
+    use crate::local::test_support::write_session_file_with;
     use crate::local::test_support::write_session_file_with_history_mode;
 
     #[tokio::test]
@@ -182,6 +265,128 @@ mod tests {
         assert!(!compressed_path.exists());
     }
 
+    #[tokio::test]
+    async fn delete_thread_rejects_referenced_paginated_history() {
+        let home = TempDir::new().expect("temp dir");
+        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+        let source_uuid = Uuid::from_u128(303);
+        let source_thread_id =
+            ThreadId::from_string(&source_uuid.to_string()).expect("valid source thread id");
+        let source_path = write_session_file_with_history_mode(
+            home.path(),
+            "2025-01-03T12-00-00",
+            source_uuid,
+            ThreadHistoryMode::Paginated,
+        )
+        .expect("source session file");
+        let child_path = write_session_file_with(
+            home.path(),
+            home.path().join(ARCHIVED_SESSIONS_SUBDIR),
+            "2025-01-03T12-00-01",
+            Uuid::from_u128(304),
+            "Archived user message",
+            Some("test-provider"),
+            ThreadHistoryMode::Paginated,
+        )
+        .expect("child session file");
+        set_history_base(
+            child_path.as_path(),
+            HistoryPosition {
+                thread_id: source_thread_id,
+                end_ordinal_exclusive: 1,
+                end_byte_offset: std::fs::metadata(source_path.as_path())
+                    .expect("source rollout metadata")
+                    .len(),
+            },
+        );
+
+        let err = store
+            .delete_thread(DeleteThreadParams {
+                thread_id: source_thread_id,
+            })
+            .await
+            .expect_err("referenced source should not be deleted");
+
+        assert_eq!(
+            err.to_string(),
+            format!(
+                "invalid thread-store request: cannot delete thread {source_thread_id}: forked history still references it"
+            )
+        );
+        assert!(source_path.exists());
+    }
+
+    #[tokio::test]
+    async fn delete_thread_ignores_unreadable_reference_metadata() {
+        let home = TempDir::new().expect("temp dir");
+        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+        let source_uuid = Uuid::from_u128(305);
+        let source_thread_id =
+            ThreadId::from_string(&source_uuid.to_string()).expect("valid source thread id");
+        let source_path = write_session_file(home.path(), "2025-01-03T12-00-00", source_uuid)
+            .expect("source session file");
+        let unreadable_path = source_path.with_file_name(format!(
+            "rollout-2025-01-03T12-00-01-{}.jsonl",
+            Uuid::from_u128(306)
+        ));
+        std::fs::write(unreadable_path, "{not json}\n").expect("unreadable rollout metadata");
+
+        store
+            .delete_thread(DeleteThreadParams {
+                thread_id: source_thread_id,
+            })
+            .await
+            .expect("unreadable metadata should not block delete");
+
+        assert!(!source_path.exists());
+    }
+
+    #[tokio::test]
+    async fn delete_threads_allows_internal_history_references() {
+        let home = TempDir::new().expect("temp dir");
+        let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None);
+        let source_uuid = Uuid::from_u128(307);
+        let source_thread_id =
+            ThreadId::from_string(&source_uuid.to_string()).expect("valid source thread id");
+        let source_path = write_session_file_with_history_mode(
+            home.path(),
+            "2025-01-03T12-00-00",
+            source_uuid,
+            ThreadHistoryMode::Paginated,
+        )
+        .expect("source session file");
+        let child_uuid = Uuid::from_u128(308);
+        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");
+        set_history_base(
+            child_path.as_path(),
+            HistoryPosition {
+                thread_id: source_thread_id,
+                end_ordinal_exclusive: 1,
+                end_byte_offset: std::fs::metadata(source_path.as_path())
+                    .expect("source rollout metadata")
+                    .len(),
+            },
+        );
+
+        store
+            .delete_threads(DeleteThreadsParams {
+                thread_ids: vec![child_thread_id, source_thread_id],
+            })
+            .await
+            .expect("internal references should not block batch delete");
+
+        assert!(!source_path.exists());
+        assert!(!child_path.exists());
+    }
+
     #[tokio::test]
     async fn delete_rollout_file_treats_vanished_path_as_already_deleted() {
         let home = TempDir::new().expect("temp dir");
@@ -272,4 +477,18 @@ SELECT
             "thread 00000000-0000-0000-0000-000000000304 not found"
         );
     }
+
+    fn set_history_base(path: &Path, history_base: HistoryPosition) {
+        let mut session_meta: serde_json::Value = serde_json::from_str(
+            std::fs::read_to_string(path)
+                .expect("read session file")
+                .lines()
+                .next()
+                .expect("session metadata"),
+        )
+        .expect("parse session metadata");
+        session_meta["payload"]["history_base"] =
+            serde_json::to_value(history_base).expect("serialize history base");
+        std::fs::write(path, format!("{session_meta}\n")).expect("write session file");
+    }
 }