codex: address PR review feedback (#16282)

codex-rs/core/src/codex.rs

@@ -818,6 +818,9 @@ pub(crate) struct Session {
     pending_mcp_server_refresh_config: Mutex<Option<McpServerRefreshConfig>>,
     pub(crate) conversation: Arc<RealtimeConversationManager>,
     pub(crate) active_turn: Mutex<Option<ActiveTurn>>,
+    /// Prevents concurrent job timers from claiming multiple jobs before a
+    /// newly started turn becomes the active turn.
+    job_start_in_progress: Mutex<bool>,
     mailbox: Mailbox,
     mailbox_rx: Mutex<MailboxReceiver>,
     idle_pending_input: Mutex<Vec<ResponseInputItem>>, // TODO (jif) merge with mailbox!
@@ -1943,6 +1946,7 @@ impl Session {
             pending_mcp_server_refresh_config: Mutex::new(None),
             conversation: Arc::new(RealtimeConversationManager::new()),
             active_turn: Mutex::new(None),
+            job_start_in_progress: Mutex::new(false),
             mailbox,
             mailbox_rx: Mutex::new(mailbox_rx),
             idle_pending_input: Mutex::new(Vec::new()),
@@ -2697,19 +2701,27 @@ impl Session {
     }
 
     pub(crate) async fn maybe_start_pending_job(self: &Arc<Self>) {
-        if self.active_turn.lock().await.is_some() {
-            return;
-        }
-        if self.has_queued_response_items_for_next_turn().await
-            || self.has_trigger_turn_mailbox_items().await
-        {
-            return;
-        }
         let Some(ClaimedJob {
             job,
             context,
             deleted_run_once_job,
-        }) = self.jobs.lock().await.claim_next_job(Utc::now())
+        }) = ({
+            let mut job_start_in_progress = self.job_start_in_progress.lock().await;
+            let has_pending_turn_inputs = self.has_queued_response_items_for_next_turn().await
+                || self.has_trigger_turn_mailbox_items().await;
+            if *job_start_in_progress
+                || self.active_turn.lock().await.is_some()
+                || has_pending_turn_inputs
+            {
+                None
+            } else {
+                let claimed = self.jobs.lock().await.claim_next_job(Utc::now());
+                if claimed.is_some() {
+                    *job_start_in_progress = true;
+                }
+                claimed
+            }
+        })
         else {
             return;
         };
@@ -2725,15 +2737,16 @@ impl Session {
         self.maybe_emit_unknown_model_warning_for_turn(turn_context.as_ref())
             .await;
         let session = Arc::clone(self);
-        if let Err(err) = tokio::task::spawn_blocking(move || {
+        let result = tokio::task::spawn_blocking(move || {
             futures::executor::block_on(session.start_task(
                 turn_context,
                 Vec::new(),
                 crate::tasks::RegularTask::new(),
             ));
         })
-        .await
-        {
+        .await;
+        *self.job_start_in_progress.lock().await = false;
+        if let Err(err) = result {
             warn!("failed to start job turn: {err}");
         }
     }

codex-rs/core/src/codex_tests.rs

@@ -2749,6 +2749,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
         pending_mcp_server_refresh_config: Mutex::new(None),
         conversation: Arc::new(RealtimeConversationManager::new()),
         active_turn: Mutex::new(None),
+        job_start_in_progress: Mutex::new(false),
         mailbox,
         mailbox_rx: Mutex::new(mailbox_rx),
         idle_pending_input: Mutex::new(Vec::new()),
@@ -3594,6 +3595,7 @@ pub(crate) async fn make_session_and_context_with_dynamic_tools_and_rx(
         pending_mcp_server_refresh_config: Mutex::new(None),
         conversation: Arc::new(RealtimeConversationManager::new()),
         active_turn: Mutex::new(None),
+        job_start_in_progress: Mutex::new(false),
         mailbox,
         mailbox_rx: Mutex::new(mailbox_rx),
         idle_pending_input: Mutex::new(Vec::new()),
@@ -3628,6 +3630,46 @@ async fn dropping_session_cancels_job_timers() {
     assert!(cancel_token.is_cancelled());
 }
 
+#[tokio::test]
+async fn maybe_start_pending_job_claims_only_one_job_while_start_is_in_progress() {
+    let (session, _, _) = make_session_and_context_with_rx().await;
+    let now = chrono::Utc::now();
+    {
+        let mut jobs = session.jobs.lock().await;
+        jobs.create_job(
+            "job-1".to_string(),
+            crate::jobs::AFTER_TURN_CRON_EXPRESSION.to_string(),
+            "first".to_string(),
+            /*run_once*/ false,
+            now,
+            /*timer_cancel*/ None,
+        )
+        .expect("first job should be created");
+        jobs.create_job(
+            "job-2".to_string(),
+            crate::jobs::AFTER_TURN_CRON_EXPRESSION.to_string(),
+            "second".to_string(),
+            /*run_once*/ false,
+            now,
+            /*timer_cancel*/ None,
+        )
+        .expect("second job should be created");
+    }
+
+    let first = Arc::clone(&session);
+    let second = Arc::clone(&session);
+    tokio::join!(
+        first.maybe_start_pending_job(),
+        second.maybe_start_pending_job()
+    );
+
+    let jobs = session.jobs.lock().await.list_jobs();
+    assert_eq!(
+        jobs.iter().filter(|job| job.last_run_at.is_some()).count(),
+        1
+    );
+}
+
 #[tokio::test]
 async fn refresh_mcp_servers_is_deferred_until_next_turn() {
     let (session, turn_context) = make_session_and_context().await;

codex-rs/core/src/jobs.rs

@@ -20,6 +20,7 @@ const EVERY_SECONDS_PREFIX: &str = "@every:";
 pub const JOB_UPDATED_BACKGROUND_EVENT_PREFIX: &str = "job_updated:";
 pub const JOB_FIRED_BACKGROUND_EVENT_PREFIX: &str = "job_fired:";
 pub const MAX_ACTIVE_JOBS_PER_THREAD: usize = 256;
+const MAX_EVERY_SECONDS: u64 = i64::MAX as u64;
 const ONE_SHOT_JOB_TURN_INSTRUCTIONS: &str =
     include_str!("../templates/jobs/one_shot_turn_instructions.md");
 const RECURRING_JOB_TURN_INSTRUCTIONS: &str =
@@ -83,7 +84,7 @@ impl JobSchedule {
             .map(str::trim)
             .and_then(parse_duration_literal)
         {
-            return Ok(Self::EverySeconds(seconds));
+            return Self::parse_every_seconds(seconds, cron_expression);
         }
 
         if let Some(seconds) = cron_expression
@@ -91,7 +92,7 @@ impl JobSchedule {
             .and_then(|raw| raw.parse::<u64>().ok())
             .filter(|seconds| *seconds > 0)
         {
-            return Ok(Self::EverySeconds(seconds));
+            return Self::parse_every_seconds(seconds, cron_expression);
         }
 
         Err(format!(
@@ -99,6 +100,15 @@ impl JobSchedule {
         ))
     }
 
+    fn parse_every_seconds(seconds: u64, cron_expression: &str) -> Result<Self, String> {
+        if seconds > MAX_EVERY_SECONDS {
+            return Err(format!(
+                "unsupported cron_expression `{cron_expression}`; @every values must be between 1 and {MAX_EVERY_SECONDS} seconds"
+            ));
+        }
+        Ok(Self::EverySeconds(seconds))
+    }
+
     fn next_run_at(self, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
         match self {
             Self::AfterTurn => None,
@@ -293,6 +303,7 @@ mod tests {
     use super::JobTurnContext;
     use super::JobsState;
     use super::MAX_ACTIVE_JOBS_PER_THREAD;
+    use super::MAX_EVERY_SECONDS;
     use super::job_prompt_input_item;
     use chrono::TimeZone;
     use chrono::Utc;
@@ -316,6 +327,23 @@ mod tests {
         );
     }
 
+    #[test]
+    fn rejects_overflowing_every_job_schedules() {
+        let too_large = MAX_EVERY_SECONDS + 1;
+        assert_eq!(
+            JobSchedule::parse(&format!("@every:{too_large}")),
+            Err(format!(
+                "unsupported cron_expression `@every:{too_large}`; @every values must be between 1 and {MAX_EVERY_SECONDS} seconds"
+            ))
+        );
+        assert_eq!(
+            JobSchedule::parse(&format!("@every {too_large}s")),
+            Err(format!(
+                "unsupported cron_expression `@every {too_large}s`; @every values must be between 1 and {MAX_EVERY_SECONDS} seconds"
+            ))
+        );
+    }
+
     #[test]
     fn claim_run_once_job_removes_it() {
         let now = Utc.timestamp_opt(100, 0).single().expect("valid timestamp");