diff --git a/codex-rs/core/src/codex/timer_runtime.rs b/codex-rs/core/src/codex/timer_runtime.rs index 0a1686751e..adac1929c1 100644 --- a/codex-rs/core/src/codex/timer_runtime.rs +++ b/codex-rs/core/src/codex/timer_runtime.rs @@ -204,6 +204,7 @@ impl Session { timer, context, deleted_one_shot_timer, + .. }) = self.claim_next_timer_for_delivery().await else { return false; @@ -531,7 +532,7 @@ impl Session { "failed to claim timer {} in sqlite: {err}", claimed.timer.id ); - return true; + return false; } }; let thread_id = self.thread_id_string(); @@ -548,7 +549,12 @@ impl Session { return false; }; state_db - .claim_recurring_thread_timer(&thread_id, &claimed.timer.id, ¶ms) + .claim_recurring_thread_timer( + &thread_id, + &claimed.timer.id, + claimed.previous_last_run_at, + ¶ms, + ) .await }; match result { diff --git a/codex-rs/core/src/timers.rs b/codex-rs/core/src/timers.rs index 98f4ecc0da..cc7e59903b 100644 --- a/codex-rs/core/src/timers.rs +++ b/codex-rs/core/src/timers.rs @@ -80,6 +80,7 @@ pub(crate) struct ClaimedTimer { pub(crate) timer: ThreadTimer, pub(crate) context: TimerInvocationContext, pub(crate) deleted_one_shot_timer: bool, + pub(crate) previous_last_run_at: Option, } #[derive(Debug)] @@ -397,12 +398,17 @@ impl TimersState { } = runtime; let is_recurring = timer.trigger.is_recurring(); let deleted_one_shot_timer = !is_recurring; + let previous_last_run_at = timer.last_run_at; if deleted_one_shot_timer { if let Some(cancel) = timer_cancel.as_ref() { cancel.cancel(); } } else { - timer.last_run_at = Some(now.timestamp()); + timer.last_run_at = Some( + previous_last_run_at + .map(|previous| now.timestamp().max(previous.saturating_add(1))) + .unwrap_or_else(|| now.timestamp()), + ); let pending_run = timer.trigger.is_idle_recurring(); self.timers.insert( timer.id.clone(), @@ -425,6 +431,7 @@ impl TimersState { queued_at: now.timestamp(), }, deleted_one_shot_timer, + previous_last_run_at, }) } } diff --git a/codex-rs/state/src/runtime/timers.rs b/codex-rs/state/src/runtime/timers.rs index ea9cf00a74..c89ecc597b 100644 --- a/codex-rs/state/src/runtime/timers.rs +++ b/codex-rs/state/src/runtime/timers.rs @@ -158,6 +158,7 @@ WHERE thread_id = ? &self, thread_id: &str, id: &str, + expected_last_run_at: Option, params: &ThreadTimerUpdateParams, ) -> anyhow::Result { let result = sqlx::query( @@ -174,6 +175,10 @@ SET trigger_json = ?, WHERE thread_id = ? AND id = ? AND pending_run = 1 + AND ( + (last_run_at IS NULL AND ? IS NULL) + OR last_run_at = ? + ) "#, ) .bind(params.trigger_json.as_str()) @@ -186,6 +191,8 @@ WHERE thread_id = ? .bind(i64::from(params.pending_run)) .bind(thread_id) .bind(id) + .bind(expected_last_run_at) + .bind(expected_last_run_at) .execute(self.pool.as_ref()) .await?; Ok(result.rows_affected() > 0) @@ -374,13 +381,13 @@ ORDER BY name assert!( runtime - .claim_recurring_thread_timer("thread-1", "timer-1", &update) + .claim_recurring_thread_timer("thread-1", "timer-1", None, &update) .await .expect("claim recurring timer") ); assert!( !runtime - .claim_recurring_thread_timer("thread-1", "timer-1", &update) + .claim_recurring_thread_timer("thread-1", "timer-1", None, &update) .await .expect("claim recurring timer again") ); @@ -397,4 +404,46 @@ ORDER BY name assert_eq!(timers[0].last_run_at, Some(110)); assert!(!timers[0].pending_run); } + + #[tokio::test] + async fn recurring_idle_claim_rejects_stale_last_run_at_even_when_pending_stays_true() { + let runtime = test_runtime().await; + let mut params = timer_params("timer-1", "thread-1"); + params.pending_run = true; + params.last_run_at = Some(100); + runtime + .create_thread_timer(¶ms) + .await + .expect("create pending timer"); + let update = ThreadTimerUpdateParams { + trigger_json: params.trigger_json.clone(), + content: params.content.clone(), + instructions: params.instructions.clone(), + meta_json: params.meta_json.clone(), + delivery: params.delivery.clone(), + next_run_at: Some(120), + last_run_at: Some(110), + pending_run: true, + }; + + assert!( + runtime + .claim_recurring_thread_timer("thread-1", "timer-1", Some(100), &update) + .await + .expect("claim recurring idle timer") + ); + assert!( + !runtime + .claim_recurring_thread_timer("thread-1", "timer-1", Some(100), &update) + .await + .expect("claim recurring idle timer again") + ); + let timers = runtime + .list_thread_timers("thread-1") + .await + .expect("list timers"); + assert_eq!(timers.len(), 1); + assert_eq!(timers[0].last_run_at, Some(110)); + assert!(timers[0].pending_run); + } }