diff --git a/codex-rs/cli/src/queue_cmd.rs b/codex-rs/cli/src/queue_cmd.rs index 77b22a9f9b..8bcce96f1b 100644 --- a/codex-rs/cli/src/queue_cmd.rs +++ b/codex-rs/cli/src/queue_cmd.rs @@ -81,6 +81,13 @@ pub(crate) async fn run_queue_command( }) .map_err(anyhow::Error::msg)?; state_db.create_thread_timer(&timer_params).await?; + remove_queued_work_if_thread_missing( + config.codex_home.as_path(), + &state_db, + &timer_params.thread_id, + QueuedWork::Timer(&timer_params.id), + ) + .await?; println!( "{}", queue_timer_success_message( @@ -103,6 +110,13 @@ pub(crate) async fn run_queue_command( unix_timestamp_now()?, ); state_db.create_thread_message(&message_params).await?; + remove_queued_work_if_thread_missing( + config.codex_home.as_path(), + &state_db, + &message_params.thread_id, + QueuedWork::Message(&message_params.id), + ) + .await?; println!( "Queued message {} for thread {}.", message_params.id, message_params.thread_id @@ -110,6 +124,35 @@ pub(crate) async fn run_queue_command( Ok(()) } +enum QueuedWork<'a> { + Timer(&'a str), + Message(&'a str), +} + +async fn remove_queued_work_if_thread_missing( + codex_home: &Path, + state_db: &StateRuntime, + thread_id: &str, + queued_work: QueuedWork<'_>, +) -> anyhow::Result<()> { + if codex_core::find_thread_path_by_id_str(codex_home, thread_id) + .await? + .is_some() + { + return Ok(()); + } + + match queued_work { + QueuedWork::Timer(id) => { + state_db.delete_thread_timer(thread_id, id).await?; + } + QueuedWork::Message(id) => { + state_db.delete_thread_message(thread_id, id).await?; + } + } + anyhow::bail!("thread `{thread_id}` was archived before queued work could be created"); +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum QueueMode { ImmediateMessage, @@ -463,4 +506,51 @@ mod tests { "no thread named `stale`" ); } + + #[tokio::test] + async fn queue_cleanup_removes_message_when_thread_disappears() { + let codex_home = TempDir::new().expect("codex home tempdir"); + let sqlite_home = TempDir::new().expect("sqlite home tempdir"); + let runtime = StateRuntime::init( + sqlite_home.path().to_path_buf(), + "test-provider".to_string(), + ) + .await + .expect("initialize state runtime"); + let thread_id = ThreadId::new().to_string(); + let params = codex_state::ThreadMessageCreateParams::new( + thread_id.clone(), + "external".to_string(), + "do work".to_string(), + /*instructions*/ None, + "{}".to_string(), + TimerDelivery::AfterTurn.as_str().to_string(), + /*queued_at*/ 100, + ); + runtime + .create_thread_message(¶ms) + .await + .expect("create message"); + + let err = remove_queued_work_if_thread_missing( + codex_home.path(), + &runtime, + &thread_id, + QueuedWork::Message(¶ms.id), + ) + .await + .expect_err("missing thread should fail after cleanup"); + + assert_eq!( + err.to_string(), + format!("thread `{thread_id}` was archived before queued work could be created") + ); + assert_eq!( + runtime + .list_thread_messages(&thread_id) + .await + .expect("list messages"), + Vec::new() + ); + } } diff --git a/codex-rs/state/src/runtime/messages.rs b/codex-rs/state/src/runtime/messages.rs index 52b10522b2..67cbbdfd33 100644 --- a/codex-rs/state/src/runtime/messages.rs +++ b/codex-rs/state/src/runtime/messages.rs @@ -70,6 +70,15 @@ ORDER BY queued_at ASC, seq ASC Ok(rows.into_iter().map(ThreadMessage::from).collect()) } + pub async fn delete_thread_message(&self, thread_id: &str, id: &str) -> anyhow::Result { + let result = sqlx::query("DELETE FROM thread_messages WHERE thread_id = ? AND id = ?") + .bind(thread_id) + .bind(id) + .execute(self.pool.as_ref()) + .await?; + Ok(result.rows_affected() > 0) + } + pub async fn claim_next_thread_message( &self, thread_id: &str, @@ -217,6 +226,55 @@ ORDER BY name assert_eq!(message.queued_at, params.queued_at); } + #[tokio::test] + async fn delete_thread_message_is_scoped_to_thread_id() { + let runtime = test_runtime().await; + runtime + .create_thread_message(&message_params( + "message-1", + "thread-1", + /*queued_at*/ 100, + )) + .await + .expect("create thread-1 message"); + runtime + .create_thread_message(&message_params( + "message-2", + "thread-2", + /*queued_at*/ 100, + )) + .await + .expect("create thread-2 message"); + + let deleted_wrong_thread = runtime + .delete_thread_message("thread-2", "message-1") + .await + .expect("delete wrong-thread message"); + assert!(!deleted_wrong_thread); + let deleted = runtime + .delete_thread_message("thread-1", "message-1") + .await + .expect("delete thread-1 message"); + assert!(deleted); + assert_eq!( + runtime + .list_thread_messages("thread-1") + .await + .expect("list thread-1 messages"), + 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()] + ); + } + #[tokio::test] async fn claim_is_scoped_to_thread_id_and_ordered() { let runtime = test_runtime().await;