feat(cron): deduplicate and protect scheduled executions (#601)

## Summary

- add task-level queue protection for scheduled cron executions
- deduplicate scheduled occurrences with a database-backed `(job_id,
scheduled_at)` claim
- add renewable run leases, heartbeat renewal, retry preservation, and
restart recovery
- claim occurrences before conversation preparation/creation
- keep schedule calculation on the theoretical occurrence grid

Closes #600

## Why

Conversation busy state alone cannot prevent duplicate callbacks from
creating two independent conversations. The occurrence claim must happen
at the task/database level before any conversation is materialized.

## Implementation

- add `queue_enabled` to cron jobs
- add `cron_job_runs` with a unique occurrence key and active-run index
- use `BEGIN IMMEDIATE` for atomic cross-process claims
- record duplicate, running, retrying, skipped, success, and error
outcomes
- use owner IDs and expiring leases for crash recovery
- preserve `scheduled_at` through retries and scheduler callbacks
- fence stale owners from updating job state after losing their lease

## Testing

- `RUSTUP_TOOLCHAIN=stable cargo fmt --all -- --check`
- `RUSTUP_TOOLCHAIN=stable cargo check -p aionui-cron -p aionui-db -p
aionui-app`
- `RUSTUP_TOOLCHAIN=stable cargo test -p aionui-cron -p aionui-db -p
aionui-api-types`
- `RUSTUP_TOOLCHAIN=stable cargo test -p aionui-app --test cron_e2e`

All checks pass on the latest upstream `main`.

## Related UI

The corresponding queue-protection UI and visual schedule builder are in
iOfficeAI/AionUi#3552.

---------

Co-authored-by: zk <>

Author: JAVA-LW

crates/aionui-api-types/src/cron.rs

@@ -123,6 +123,7 @@ pub struct CronJobStateDto {
     pub run_count: i64,
     pub retry_count: i64,
     pub max_retries: i64,
+    pub queue_enabled: bool,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -160,6 +161,8 @@ pub struct CreateCronJobRequest {
     #[serde(default)]
     pub execution_mode: Option<String>,
     #[serde(default)]
+    pub queue_enabled: bool,
+    #[serde(default)]
     pub agent_config: Option<CronAgentConfigWriteDto>,
 }
 
@@ -209,6 +212,8 @@ pub struct UpdateCronJobRequest {
     pub conversation_title: Option<String>,
     #[serde(default)]
     pub max_retries: Option<i64>,
+    #[serde(default)]
+    pub queue_enabled: Option<bool>,
 }
 
 // ---------------------------------------------------------------------------
@@ -609,6 +614,7 @@ mod tests {
                 run_count: 5,
                 retry_count: 0,
                 max_retries: 3,
+                queue_enabled: false,
             },
         }
     }
@@ -677,6 +683,7 @@ mod tests {
                 run_count: 0,
                 retry_count: 0,
                 max_retries: 3,
+                queue_enabled: true,
             },
         };
         let json = serde_json::to_value(&resp).unwrap();

crates/aionui-app/src/router/state.rs

