feat: add pause/unpause functionality for scheduled jobs (#2698)

crates/goose-cli/src/commands/schedule.rs

@@ -33,6 +33,7 @@ pub async fn handle_schedule_add(
         cron,
         last_run: None,
         currently_running: false,
+        paused: false,
     };
 
     let scheduler_storage_path =

crates/goose-server/src/openapi.rs

@@ -42,6 +42,8 @@ use utoipa::OpenApi;
         super::routes::schedule::list_schedules,
         super::routes::schedule::delete_schedule,
         super::routes::schedule::run_now_handler,
+        super::routes::schedule::pause_schedule,
+        super::routes::schedule::unpause_schedule,
         super::routes::schedule::sessions_handler
     ),
     components(schemas(

crates/goose-server/src/routes/schedule.rs

@@ -94,6 +94,7 @@ async fn create_schedule(
         cron: req.cron,
         last_run: None,
         currently_running: false,
+        paused: false,
     };
     scheduler
         .add_scheduled_job(job.clone())
@@ -260,12 +261,86 @@ async fn sessions_handler(
     }
 }
 
+#[utoipa::path(
+    post,
+    path = "/schedule/{id}/pause",
+    params(
+        ("id" = String, Path, description = "ID of the schedule to pause")
+    ),
+    responses(
+        (status = 204, description = "Scheduled job paused successfully"),
+        (status = 404, description = "Scheduled job not found"),
+        (status = 400, description = "Cannot pause a currently running job"),
+        (status = 500, description = "Internal server error")
+    ),
+    tag = "schedule"
+)]
+#[axum::debug_handler]
+async fn pause_schedule(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    Path(id): Path<String>,
+) -> Result<StatusCode, StatusCode> {
+    verify_secret_key(&headers, &state)?;
+    let scheduler = state
+        .scheduler()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    scheduler.pause_schedule(&id).await.map_err(|e| {
+        eprintln!("Error pausing schedule '{}': {:?}", id, e);
+        match e {
+            goose::scheduler::SchedulerError::JobNotFound(_) => StatusCode::NOT_FOUND,
+            goose::scheduler::SchedulerError::AnyhowError(_) => StatusCode::BAD_REQUEST,
+            _ => StatusCode::INTERNAL_SERVER_ERROR,
+        }
+    })?;
+    Ok(StatusCode::NO_CONTENT)
+}
+
+#[utoipa::path(
+    post,
+    path = "/schedule/{id}/unpause",
+    params(
+        ("id" = String, Path, description = "ID of the schedule to unpause")
+    ),
+    responses(
+        (status = 204, description = "Scheduled job unpaused successfully"),
+        (status = 404, description = "Scheduled job not found"),
+        (status = 500, description = "Internal server error")
+    ),
+    tag = "schedule"
+)]
+#[axum::debug_handler]
+async fn unpause_schedule(
+    State(state): State<Arc<AppState>>,
+    headers: HeaderMap,
+    Path(id): Path<String>,
+) -> Result<StatusCode, StatusCode> {
+    verify_secret_key(&headers, &state)?;
+    let scheduler = state
+        .scheduler()
+        .await
+        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
+
+    scheduler.unpause_schedule(&id).await.map_err(|e| {
+        eprintln!("Error unpausing schedule '{}': {:?}", id, e);
+        match e {
+            goose::scheduler::SchedulerError::JobNotFound(_) => StatusCode::NOT_FOUND,
+            _ => StatusCode::INTERNAL_SERVER_ERROR,
+        }
+    })?;
+    Ok(StatusCode::NO_CONTENT)
+}
+
 pub fn routes(state: Arc<AppState>) -> Router {
     Router::new()
         .route("/schedule/create", post(create_schedule))
         .route("/schedule/list", get(list_schedules))
         .route("/schedule/delete/{id}", delete(delete_schedule)) // Corrected
         .route("/schedule/{id}/run_now", post(run_now_handler)) // Corrected
+        .route("/schedule/{id}/pause", post(pause_schedule))
+        .route("/schedule/{id}/unpause", post(unpause_schedule))
         .route("/schedule/{id}/sessions", get(sessions_handler)) // Corrected
         .with_state(state)
 }

crates/goose/src/scheduler.rs

@@ -109,6 +109,8 @@ pub struct ScheduledJob {
     pub last_run: Option<DateTime<Utc>>,
     #[serde(default)]
     pub currently_running: bool,
+    #[serde(default)]
+    pub paused: bool,
 }
 
 async fn persist_jobs_from_arc(
@@ -219,6 +221,21 @@ impl Scheduler {
             let job_to_execute = job_for_task.clone(); // Clone for run_scheduled_job_internal
 
             Box::pin(async move {
+                // Check if the job is paused before executing
+                let should_execute = {
+                    let jobs_map_guard = current_jobs_arc.lock().await;
+                    if let Some((_, current_job_in_map)) = jobs_map_guard.get(&task_job_id) {
+                        !current_job_in_map.paused
+                    } else {
+                        false
+                    }
+                };
+
+                if !should_execute {
+                    tracing::info!("Skipping execution of paused job '{}'", &task_job_id);
+                    return;
+                }
+
                 let current_time = Utc::now();
                 let mut needs_persist = false;
                 {
@@ -319,6 +336,21 @@ impl Scheduler {
                 let job_to_execute = job_for_task.clone(); // Clone for run_scheduled_job_internal
 
                 Box::pin(async move {
+                    // Check if the job is paused before executing
+                    let should_execute = {
+                        let jobs_map_guard = current_jobs_arc.lock().await;
+                        if let Some((_, stored_job)) = jobs_map_guard.get(&task_job_id) {
+                            !stored_job.paused
+                        } else {
+                            false
+                        }
+                    };
+
+                    if !should_execute {
+                        tracing::info!("Skipping execution of paused job '{}'", &task_job_id);
+                        return;
+                    }
+
                     let current_time = Utc::now();
                     let mut needs_persist = false;
                     {
@@ -515,6 +547,36 @@ impl Scheduler {
             ))),
         }
     }
+
+    pub async fn pause_schedule(&self, sched_id: &str) -> Result<(), SchedulerError> {
+        let mut jobs_guard = self.jobs.lock().await;
+        match jobs_guard.get_mut(sched_id) {
+            Some((_, job_def)) => {
+                if job_def.currently_running {
+                    return Err(SchedulerError::AnyhowError(anyhow!(
+                        "Cannot pause schedule '{}' while it's currently running",
+                        sched_id
+                    )));
+                }
+                job_def.paused = true;
+                self.persist_jobs_to_storage_with_guard(&jobs_guard).await?;
+                Ok(())
+            }
+            None => Err(SchedulerError::JobNotFound(sched_id.to_string())),
+        }
+    }
+
+    pub async fn unpause_schedule(&self, sched_id: &str) -> Result<(), SchedulerError> {
+        let mut jobs_guard = self.jobs.lock().await;
+        match jobs_guard.get_mut(sched_id) {
+            Some((_, job_def)) => {
+                job_def.paused = false;
+                self.persist_jobs_to_storage_with_guard(&jobs_guard).await?;
+                Ok(())
+            }
+            None => Err(SchedulerError::JobNotFound(sched_id.to_string())),
+        }
+    }
 }
 
 #[derive(Debug)]
@@ -858,6 +920,7 @@ mod tests {
             cron: "* * * * * * ".to_string(), // Runs every second for quick testing
             last_run: None,
             currently_running: false,
+            paused: false,
         };
 
         // Create the mock provider instance for the test