codex: clean up queue rows for archived threads (#17380)

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(&params)
+            .await
+            .expect("create message");
+
+        let err = remove_queued_work_if_thread_missing(
+            codex_home.path(),
+            &runtime,
+            &thread_id,
+            QueuedWork::Message(&params.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()
+        );
+    }
 }

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<bool> {
+        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<_>>(),
+            vec!["message-2".to_string()]
+        );
+    }
+
     #[tokio::test]
     async fn claim_is_scoped_to_thread_id_and_ordered() {
         let runtime = test_runtime().await;