@@ -706,11 +706,11 @@ pub fn build_cron_state(services: &AppServices) -> CronRouterState {
     let tick_service_ref: Arc<CronServiceTickRef> = Arc::new(CronServiceTickRef::default());
     let tick_ref = tick_service_ref.clone();
     let scheduler = Arc::new(aionui_cron::scheduler::CronScheduler::new(Arc::new(
-        move |job_id: String| {
+        move |tick: aionui_cron::scheduler::ScheduledTick| {
             let svc = tick_ref.0.lock().unwrap().clone();
             tokio::spawn(async move {
                 if let Some(svc) = svc {
-                    svc.tick(&job_id).await;
+                    svc.tick(&tick.job_id, tick.scheduled_at).await;
                 }
             });
         },

crates/aionui-app/tests/cron_e2e.rs

@@ -402,6 +402,7 @@ async fn cj5b_run_now_legacy_workspace_with_whitespace_succeeds() {
             run_count: 0,
             retry_count: 0,
             max_retries: 3,
+            queue_enabled: false,
         })
         .await
         .unwrap();

crates/aionui-cron/src/artifacts.rs

@@ -125,6 +125,7 @@ mod tests {
             run_count: 0,
             retry_count: 0,
             max_retries: 3,
+            queue_enabled: false,
         }
     }
 

crates/aionui-cron/src/events.rs

@@ -138,6 +138,7 @@ mod tests {
                 run_count: 0,
                 retry_count: 0,
                 max_retries: 3,
+                queue_enabled: false,
             },
         }
     }

crates/aionui-cron/src/executor.rs

@@ -112,6 +112,24 @@ impl JobExecutor {
             return Err(ExecutionResult::Error { message: e.to_string() });
         }
 
+        if job.queue_enabled && job.execution_mode == ExecutionMode::NewConversation {
+            match self.find_active_cron_conversation(job).await {
+                Ok(Some(conversation_id)) => {
+                    info!(
+                        job_id = %job.id,
+                        conversation_id,
+                        "Cron queue is enabled and a previous execution is still active; skipping trigger"
+                    );
+                    return Err(ExecutionResult::Skipped);
+                }
+                Ok(None) => {}
+                Err(e) => {
+                    error!(job_id = %job.id, error = %e, "Failed to inspect previous cron executions");
+                    return Err(ExecutionResult::Error { message: e.to_string() });
+                }
+            }
+        }
+
         let conversation_id = match self.resolve_conversation(job, saved_skill.as_ref()).await {
             Ok(id) => id,
             Err(e) => {
@@ -444,6 +462,11 @@ impl JobExecutor {
 
 impl JobExecutor {
     fn handle_busy(&self, job: &CronJob) -> ExecutionResult {
+        if job.queue_enabled {
+            info!(job_id = %job.id, "Cron queue is enabled; skipping overlapping execution");
+            return ExecutionResult::Skipped;
+        }
+
         let max_retries = job.max_retries;
         let current_retry = job.retry_count;
 
@@ -504,6 +527,15 @@ impl JobExecutor {
         }
     }
 
+    async fn find_active_cron_conversation(&self, job: &CronJob) -> Result<Option<String>, CronError> {
+        let user_id = self.resolve_conversation_owner_user_id(job).await?;
+        let conversations = self.conversation_repo.list_by_cron_job(&user_id, &job.id).await?;
+        Ok(conversations
+            .into_iter()
+            .find(|conversation| self.is_conversation_claimed(&conversation.id))
+            .map(|conversation| conversation.id))
+    }
+
     async fn create_new_conversation(
         &self,
         job: &CronJob,
@@ -1243,6 +1275,7 @@ mod tests {
             run_count: 0,
             retry_count: 0,
             max_retries: 3,
+            queue_enabled: false,
         }
     }
 
@@ -1268,6 +1301,7 @@ mod tests {
         let job = CronJob {
             retry_count: 1,
             max_retries: 3,
+            queue_enabled: false,
             ..sample_job()
         };
         let result = executor.handle_busy(&job);
@@ -1281,6 +1315,7 @@ mod tests {
         let job = CronJob {
             retry_count: 3,
             max_retries: 3,
+            queue_enabled: false,
             ..sample_job()
         };
         let result = executor.handle_busy(&job);
@@ -1294,6 +1329,7 @@ mod tests {
         let job = CronJob {
             retry_count: 5,
             max_retries: 3,
+            queue_enabled: false,
             ..sample_job()
         };
         let result = executor.handle_busy(&job);
@@ -1307,12 +1343,26 @@ mod tests {
         let job = CronJob {
             retry_count: 0,
             max_retries: 3,
+            queue_enabled: false,
             ..sample_job()
         };
         let result = executor.handle_busy(&job);
         assert_eq!(result, ExecutionResult::Retrying { attempt: 1 });
     }
 
+    #[tokio::test]
+    async fn handle_busy_returns_skipped_when_queue_is_enabled() {
+        let executor = make_executor_for_busy_tests();
+        let job = CronJob {
+            retry_count: 0,
+            max_retries: 3,
+            queue_enabled: true,
+            ..sample_job()
+        };
+
+        assert_eq!(executor.handle_busy(&job), ExecutionResult::Skipped);
+    }
+
     #[tokio::test]
     async fn execute_returns_retrying_when_runtime_state_is_already_claimed() {
         let agent = Arc::new(RecordingAgent::new("conv_1", "default", true));
@@ -1332,6 +1382,27 @@ mod tests {
         drop(claim);
     }
 
+    #[tokio::test]
+    async fn execute_returns_skipped_when_queue_is_enabled_and_conversation_is_claimed() {
+        let agent = Arc::new(RecordingAgent::new("conv_1", "default", true));
+        let executor = make_executor_with_agent(AgentInstance::Mock(agent.clone()));
+        let job = CronJob {
+            queue_enabled: true,
+            ..sample_job()
+        };
+        let claim = executor
+            .conversation_service
+            .runtime_state()
+            .try_claim_turn(&job.conversation_id, "turn-existing")
+            .expect("runtime claim should succeed");
+
+        let result = executor.execute(&job).await;
+
+        assert_eq!(result, ExecutionResult::Skipped);
+        assert_eq!(agent.send_calls(), 0, "queue protection should avoid send attempts");
+        drop(claim);
+    }
+
     #[tokio::test]
     async fn prepare_run_now_returns_active_conversation_when_runtime_state_is_already_claimed() {
         let agent = Arc::new(RecordingAgent::new("conv_1", "default", true));