diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index a6d1794af0..25a422799b 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -465,7 +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::DeleteThreadParams as StoreDeleteThreadParams; +use codex_thread_store::DeleteThreadsParams as StoreDeleteThreadsParams; use codex_thread_store::GitInfoPatch as StoreGitInfoPatch; use codex_thread_store::ListItemsParams as StoreListItemsParams; use codex_thread_store::ListThreadsParams as StoreListThreadsParams; 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 ae78b1f4bc..889d545726 100644 --- a/codex-rs/app-server/src/request_processors/thread_delete.rs +++ b/codex-rs/app-server/src/request_processors/thread_delete.rs @@ -47,25 +47,12 @@ impl ThreadRequestProcessor { let mut delete_order: Vec<_> = thread_ids.iter().skip(1).rev().copied().collect(); delete_order.push(thread_id); - for thread_id_to_delete in delete_order.iter().copied() { - match self - .thread_store - .delete_thread(StoreDeleteThreadParams { - thread_id: thread_id_to_delete, - }) - .await - { - Ok(()) => {} - Err(ThreadStoreError::ThreadNotFound { .. }) => { - warn!( - "thread {thread_id_to_delete} was already missing while deleting {thread_id}" - ); - } - Err(err) => { - return Err(thread_store_delete_error(err)); - } - } - } + self.thread_store + .delete_threads(StoreDeleteThreadsParams { + thread_ids: delete_order.clone(), + }) + .await + .map_err(thread_store_delete_error)?; if let Some(state_db) = self.state_db.as_ref() { state_db 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 a2aaf7a119..96cf2add63 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_delete.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_delete.rs @@ -14,6 +14,7 @@ use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_core::find_thread_path_by_id_str; use codex_protocol::ThreadId; +use codex_protocol::protocol::HistoryPosition; use codex_state::DirectionalThreadSpawnEdgeStatus; use codex_state::StateRuntime; use pretty_assertions::assert_eq; @@ -111,6 +112,96 @@ async fn thread_delete_deletes_spawned_descendants() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_delete_preflights_external_fork_references_for_spawned_subtrees() -> Result<()> { + let codex_home = TempDir::new()?; + + let parent_id = create_delete_test_rollout(codex_home.path(), /*minute*/ 0, "parent")?; + let child_id = create_delete_test_rollout(codex_home.path(), /*minute*/ 1, "child")?; + let external_id = create_delete_test_rollout(codex_home.path(), /*minute*/ 2, "external")?; + let parent_thread_id = ThreadId::from_string(&parent_id)?; + let child_thread_id = ThreadId::from_string(&child_id)?; + let external_thread_id = ThreadId::from_string(&external_id)?; + let parent_path = find_thread_path_by_id_str( + codex_home.path(), + &parent_thread_id.to_string(), + /*state_db_ctx*/ None, + ) + .await? + .expect("parent rollout path"); + let external_path = find_thread_path_by_id_str( + codex_home.path(), + &external_thread_id.to_string(), + /*state_db_ctx*/ None, + ) + .await? + .expect("external rollout path"); + let mut external_meta: serde_json::Value = serde_json::from_str( + std::fs::read_to_string(external_path.as_path())? + .lines() + .next() + .expect("external session metadata"), + )?; + external_meta["payload"]["history_base"] = serde_json::to_value(HistoryPosition { + thread_id: parent_thread_id, + end_ordinal_exclusive: 1, + end_byte_offset: std::fs::metadata(parent_path.as_path())?.len(), + })?; + std::fs::write(external_path.as_path(), format!("{external_meta}\n"))?; + + let state_db = + StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".into()).await?; + state_db + .upsert_thread_spawn_edge( + parent_thread_id, + child_thread_id, + DirectionalThreadSpawnEdgeStatus::Closed, + ) + .await?; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build() + .await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let delete_id = mcp + .send_thread_delete_request(ThreadDeleteParams { + thread_id: parent_id.clone(), + }) + .await?; + let delete_err: JSONRPCError = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_error_message(RequestId::Integer(delete_id)), + ) + .await??; + assert_eq!( + delete_err.error.message, + format!("cannot delete thread {parent_thread_id}: forked history still references it") + ); + + for thread_id in [parent_thread_id, child_thread_id, external_thread_id] { + assert!( + find_thread_path_by_id_str( + codex_home.path(), + &thread_id.to_string(), + /*state_db_ctx*/ None, + ) + .await? + .is_some(), + "expected rollout for {thread_id} to remain" + ); + } + assert_eq!( + state_db + .list_thread_spawn_descendants(parent_thread_id) + .await?, + vec![child_thread_id] + ); + Ok(()) +} + fn create_delete_test_rollout(codex_home: &Path, minute: u8, preview: &str) -> Result { create_fake_rollout( codex_home, diff --git a/codex-rs/rollout/src/compression.rs b/codex-rs/rollout/src/compression.rs index e64cc318a6..d422384d40 100644 --- a/codex-rs/rollout/src/compression.rs +++ b/codex-rs/rollout/src/compression.rs @@ -242,6 +242,7 @@ mod worker { use tokio::task::JoinSet; use crate::ARCHIVED_SESSIONS_SUBDIR; + use crate::RolloutReferenceIndex; use crate::SESSIONS_SUBDIR; use super::RolloutFile; @@ -361,6 +362,15 @@ mod worker { let started_at = Instant::now(); let result = async { cleanup_stale_temps(codex_home.as_path()).await?; + let Some(reference_index) = RolloutReferenceIndex::scan_until( + codex_home.as_path(), + started_at, + WORKER_MAX_RUNTIME, + ) + .await? + else { + return Ok(CompressionStats::default()); + }; let mut stats = CompressionStats::default(); for root in [ codex_home.join(ARCHIVED_SESSIONS_SUBDIR), @@ -369,7 +379,8 @@ mod worker { if started_at.elapsed() >= WORKER_MAX_RUNTIME { break; } - compress_rollouts_in_root(root.as_path(), started_at, &mut stats).await?; + compress_rollouts_in_root(root.as_path(), started_at, &reference_index, &mut stats) + .await?; } Ok::<_, io::Error>(stats) } @@ -409,6 +420,7 @@ mod worker { async fn compress_rollouts_in_root( root: &Path, started_at: Instant, + reference_index: &RolloutReferenceIndex, stats: &mut CompressionStats, ) -> io::Result<()> { if !tokio::fs::try_exists(root).await.unwrap_or(false) { @@ -467,6 +479,21 @@ mod worker { continue; } let path = rollout_file.into_path(); + let Ok(meta) = crate::read_session_meta_line(path.as_path()).await else { + stats.skipped = stats.skipped.saturating_add(1); + metrics::file("skipped_unreadable_meta"); + continue; + }; + if reference_index.reference_count(meta.meta.id) > 0 { + stats.skipped = stats.skipped.saturating_add(1); + metrics::file("skipped_referenced"); + continue; + } + if meta.meta.history_base.is_some() { + stats.skipped = stats.skipped.saturating_add(1); + metrics::file("skipped_fork_pointer"); + continue; + } stats.scanned = stats.scanned.saturating_add(1); metrics::file("scanned"); while jobs.len() >= MAX_CONCURRENT_COMPRESSION_JOBS { diff --git a/codex-rs/rollout/src/compression_tests.rs b/codex-rs/rollout/src/compression_tests.rs index 2c2688ff30..f5c74a353e 100644 --- a/codex-rs/rollout/src/compression_tests.rs +++ b/codex-rs/rollout/src/compression_tests.rs @@ -7,6 +7,7 @@ use std::time::SystemTime; use codex_protocol::ThreadId; use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::HistoryPosition; use codex_protocol::protocol::InitialHistory; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::RolloutLine; @@ -306,6 +307,86 @@ async fn worker_compresses_old_active_and_archived_rollouts() -> anyhow::Result< Ok(()) } +#[tokio::test] +async fn worker_skips_archived_paginated_fork_pointer_chain() -> anyhow::Result<()> { + let home = TempDir::new()?; + let source_uuid = Uuid::from_u128(16); + let source_id = ThreadId::from_string(&source_uuid.to_string())?; + let source_path = rollout_path(home.path(), "2025-01-03T12-00-00", source_uuid); + write_rollout(&source_path, source_id, "referenced source")?; + set_old_mtime(&source_path)?; + + let child_uuid = Uuid::from_u128(17); + let child_id = ThreadId::from_string(&child_uuid.to_string())?; + let child_path = archived_rollout_path(home.path(), "2025-01-03T12-00-01", child_uuid); + write_rollout(&child_path, child_id, "fork child")?; + set_history_base( + child_path.as_path(), + HistoryPosition { + thread_id: source_id, + end_ordinal_exclusive: 2, + end_byte_offset: std::fs::metadata(source_path.as_path())?.len(), + }, + )?; + set_old_mtime(&child_path)?; + + worker::run(home.path().to_path_buf()).await?; + + assert!(source_path.exists()); + assert!(child_path.exists()); + Ok(()) +} + +#[tokio::test] +async fn worker_skips_source_referenced_by_archived_compressed_rollout() -> anyhow::Result<()> { + let home = TempDir::new()?; + let source_uuid = Uuid::from_u128(18); + let source_id = ThreadId::from_string(&source_uuid.to_string())?; + let source_path = rollout_path(home.path(), "2025-01-03T12-00-00", source_uuid); + write_rollout(&source_path, source_id, "referenced source")?; + set_old_mtime(&source_path)?; + + let child_uuid = Uuid::from_u128(19); + let child_id = ThreadId::from_string(&child_uuid.to_string())?; + let child_path = archived_rollout_path(home.path(), "2025-01-03T12-00-01", child_uuid); + write_rollout(&child_path, child_id, "fork child")?; + set_history_base( + child_path.as_path(), + HistoryPosition { + thread_id: source_id, + end_ordinal_exclusive: 2, + end_byte_offset: std::fs::metadata(source_path.as_path())?.len(), + }, + )?; + compress_now(child_path.as_path())?; + + worker::run(home.path().to_path_buf()).await?; + + assert!(source_path.exists()); + Ok(()) +} + +#[tokio::test] +async fn worker_skips_unreadable_metadata_without_blocking_other_compression() -> anyhow::Result<()> +{ + let home = TempDir::new()?; + let source_uuid = Uuid::from_u128(20); + let source_id = ThreadId::from_string(&source_uuid.to_string())?; + let source_path = rollout_path(home.path(), "2025-01-03T12-00-00", source_uuid); + write_rollout(&source_path, source_id, "candidate")?; + set_old_mtime(&source_path)?; + + let unreadable_path = rollout_path(home.path(), "2025-01-03T12-00-01", Uuid::from_u128(21)); + fs::write(unreadable_path.as_path(), "{not json}\n")?; + + worker::run(home.path().to_path_buf()).await?; + + assert!(!source_path.exists()); + assert!(compressed_rollout_path(&source_path).exists()); + assert!(unreadable_path.exists()); + Ok(()) +} + #[tokio::test] async fn resume_materializes_compressed_rollout_path() -> anyhow::Result<()> { let home = TempDir::new()?; @@ -637,6 +718,21 @@ fn write_rollout(path: &std::path::Path, thread_id: ThreadId, message: &str) -> Ok(()) } +fn set_history_base(path: &std::path::Path, history_base: HistoryPosition) -> anyhow::Result<()> { + let contents = fs::read_to_string(path)?; + let mut lines = contents.lines(); + let mut head: serde_json::Value = serde_json::from_str(lines.next().expect("session meta"))?; + head["payload"]["history_base"] = serde_json::to_value(history_base)?; + let mut updated = serde_json::to_string(&head)?; + for line in lines { + updated.push('\n'); + updated.push_str(line); + } + updated.push('\n'); + fs::write(path, updated)?; + Ok(()) +} + fn compress_now(path: &std::path::Path) -> anyhow::Result<()> { let compressed_path = compressed_rollout_path(path); let input = fs::File::open(path)?; diff --git a/codex-rs/rollout/src/lib.rs b/codex-rs/rollout/src/lib.rs index 450113218b..ce3ece61be 100644 --- a/codex-rs/rollout/src/lib.rs +++ b/codex-rs/rollout/src/lib.rs @@ -14,6 +14,7 @@ mod persistence_metrics; pub(crate) mod policy; pub(crate) mod recorder; mod reverse_jsonl_scanner; +mod rollout_reference_index; pub(crate) mod search; pub(crate) mod session_index; mod sqlite_metrics; @@ -73,6 +74,7 @@ pub use recorder::RolloutRecorderParams; pub use recorder::append_rollout_item_to_path; pub use reverse_jsonl_scanner::ReverseJsonlScanner; pub use reverse_jsonl_scanner::ScanOutcome; +pub use rollout_reference_index::RolloutReferenceIndex; pub use search::first_rollout_content_match_snippet; pub use search::search_rollout_matches; pub use search::search_rollout_paths; diff --git a/codex-rs/rollout/src/rollout_reference_index.rs b/codex-rs/rollout/src/rollout_reference_index.rs new file mode 100644 index 0000000000..85f2ba6a2e --- /dev/null +++ b/codex-rs/rollout/src/rollout_reference_index.rs @@ -0,0 +1,160 @@ +//! Indexes direct fork references found in local rollout files. + +use std::collections::HashMap; +use std::collections::HashSet; +use std::io; +use std::path::Path; +use std::time::Duration; +use std::time::Instant; + +use codex_protocol::ThreadId; +use codex_protocol::protocol::HistoryPosition; + +use crate::ARCHIVED_SESSIONS_SUBDIR; +use crate::SESSIONS_SUBDIR; +use crate::compression::RolloutFile; + +/// Direct history-base edges discovered from local rollout metadata. +/// +/// This is a physical-rollout index, not a logical lineage resolver. Callers use it to answer +/// cheap inverse-reference questions without each reimplementing rollout discovery. +#[derive(Debug, Default)] +pub struct RolloutReferenceIndex { + history_base_by_thread: HashMap, + reference_counts_by_thread: HashMap, +} + +impl RolloutReferenceIndex { + /// Scans active and archived local rollout metadata without a deadline. + pub async fn scan(codex_home: &Path) -> io::Result { + let Some(index) = Self::scan_with_deadline(codex_home, ScanDeadline::Unlimited).await? + else { + return Err(io::Error::other( + "unlimited rollout reference scan exceeded a deadline", + )); + }; + Ok(index) + } + + /// Scans active and archived local rollout metadata until the worker deadline expires. + /// + /// Returns None instead of a partial index when the deadline expires. + pub(crate) async fn scan_until( + codex_home: &Path, + started_at: Instant, + max_runtime: Duration, + ) -> io::Result> { + Self::scan_with_deadline( + codex_home, + ScanDeadline::Until { + started_at, + max_runtime, + }, + ) + .await + } + + /// Returns how many other discovered rollouts directly reference thread_id. + pub fn reference_count(&self, thread_id: ThreadId) -> usize { + self.reference_counts_by_thread + .get(&thread_id) + .copied() + .unwrap_or_default() + } + + /// Returns the direct history-base edge for thread_id, if one was discovered. + pub fn history_base(&self, thread_id: ThreadId) -> Option<&HistoryPosition> { + self.history_base_by_thread.get(&thread_id) + } + + async fn scan_with_deadline( + codex_home: &Path, + deadline: ScanDeadline, + ) -> io::Result> { + let mut history_base_by_thread = HashMap::new(); + let mut seen_thread_ids = HashSet::new(); + let mut stack = vec![ + codex_home.join(ARCHIVED_SESSIONS_SUBDIR), + codex_home.join(SESSIONS_SUBDIR), + ]; + while let Some(directory) = stack.pop() { + if deadline.expired() { + return Ok(None); + } + let mut entries = match tokio::fs::read_dir(directory.as_path()).await { + Ok(entries) => entries, + Err(err) if err.kind() == io::ErrorKind::NotFound => continue, + Err(err) => return Err(err), + }; + loop { + if deadline.expired() { + return Ok(None); + } + let Some(entry) = entries.next_entry().await? else { + break; + }; + let path = entry.path(); + let file_type = entry.file_type().await?; + if file_type.is_dir() { + stack.push(path); + continue; + } + if !file_type.is_file() { + continue; + } + let Some(rollout_file) = RolloutFile::from_path(path) else { + continue; + }; + let Ok(meta) = crate::read_session_meta_line(rollout_file.path()).await else { + continue; + }; + let thread_id = meta.meta.id; + if !seen_thread_ids.insert(thread_id) { + continue; + } + if let Some(history_base) = meta.meta.history_base { + history_base_by_thread.insert(thread_id, history_base); + } + } + } + + let mut reference_counts_by_thread = HashMap::new(); + for (thread_id, history_base) in &history_base_by_thread { + if history_base.thread_id == *thread_id { + continue; + } + *reference_counts_by_thread + .entry(history_base.thread_id) + .or_default() += 1; + } + Ok(Some(Self { + history_base_by_thread, + reference_counts_by_thread, + })) + } +} + +#[derive(Clone, Copy)] +enum ScanDeadline { + Unlimited, + Until { + started_at: Instant, + max_runtime: Duration, + }, +} + +impl ScanDeadline { + fn expired(self) -> bool { + match self { + Self::Unlimited => false, + Self::Until { + started_at, + max_runtime, + } => started_at.elapsed() >= max_runtime, + } + } +} + +#[cfg(test)] +#[path = "rollout_reference_index_tests.rs"] +mod tests; diff --git a/codex-rs/rollout/src/rollout_reference_index_tests.rs b/codex-rs/rollout/src/rollout_reference_index_tests.rs new file mode 100644 index 0000000000..386308a682 --- /dev/null +++ b/codex-rs/rollout/src/rollout_reference_index_tests.rs @@ -0,0 +1,168 @@ +use std::fs; +use std::path::Path; +use std::path::PathBuf; +use std::time::Duration; +use std::time::Instant; + +use codex_protocol::ThreadId; +use codex_protocol::protocol::HistoryPosition; +use pretty_assertions::assert_eq; +use tempfile::TempDir; +use uuid::Uuid; + +use super::RolloutReferenceIndex; + +#[tokio::test] +async fn scans_active_archived_and_compressed_history_bases() -> anyhow::Result<()> { + let home = TempDir::new()?; + let source_id = thread_id(Uuid::from_u128(1))?; + let active_base = history_position(source_id); + let archived_base = history_position(source_id); + let compressed_base = history_position(source_id); + let active_child_id = thread_id(Uuid::from_u128(2))?; + let archived_child_id = thread_id(Uuid::from_u128(3))?; + let compressed_child_id = thread_id(Uuid::from_u128(4))?; + + write_rollout( + active_rollout_path(home.path(), Uuid::from_u128(2)), + active_child_id, + Some(active_base), + )?; + write_rollout( + archived_rollout_path(home.path(), Uuid::from_u128(3)), + archived_child_id, + Some(archived_base), + )?; + let compressed_path = archived_rollout_path(home.path(), Uuid::from_u128(4)); + write_rollout( + compressed_path.clone(), + compressed_child_id, + Some(compressed_base), + )?; + compress_now(compressed_path.as_path())?; + + let index = RolloutReferenceIndex::scan(home.path()).await?; + + assert_eq!(index.reference_count(source_id), 3); + assert_eq!(index.history_base(active_child_id), Some(&active_base)); + assert_eq!(index.history_base(archived_child_id), Some(&archived_base)); + assert_eq!( + index.history_base(compressed_child_id), + Some(&compressed_base) + ); + Ok(()) +} + +#[tokio::test] +async fn active_duplicate_wins_without_double_counting() -> anyhow::Result<()> { + let home = TempDir::new()?; + let active_source_id = thread_id(Uuid::from_u128(8))?; + let archived_source_id = thread_id(Uuid::from_u128(9))?; + let child_uuid = Uuid::from_u128(10); + let child_id = thread_id(Uuid::from_u128(10))?; + let active_base = history_position(active_source_id); + write_rollout( + active_rollout_path(home.path(), child_uuid), + child_id, + Some(active_base), + )?; + write_rollout( + archived_rollout_path(home.path(), child_uuid), + child_id, + Some(history_position(archived_source_id)), + )?; + + let index = RolloutReferenceIndex::scan(home.path()).await?; + + assert_eq!(index.history_base(child_id), Some(&active_base)); + assert_eq!(index.reference_count(active_source_id), 1); + assert_eq!(index.reference_count(archived_source_id), 0); + Ok(()) +} + +#[tokio::test] +async fn self_history_base_does_not_count_as_reference() -> anyhow::Result<()> { + let home = TempDir::new()?; + let thread_id = thread_id(Uuid::from_u128(11))?; + let history_base = history_position(thread_id); + write_rollout( + active_rollout_path(home.path(), Uuid::from_u128(11)), + thread_id, + Some(history_base), + )?; + + let index = RolloutReferenceIndex::scan(home.path()).await?; + + assert_eq!(index.history_base(thread_id), Some(&history_base)); + assert_eq!(index.reference_count(thread_id), 0); + Ok(()) +} + +#[tokio::test] +async fn expired_deadline_returns_none() -> anyhow::Result<()> { + let home = TempDir::new()?; + + let index = + RolloutReferenceIndex::scan_until(home.path(), Instant::now(), Duration::ZERO).await?; + + assert!(index.is_none()); + Ok(()) +} + +fn active_rollout_path(home: &Path, uuid: Uuid) -> PathBuf { + home.join("sessions/2025/01/03") + .join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl")) +} + +fn archived_rollout_path(home: &Path, uuid: Uuid) -> PathBuf { + home.join("archived_sessions") + .join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl")) +} + +fn write_rollout( + path: PathBuf, + thread_id: ThreadId, + history_base: Option, +) -> anyhow::Result<()> { + fs::create_dir_all(path.parent().expect("rollout parent"))?; + let session_meta = serde_json::json!({ + "timestamp": "2025-01-03T12:00:00Z", + "type": "session_meta", + "payload": { + "id": thread_id, + "timestamp": "2025-01-03T12:00:00Z", + "cwd": path.parent().expect("rollout parent"), + "originator": "test", + "cli_version": "test", + "source": "cli", + "model_provider": "test-provider", + "history_mode": "paginated", + "history_base": history_base, + }, + }); + fs::write(path, format!("{session_meta}\n"))?; + Ok(()) +} + +fn history_position(thread_id: ThreadId) -> HistoryPosition { + HistoryPosition { + thread_id, + end_ordinal_exclusive: 2, + end_byte_offset: 100, + } +} + +fn thread_id(uuid: Uuid) -> anyhow::Result { + Ok(ThreadId::from_string(&uuid.to_string())?) +} + +fn compress_now(path: &Path) -> anyhow::Result<()> { + let compressed_path = path.with_extension("jsonl.zst"); + let input = fs::File::open(path)?; + let output = fs::File::create(compressed_path)?; + let mut encoder = zstd::stream::write::Encoder::new(output, 3)?; + std::io::copy(&mut std::io::BufReader::new(input), &mut encoder)?; + encoder.finish()?; + fs::remove_file(path)?; + Ok(()) +} diff --git a/codex-rs/thread-store/src/lib.rs b/codex-rs/thread-store/src/lib.rs index e218387135..878b0df0e9 100644 --- a/codex-rs/thread-store/src/lib.rs +++ b/codex-rs/thread-store/src/lib.rs @@ -27,6 +27,7 @@ pub use types::ArchiveThreadParams; pub use types::ClearableField; pub use types::CreateThreadParams; pub use types::DeleteThreadParams; +pub use types::DeleteThreadsParams; pub use types::ExtraConfig; pub use types::GitInfoPatch; pub use types::ItemPage; diff --git a/codex-rs/thread-store/src/local/delete_thread.rs b/codex-rs/thread-store/src/local/delete_thread.rs index d7a44b8050..5b3e3ecfde 100644 --- a/codex-rs/thread-store/src/local/delete_thread.rs +++ b/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::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"); + } } diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 576642267b..1b5d6c8ede 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -34,6 +34,7 @@ use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::DeleteThreadParams; +use crate::DeleteThreadsParams; use crate::ItemPage; use crate::ListItemsParams; use crate::ListThreadsParams; @@ -401,6 +402,10 @@ impl ThreadStore for LocalThreadStore { fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreFuture<'_, ()> { Box::pin(async move { delete_thread::delete_thread(self, params).await }) } + + fn delete_threads(&self, params: DeleteThreadsParams) -> ThreadStoreFuture<'_, ()> { + Box::pin(async move { delete_thread::delete_threads(self, params).await }) + } } #[cfg(test)] diff --git a/codex-rs/thread-store/src/store.rs b/codex-rs/thread-store/src/store.rs index d596acd9e8..f8e798f612 100644 --- a/codex-rs/thread-store/src/store.rs +++ b/codex-rs/thread-store/src/store.rs @@ -8,6 +8,7 @@ use crate::AppendThreadItemsParams; use crate::ArchiveThreadParams; use crate::CreateThreadParams; use crate::DeleteThreadParams; +use crate::DeleteThreadsParams; use crate::ItemPage; use crate::ListItemsParams; use crate::ListThreadsParams; @@ -171,4 +172,20 @@ pub trait ThreadStore: Any + Send + Sync { /// Deletes a thread's persisted rollout data and associated metadata. fn delete_thread(&self, params: DeleteThreadParams) -> ThreadStoreFuture<'_, ()>; + + /// Deletes threads 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`]. + fn delete_threads(&self, params: DeleteThreadsParams) -> ThreadStoreFuture<'_, ()> { + Box::pin(async move { + for thread_id in params.thread_ids { + match self.delete_thread(DeleteThreadParams { thread_id }).await { + Ok(()) | Err(ThreadStoreError::ThreadNotFound { .. }) => {} + Err(err) => return Err(err), + } + } + Ok(()) + }) + } } diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index 37b02cb51a..87bcbf755b 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -788,6 +788,13 @@ pub struct DeleteThreadParams { pub thread_id: ThreadId, } +/// Parameters for deleting a set of threads as one store operation. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct DeleteThreadsParams { + /// Thread ids to delete, in the order their persisted data should be removed. + pub thread_ids: Vec, +} + #[cfg(test)] mod tests { use pretty_assertions::assert_eq;