Clear delivery state when archiving threads

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(())
         }

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<_>>(),
+        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<_>>(),
+        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,
+    }
+}

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;

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> {
+        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<_>>(),
+            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<_>>(),
+            vec!["timer-2".to_string()]
+        );
+    }
+}