diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 53f4a876c2..26e9031e7c 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -5708,6 +5708,10 @@ impl CodexMessageProcessor { let _ = ctx .mark_archived(thread_id, archived_path.as_path(), Utc::now()) .await; + let thread_id_str = thread_id.to_string(); + if let Err(err) = ctx.delete_thread_delivery_state(&thread_id_str).await { + warn!("failed to delete delivery state for archived thread {thread_id}: {err}"); + } } Ok(()) } 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 20fd6fd884..0235a9a133 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_archive.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_archive.rs @@ -20,6 +20,9 @@ use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::UserInput; use codex_core::ARCHIVED_SESSIONS_SUBDIR; use codex_core::find_thread_path_by_id_str; +use codex_state::StateRuntime; +use codex_state::ThreadMessageCreateParams; +use codex_state::ThreadTimerCreateParams; use pretty_assertions::assert_eq; use std::path::Path; use tempfile::TempDir; @@ -118,6 +121,26 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { .expect("expected rollout path for thread id to exist after materialization"); assert_paths_match_on_disk(&discovered_path, &rollout_path)?; + let state_db = StateRuntime::init(codex_home.path().to_path_buf(), "mock_provider".to_string()) + .await + .expect("initialize state db"); + state_db + .create_thread_message(&message_params("message-1", &thread.id)) + .await + .expect("create archived thread message"); + state_db + .create_thread_message(&message_params("message-2", "other-thread")) + .await + .expect("create other thread message"); + state_db + .create_thread_timer(&timer_params("timer-1", &thread.id)) + .await + .expect("create archived thread timer"); + state_db + .create_thread_timer(&timer_params("timer-2", "other-thread")) + .await + .expect("create other thread timer"); + let archive_id = mcp .send_thread_archive_request(ThreadArchiveParams { thread_id: thread.id.clone(), @@ -156,6 +179,40 @@ async fn thread_archive_requires_materialized_rollout() -> Result<()> { "expected archived rollout path {} to exist", archived_rollout_path.display() ); + assert_eq!( + state_db + .list_thread_messages(&thread.id) + .await + .expect("list archived thread messages"), + Vec::new() + ); + assert_eq!( + state_db + .list_thread_timers(&thread.id) + .await + .expect("list archived thread timers"), + Vec::new() + ); + assert_eq!( + state_db + .list_thread_messages("other-thread") + .await + .expect("list other thread messages") + .into_iter() + .map(|message| message.id) + .collect::>(), + vec!["message-2".to_string()] + ); + assert_eq!( + state_db + .list_thread_timers("other-thread") + .await + .expect("list other thread timers") + .into_iter() + .map(|timer| timer.id) + .collect::>(), + vec!["timer-2".to_string()] + ); Ok(()) } @@ -323,3 +380,34 @@ fn assert_paths_match_on_disk(actual: &Path, expected: &Path) -> std::io::Result assert_eq!(actual, expected); Ok(()) } + +fn message_params(id: &str, thread_id: &str) -> ThreadMessageCreateParams { + ThreadMessageCreateParams { + id: id.to_string(), + thread_id: thread_id.to_string(), + source: "external".to_string(), + content: "do something".to_string(), + instructions: None, + meta_json: "{}".to_string(), + delivery: "after-turn".to_string(), + queued_at: 100, + } +} + +fn timer_params(id: &str, thread_id: &str) -> ThreadTimerCreateParams { + ThreadTimerCreateParams { + id: id.to_string(), + thread_id: thread_id.to_string(), + source: "agent".to_string(), + client_id: "codex-tui".to_string(), + trigger_json: r#"{"kind":"delay","seconds":10,"repeat":false}"#.to_string(), + content: "run tests".to_string(), + instructions: None, + meta_json: "{}".to_string(), + delivery: "after-turn".to_string(), + created_at: 100, + next_run_at: Some(110), + last_run_at: None, + pending_run: false, + } +} diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 43da16ae07..bf494bf6a5 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -58,6 +58,7 @@ use tracing::warn; mod agent_jobs; mod backfill; +mod delivery_state; mod logs; mod memories; mod messages; diff --git a/codex-rs/state/src/runtime/delivery_state.rs b/codex-rs/state/src/runtime/delivery_state.rs new file mode 100644 index 0000000000..a78f821d87 --- /dev/null +++ b/codex-rs/state/src/runtime/delivery_state.rs @@ -0,0 +1,131 @@ +//! Cleanup operations for per-thread delivery state. +//! +//! Timers and queued messages are stored independently because they have +//! different runtime behavior, but thread lifecycle operations need to treat +//! them as one unit. This module owns that cross-table cleanup. + +use super::*; + +impl StateRuntime { + /// Delete all queued messages and timers associated with `thread_id`. + pub async fn delete_thread_delivery_state(&self, thread_id: &str) -> anyhow::Result<()> { + let mut tx = self.pool.begin().await?; + sqlx::query("DELETE FROM thread_messages WHERE thread_id = ?") + .bind(thread_id) + .execute(&mut *tx) + .await?; + sqlx::query("DELETE FROM thread_timers WHERE thread_id = ?") + .bind(thread_id) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::StateRuntime; + use super::test_support::unique_temp_dir; + use crate::ThreadMessageCreateParams; + use crate::ThreadTimerCreateParams; + use pretty_assertions::assert_eq; + + fn message_params(id: &str, thread_id: &str) -> ThreadMessageCreateParams { + ThreadMessageCreateParams { + id: id.to_string(), + thread_id: thread_id.to_string(), + source: "external".to_string(), + content: "do something".to_string(), + instructions: None, + meta_json: "{}".to_string(), + delivery: "after-turn".to_string(), + queued_at: 100, + } + } + + fn timer_params(id: &str, thread_id: &str) -> ThreadTimerCreateParams { + ThreadTimerCreateParams { + id: id.to_string(), + thread_id: thread_id.to_string(), + source: "agent".to_string(), + client_id: "codex-tui".to_string(), + trigger_json: r#"{"kind":"delay","seconds":10,"repeat":false}"#.to_string(), + content: "run tests".to_string(), + instructions: None, + meta_json: "{}".to_string(), + delivery: "after-turn".to_string(), + created_at: 100, + next_run_at: Some(110), + last_run_at: None, + pending_run: false, + } + } + + async fn test_runtime() -> std::sync::Arc { + StateRuntime::init(unique_temp_dir(), "test-provider".to_string()) + .await + .expect("initialize runtime") + } + + #[tokio::test] + async fn delete_thread_delivery_state_removes_messages_and_timers_for_thread() { + let runtime = test_runtime().await; + runtime + .create_thread_message(&message_params("message-1", "thread-1")) + .await + .expect("create thread-1 message"); + runtime + .create_thread_message(&message_params("message-2", "thread-2")) + .await + .expect("create thread-2 message"); + runtime + .create_thread_timer(&timer_params("timer-1", "thread-1")) + .await + .expect("create thread-1 timer"); + runtime + .create_thread_timer(&timer_params("timer-2", "thread-2")) + .await + .expect("create thread-2 timer"); + + runtime + .delete_thread_delivery_state("thread-1") + .await + .expect("delete delivery state"); + + assert_eq!( + runtime + .list_thread_messages("thread-1") + .await + .expect("list thread-1 messages"), + Vec::new() + ); + assert_eq!( + runtime + .list_thread_timers("thread-1") + .await + .expect("list thread-1 timers"), + Vec::new() + ); + assert_eq!( + runtime + .list_thread_messages("thread-2") + .await + .expect("list thread-2 messages") + .into_iter() + .map(|message| message.id) + .collect::>(), + vec!["message-2".to_string()] + ); + assert_eq!( + runtime + .list_thread_timers("thread-2") + .await + .expect("list thread-2 timers") + .into_iter() + .map(|timer| timer.id) + .collect::>(), + vec!["timer-2".to_string()] + ); + } +}