Add automation state CRUD

codex-rs/state/src/lib.rs

@@ -116,6 +116,8 @@ pub const GOALS_DB_FILENAME: &str = "goals_1.sqlite";
 pub const MEMORIES_DB_FILENAME: &str = "memories_1.sqlite";
 pub const AUTOMATIONS_DB_FILENAME: &str = "automations_1.sqlite";
 pub const STATE_DB_FILENAME: &str = "state_5.sqlite";
+pub const DEFAULT_AUTOMATION_RRULE: &str = "FREQ=HOURLY;INTERVAL=24;BYMINUTE=0";
+pub const AUTOMATION_RUN_JITTER_WINDOW_SECS: i64 = 120;
 
 /// Errors encountered during DB operations. Tags: [stage]
 pub const DB_ERROR_METRIC: &str = "codex.db.error";

codex-rs/state/src/runtime.rs

@@ -60,6 +60,7 @@ use std::time::Instant;
 use tracing::warn;
 
 mod agent_jobs;
+mod automation_schedule;
 mod automations;
 mod backfill;
 mod external_agent_config_imports;

codex-rs/state/src/runtime/automation_schedule.rs

@@ -0,0 +1,294 @@
+use crate::AUTOMATION_RUN_JITTER_WINDOW_SECS;
+use crate::AutomationKind;
+use chrono::DateTime;
+use chrono::Datelike;
+use chrono::Days;
+use chrono::Duration;
+use chrono::NaiveDate;
+use chrono::NaiveTime;
+use chrono::TimeZone;
+use chrono::Timelike;
+use chrono::Utc;
+use chrono::Weekday;
+use std::collections::BTreeMap;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum ScheduleFrequency {
+    Minutely,
+    Hourly,
+    Daily,
+    Weekly,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(super) struct AutomationSchedule {
+    frequency: ScheduleFrequency,
+    interval: u32,
+    hour: Option<u32>,
+    minute: Option<u32>,
+    weekdays: Vec<Weekday>,
+    one_shot: bool,
+}
+
+impl AutomationSchedule {
+    pub(super) fn parse(rrule: &str) -> anyhow::Result<Self> {
+        let normalized_rrule = rrule
+            .lines()
+            .map(str::trim)
+            .find_map(|line| line.strip_prefix("RRULE:").map(str::trim))
+            .unwrap_or_else(|| rrule.trim().strip_prefix("RRULE:").unwrap_or(rrule.trim()));
+        let mut entries = BTreeMap::new();
+        for part in normalized_rrule
+            .split(';')
+            .map(str::trim)
+            .filter(|part| !part.is_empty())
+        {
+            let Some((key, value)) = part.split_once('=') else {
+                anyhow::bail!("invalid rrule component: {part}");
+            };
+            entries.insert(key.to_ascii_uppercase(), value.trim().to_ascii_uppercase());
+        }
+
+        let frequency = match entries.get("FREQ").map(String::as_str) {
+            Some("MINUTELY") => ScheduleFrequency::Minutely,
+            Some("HOURLY") => ScheduleFrequency::Hourly,
+            Some("DAILY") => ScheduleFrequency::Daily,
+            Some("WEEKLY") => ScheduleFrequency::Weekly,
+            Some(other) => anyhow::bail!("unsupported rrule frequency: {other}"),
+            None => anyhow::bail!("rrule must include FREQ"),
+        };
+        let interval = entries
+            .get("INTERVAL")
+            .map(|value| value.parse::<u32>())
+            .transpose()
+            .map_err(|_| anyhow::anyhow!("invalid rrule INTERVAL"))?
+            .unwrap_or(1);
+        if interval == 0 {
+            anyhow::bail!("rrule INTERVAL must be >= 1");
+        }
+        let count = entries
+            .get("COUNT")
+            .map(|value| value.parse::<u32>())
+            .transpose()
+            .map_err(|_| anyhow::anyhow!("invalid rrule COUNT"))?;
+        if matches!(count, Some(value) if value > 1) {
+            anyhow::bail!("rrule COUNT > 1 is not supported");
+        }
+
+        let minute = parse_optional_u32(entries.get("BYMINUTE"), "BYMINUTE", 59)?;
+        let hour = parse_optional_u32(entries.get("BYHOUR"), "BYHOUR", 23)?;
+        let weekdays = parse_weekdays(entries.get("BYDAY"))?;
+
+        match frequency {
+            ScheduleFrequency::Minutely => {
+                if hour.is_some() || minute.is_some() || !weekdays.is_empty() {
+                    anyhow::bail!("MINUTELY rrules only support FREQ, INTERVAL, and COUNT");
+                }
+            }
+            ScheduleFrequency::Hourly => {
+                if hour.is_some() || !weekdays.is_empty() {
+                    anyhow::bail!("HOURLY rrules only support BYMINUTE, INTERVAL, and COUNT");
+                }
+            }
+            ScheduleFrequency::Daily => {
+                if hour.is_none() || minute.is_none() {
+                    anyhow::bail!("DAILY rrules require BYHOUR and BYMINUTE");
+                }
+                if !weekdays.is_empty() {
+                    anyhow::bail!("DAILY rrules do not support BYDAY");
+                }
+            }
+            ScheduleFrequency::Weekly => {
+                if interval != 1 {
+                    anyhow::bail!("WEEKLY rrules only support INTERVAL=1");
+                }
+                if hour.is_none() || minute.is_none() {
+                    anyhow::bail!("WEEKLY rrules require BYHOUR and BYMINUTE");
+                }
+                if weekdays.is_empty() {
+                    anyhow::bail!("WEEKLY rrules require BYDAY");
+                }
+            }
+        }
+
+        Ok(Self {
+            frequency,
+            interval,
+            hour,
+            minute,
+            weekdays,
+            one_shot: count == Some(1),
+        })
+    }
+}
+
+pub(super) fn compute_next_run_at(
+    kind: AutomationKind,
+    automation_id: &str,
+    schedule: &AutomationSchedule,
+    last_run_at: Option<DateTime<Utc>>,
+    now: DateTime<Utc>,
+) -> anyhow::Result<Option<DateTime<Utc>>> {
+    if schedule.one_shot && last_run_at.is_some() {
+        return Ok(None);
+    }
+
+    let scheduled = match schedule.frequency {
+        ScheduleFrequency::Minutely => align_minutely(schedule.interval, now),
+        ScheduleFrequency::Hourly => align_hourly(schedule.interval, schedule.minute, now)?,
+        ScheduleFrequency::Daily => {
+            align_daily(schedule.interval, schedule.hour, schedule.minute, now)?
+        }
+        ScheduleFrequency::Weekly => {
+            align_weekly(schedule.hour, schedule.minute, &schedule.weekdays, now)?
+        }
+    };
+    let jitter_seconds = if should_jitter(kind, schedule) {
+        deterministic_automation_jitter_seconds(automation_id, scheduled.timestamp())
+    } else {
+        0
+    };
+    Ok(Some(scheduled + Duration::seconds(jitter_seconds)))
+}
+
+fn parse_optional_u32(value: Option<&String>, key: &str, max: u32) -> anyhow::Result<Option<u32>> {
+    let Some(value) = value else {
+        return Ok(None);
+    };
+    let parsed = value
+        .parse::<u32>()
+        .map_err(|_| anyhow::anyhow!("invalid rrule {key}"))?;
+    if parsed > max {
+        anyhow::bail!("rrule {key} must be <= {max}");
+    }
+    Ok(Some(parsed))
+}
+
+fn parse_weekdays(value: Option<&String>) -> anyhow::Result<Vec<Weekday>> {
+    let Some(value) = value else {
+        return Ok(Vec::new());
+    };
+    value
+        .split(',')
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(|value| match value {
+            "MO" => Ok(Weekday::Mon),
+            "TU" => Ok(Weekday::Tue),
+            "WE" => Ok(Weekday::Wed),
+            "TH" => Ok(Weekday::Thu),
+            "FR" => Ok(Weekday::Fri),
+            "SA" => Ok(Weekday::Sat),
+            "SU" => Ok(Weekday::Sun),
+            _ => Err(anyhow::anyhow!("invalid rrule weekday: {value}")),
+        })
+        .collect()
+}
+
+fn should_jitter(kind: AutomationKind, schedule: &AutomationSchedule) -> bool {
+    if schedule.one_shot {
+        return false;
+    }
+    if kind == AutomationKind::Heartbeat
+        && matches!(
+            schedule.frequency,
+            ScheduleFrequency::Minutely | ScheduleFrequency::Hourly
+        )
+    {
+        return false;
+    }
+    matches!(
+        schedule.frequency,
+        ScheduleFrequency::Hourly | ScheduleFrequency::Daily | ScheduleFrequency::Weekly
+    )
+}
+
+fn deterministic_automation_jitter_seconds(automation_id: &str, run_at: i64) -> i64 {
+    if AUTOMATION_RUN_JITTER_WINDOW_SECS <= 0 {
+        return 0;
+    }
+    let mut hash = 0_u64;
+    for byte in automation_id.as_bytes() {
+        hash = hash.wrapping_mul(131).wrapping_add(u64::from(*byte));
+    }
+    hash = hash.wrapping_mul(131).wrapping_add(run_at as u64);
+    (hash % AUTOMATION_RUN_JITTER_WINDOW_SECS as u64) as i64
+}
+
+fn align_minutely(interval: u32, now: DateTime<Utc>) -> DateTime<Utc> {
+    let truncated = now
+        .with_second(0)
+        .and_then(|value| value.with_nanosecond(0))
+        .unwrap_or(now);
+    truncated + Duration::minutes(i64::from(interval))
+}
+
+fn align_hourly(
+    interval: u32,
+    minute: Option<u32>,
+    now: DateTime<Utc>,
+) -> anyhow::Result<DateTime<Utc>> {
+    let minute = minute.unwrap_or(now.minute());
+    let current = now
+        .with_minute(minute)
+        .and_then(|value| value.with_second(0))
+        .and_then(|value| value.with_nanosecond(0))
+        .ok_or_else(|| anyhow::anyhow!("failed to build hourly schedule candidate"))?;
+    if current > now {
+        return Ok(current);
+    }
+    Ok(current + Duration::hours(i64::from(interval)))
+}
+
+fn align_daily(
+    interval: u32,
+    hour: Option<u32>,
+    minute: Option<u32>,
+    now: DateTime<Utc>,
+) -> anyhow::Result<DateTime<Utc>> {
+    let candidate = date_time_for(now.date_naive(), hour, minute)?;
+    if candidate > now {
+        return Ok(candidate);
+    }
+    let next_date = now
+        .date_naive()
+        .checked_add_days(Days::new(u64::from(interval)))
+        .ok_or_else(|| anyhow::anyhow!("failed to compute daily next run"))?;
+    date_time_for(next_date, hour, minute)
+}
+
+fn align_weekly(
+    hour: Option<u32>,
+    minute: Option<u32>,
+    weekdays: &[Weekday],
+    now: DateTime<Utc>,
+) -> anyhow::Result<DateTime<Utc>> {
+    for offset in 0..=7_u64 {
+        let candidate_date = now
+            .date_naive()
+            .checked_add_days(Days::new(offset))
+            .ok_or_else(|| anyhow::anyhow!("failed to compute weekly next run"))?;
+        if !weekdays.contains(&candidate_date.weekday()) {
+            continue;
+        }
+        let candidate = date_time_for(candidate_date, hour, minute)?;
+        if candidate > now {
+            return Ok(candidate);
+        }
+    }
+    anyhow::bail!("failed to compute weekly next run")
+}
+
+fn date_time_for(
+    date: NaiveDate,
+    hour: Option<u32>,
+    minute: Option<u32>,
+) -> anyhow::Result<DateTime<Utc>> {
+    let time = NaiveTime::from_hms_opt(
+        hour.ok_or_else(|| anyhow::anyhow!("schedule missing hour"))?,
+        minute.ok_or_else(|| anyhow::anyhow!("schedule missing minute"))?,
+        0,
+    )
+    .ok_or_else(|| anyhow::anyhow!("failed to build schedule time"))?;
+    Ok(Utc.from_utc_datetime(&date.and_time(time)))
+}

codex-rs/state/src/runtime/automations.rs

@@ -1,6 +1,21 @@
+use super::automation_schedule::AutomationSchedule;
+use super::automation_schedule::compute_next_run_at;
 use super::*;
 use crate::Automation;
+use crate::AutomationCreateParams;
+use crate::AutomationDispatchSettings;
+use crate::AutomationStatus;
+use crate::AutomationTarget;
+use crate::AutomationUpdateParams;
+use crate::DEFAULT_AUTOMATION_RRULE;
 use crate::model::AutomationRow;
+use chrono::DateTime;
+use chrono::Utc;
+use codex_protocol::ThreadId;
+use uuid::Uuid;
+
+const AUTOMATION_NAME_MAX_CHARS: usize = 256;
+const AUTOMATION_PROMPT_MAX_CHARS: usize = 2_000;
 
 impl StateRuntime {
     pub async fn list_automations(&self) -> anyhow::Result<Vec<Automation>> {
@@ -19,16 +34,434 @@ ORDER BY updated_at DESC, id ASC
     }
 
     pub async fn get_automation(&self, id: &str) -> anyhow::Result<Option<Automation>> {
-        let row = sqlx::query_as::<_, AutomationRow>(
+        let row = load_automation_row(self.automations_pool.as_ref(), id).await?;
+        row.map(Automation::try_from).transpose()
+    }
+
+    pub async fn create_automation(
+        &self,
+        params: &AutomationCreateParams,
+    ) -> anyhow::Result<Automation> {
+        validate_automation_params(
+            params.name.as_str(),
+            params.prompt.as_str(),
+            &params.target,
+            params.dispatch_settings.as_ref(),
+        )?;
+        let mut tx = self.automations_pool.begin_with("BEGIN IMMEDIATE").await?;
+        if params.status == AutomationStatus::Active {
+            ensure_no_active_heartbeat_conflict(&mut tx, &params.target, /*exclude_id*/ None)
+                .await?;
+        }
+
+        let now = Utc::now();
+        let id = format!("automation-{}", Uuid::new_v4());
+        let rrule = resolve_automation_rrule(params.rrule.as_deref());
+        let next_run_at = next_run_at_for(
+            id.as_str(),
+            &params.target,
+            params.status,
+            rrule.as_str(),
+            /*last_run_at*/ None,
+            now,
+        )?;
+        let dispatch_json = DispatchSettingsJson::from(params.dispatch_settings.as_ref())?;
+
+        sqlx::query(
             r#"
-SELECT *
-FROM automations
+INSERT INTO automations (
+    id,
+    owner_thread_id,
+    kind,
+    name,
+    prompt,
+    status,
+    rrule,
+    next_run_at,
+    last_run_at,
+    created_at,
+    updated_at,
+    model,
+    reasoning_effort,
+    cron_cwds_json,
+    target_thread_id,
+    dispatch_workspace_roots_json,
+    dispatch_approval_policy_json,
+    dispatch_approvals_reviewer_json,
+    dispatch_permission_profile_json
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            "#,
+        )
+        .bind(id.as_str())
+        .bind(params.owner_thread_id.to_string())
+        .bind(params.target.kind().as_str())
+        .bind(params.name.trim())
+        .bind(params.prompt.trim())
+        .bind(params.status.as_str())
+        .bind(rrule.as_str())
+        .bind(next_run_at.map(|value| value.timestamp()))
+        .bind(now.timestamp())
+        .bind(now.timestamp())
+        .bind(trim_to_option(params.model.as_deref()))
+        .bind(params.reasoning_effort.as_ref().map(ToString::to_string))
+        .bind(cron_cwds_json(&params.target)?.as_deref())
+        .bind(target_thread_id(&params.target).as_deref())
+        .bind(dispatch_json.workspace_roots.as_deref())
+        .bind(dispatch_json.approval_policy.as_deref())
+        .bind(dispatch_json.approvals_reviewer.as_deref())
+        .bind(dispatch_json.permission_profile.as_deref())
+        .execute(&mut *tx)
+        .await?;
+        tx.commit().await?;
+
+        self.get_automation(id.as_str())
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("failed to load created automation {id}"))
+    }
+
+    pub async fn update_automation(
+        &self,
+        params: &AutomationUpdateParams,
+    ) -> anyhow::Result<Option<Automation>> {
+        self.update_automation_inner(params, /*expected_owner_thread_id*/ None)
+            .await
+    }
+
+    pub async fn update_automation_if_owner(
+        &self,
+        params: &AutomationUpdateParams,
+        expected_owner_thread_id: ThreadId,
+    ) -> anyhow::Result<Option<Automation>> {
+        self.update_automation_inner(params, Some(expected_owner_thread_id))
+            .await
+    }
+
+    async fn update_automation_inner(
+        &self,
+        params: &AutomationUpdateParams,
+        expected_owner_thread_id: Option<ThreadId>,
+    ) -> anyhow::Result<Option<Automation>> {
+        validate_automation_params(
+            params.name.as_str(),
+            params.prompt.as_str(),
+            &params.target,
+            params.dispatch_settings.as_ref(),
+        )?;
+
+        let mut tx = self.automations_pool.begin_with("BEGIN IMMEDIATE").await?;
+        let Some(existing_row) = load_automation_row_tx(&mut tx, params.id.as_str()).await? else {
+            tx.commit().await?;
+            return Ok(None);
+        };
+        if let Some(expected_owner_thread_id) = expected_owner_thread_id
+            && existing_row.owner_thread_id != expected_owner_thread_id.to_string()
+        {
+            tx.commit().await?;
+            return Ok(None);
+        }
+        let existing = Automation::try_from(existing_row.clone())?;
+        if existing.target.kind() != params.target.kind() {
+            anyhow::bail!("changing automation kind is not supported");
+        }
+        if params.status == AutomationStatus::Active {
+            ensure_no_active_heartbeat_conflict(&mut tx, &params.target, Some(params.id.as_str()))
+                .await?;
+        }
+
+        let now = Utc::now();
+        let rrule = resolve_automation_rrule(params.rrule.as_deref());
+        let schedule_changed = existing.rrule != rrule;
+        let target_changed = existing.target != params.target;
+        let next_run_at = if params.status == AutomationStatus::Active {
+            if existing.status != AutomationStatus::Active || schedule_changed || target_changed {
+                next_run_at_for(
+                    params.id.as_str(),
+                    &params.target,
+                    params.status,
+                    rrule.as_str(),
+                    existing.last_run_at,
+                    now,
+                )?
+            } else {
+                existing.next_run_at
+            }
+        } else {
+            None
+        };
+        let dispatch_json = DispatchSettingsJson::from(params.dispatch_settings.as_ref())?;
+
+        sqlx::query(
+            r#"
+UPDATE automations
+SET
+    owner_thread_id = ?,
+    name = ?,
+    prompt = ?,
+    status = ?,
+    rrule = ?,
+    next_run_at = ?,
+    updated_at = ?,
+    model = ?,
+    reasoning_effort = ?,
+    cron_cwds_json = ?,
+    target_thread_id = ?,
+    dispatch_workspace_roots_json = ?,
+    dispatch_approval_policy_json = ?,
+    dispatch_approvals_reviewer_json = ?,
+    dispatch_permission_profile_json = ?
 WHERE id = ?
             "#,
         )
-        .bind(id)
-        .fetch_optional(self.automations_pool.as_ref())
+        .bind(params.owner_thread_id.to_string())
+        .bind(params.name.trim())
+        .bind(params.prompt.trim())
+        .bind(params.status.as_str())
+        .bind(rrule.as_str())
+        .bind(next_run_at.map(|value| value.timestamp()))
+        .bind(now.timestamp())
+        .bind(trim_to_option(params.model.as_deref()))
+        .bind(params.reasoning_effort.as_ref().map(ToString::to_string))
+        .bind(cron_cwds_json(&params.target)?.as_deref())
+        .bind(target_thread_id(&params.target).as_deref())
+        .bind(dispatch_json.workspace_roots.as_deref())
+        .bind(dispatch_json.approval_policy.as_deref())
+        .bind(dispatch_json.approvals_reviewer.as_deref())
+        .bind(dispatch_json.permission_profile.as_deref())
+        .bind(params.id.as_str())
+        .execute(&mut *tx)
         .await?;
-        row.map(Automation::try_from).transpose()
+        tx.commit().await?;
+
+        self.get_automation(params.id.as_str()).await
+    }
+
+    pub async fn delete_automation(&self, id: &str) -> anyhow::Result<bool> {
+        self.delete_automation_inner(id, /*expected_owner_thread_id*/ None)
+            .await
+    }
+
+    pub async fn delete_automation_if_owner(
+        &self,
+        id: &str,
+        expected_owner_thread_id: ThreadId,
+    ) -> anyhow::Result<bool> {
+        self.delete_automation_inner(id, Some(expected_owner_thread_id))
+            .await
+    }
+
+    async fn delete_automation_inner(
+        &self,
+        id: &str,
+        expected_owner_thread_id: Option<ThreadId>,
+    ) -> anyhow::Result<bool> {
+        let mut query = sqlx::query(if expected_owner_thread_id.is_some() {
+            "DELETE FROM automations WHERE id = ? AND owner_thread_id = ?"
+        } else {
+            "DELETE FROM automations WHERE id = ?"
+        })
+        .bind(id);
+        if let Some(expected_owner_thread_id) = expected_owner_thread_id {
+            query = query.bind(expected_owner_thread_id.to_string());
+        }
+        let result = query.execute(self.automations_pool.as_ref()).await?;
+        Ok(result.rows_affected() == 1)
     }
 }
+
+async fn load_automation_row(
+    pool: &sqlx::SqlitePool,
+    id: &str,
+) -> anyhow::Result<Option<AutomationRow>> {
+    sqlx::query_as::<_, AutomationRow>(
+        r#"
+SELECT *
+FROM automations
+WHERE id = ?
+        "#,
+    )
+    .bind(id)
+    .fetch_optional(pool)
+    .await
+    .map_err(Into::into)
+}
+
+async fn load_automation_row_tx(
+    conn: &mut sqlx::SqliteConnection,
+    id: &str,
+) -> anyhow::Result<Option<AutomationRow>> {
+    sqlx::query_as::<_, AutomationRow>(
+        r#"
+SELECT *
+FROM automations
+WHERE id = ?
+        "#,
+    )
+    .bind(id)
+    .fetch_optional(&mut *conn)
+    .await
+    .map_err(Into::into)
+}
+
+async fn ensure_no_active_heartbeat_conflict(
+    conn: &mut sqlx::SqliteConnection,
+    target: &AutomationTarget,
+    exclude_id: Option<&str>,
+) -> anyhow::Result<()> {
+    let AutomationTarget::Heartbeat { thread_id } = target else {
+        return Ok(());
+    };
+    let existing_id = match exclude_id {
+        Some(exclude_id) => {
+            sqlx::query_scalar::<_, String>(
+                r#"
+SELECT id
+FROM automations
+WHERE kind = 'heartbeat'
+  AND status = 'ACTIVE'
+  AND target_thread_id = ?
+  AND id != ?
+LIMIT 1
+                "#,
+            )
+            .bind(thread_id.to_string())
+            .bind(exclude_id)
+            .fetch_optional(&mut *conn)
+            .await?
+        }
+        None => {
+            sqlx::query_scalar::<_, String>(
+                r#"
+SELECT id
+FROM automations
+WHERE kind = 'heartbeat'
+  AND status = 'ACTIVE'
+  AND target_thread_id = ?
+LIMIT 1
+                "#,
+            )
+            .bind(thread_id.to_string())
+            .fetch_optional(&mut *conn)
+            .await?
+        }
+    };
+    if let Some(existing_id) = existing_id {
+        anyhow::bail!("active heartbeat already exists for thread {thread_id}: {existing_id}");
+    }
+    Ok(())
+}
+
+fn validate_automation_params(
+    name: &str,
+    prompt: &str,
+    target: &AutomationTarget,
+    dispatch_settings: Option<&AutomationDispatchSettings>,
+) -> anyhow::Result<()> {
+    let name = name.trim();
+    if name.is_empty() {
+        anyhow::bail!("automation name is required");
+    }
+    if name.chars().count() > AUTOMATION_NAME_MAX_CHARS {
+        anyhow::bail!("automation name is too long");
+    }
+    let prompt = prompt.trim();
+    if prompt.is_empty() {
+        anyhow::bail!("automation prompt is required");
+    }
+    if prompt.chars().count() > AUTOMATION_PROMPT_MAX_CHARS {
+        anyhow::bail!("automation prompt is too long");
+    }
+    match target {
+        AutomationTarget::Cron { cwds } => {
+            if cwds.is_empty() {
+                anyhow::bail!("cron automations require at least one cwd");
+            }
+            for cwd in cwds {
+                if !cwd.is_absolute() {
+                    anyhow::bail!("cron automation cwd must be absolute: {}", cwd.display());
+                }
+            }
+        }
+        AutomationTarget::Heartbeat { .. } => {
+            if dispatch_settings.is_some() {
+                anyhow::bail!("heartbeat automations do not use dispatch settings");
+            }
+        }
+    }
+    Ok(())
+}
+
+fn resolve_automation_rrule(rrule: Option<&str>) -> String {
+    trim_to_option(rrule).unwrap_or_else(|| DEFAULT_AUTOMATION_RRULE.to_string())
+}
+
+fn next_run_at_for(
+    automation_id: &str,
+    target: &AutomationTarget,
+    status: AutomationStatus,
+    rrule: &str,
+    last_run_at: Option<DateTime<Utc>>,
+    now: DateTime<Utc>,
+) -> anyhow::Result<Option<DateTime<Utc>>> {
+    let schedule = AutomationSchedule::parse(rrule)?;
+    if status != AutomationStatus::Active {
+        return Ok(None);
+    }
+    compute_next_run_at(target.kind(), automation_id, &schedule, last_run_at, now)
+}
+
+fn trim_to_option(value: Option<&str>) -> Option<String> {
+    value
+        .map(str::trim)
+        .filter(|value| !value.is_empty())
+        .map(ToString::to_string)
+}
+
+fn cron_cwds_json(target: &AutomationTarget) -> anyhow::Result<Option<String>> {
+    match target {
+        AutomationTarget::Cron { cwds } => {
+            let cwds = cwds
+                .iter()
+                .map(|cwd| cwd.display().to_string())
+                .collect::<Vec<_>>();
+            Ok(Some(serde_json::to_string(&cwds)?))
+        }
+        AutomationTarget::Heartbeat { .. } => Ok(None),
+    }
+}
+
+fn target_thread_id(target: &AutomationTarget) -> Option<String> {
+    match target {
+        AutomationTarget::Cron { .. } => None,
+        AutomationTarget::Heartbeat { thread_id } => Some(thread_id.to_string()),
+    }
+}
+
+struct DispatchSettingsJson {
+    workspace_roots: Option<String>,
+    approval_policy: Option<String>,
+    approvals_reviewer: Option<String>,
+    permission_profile: Option<String>,
+}
+
+impl DispatchSettingsJson {
+    fn from(settings: Option<&AutomationDispatchSettings>) -> anyhow::Result<Self> {
+        Ok(match settings {
+            Some(settings) => Self {
+                workspace_roots: Some(serde_json::to_string(&settings.workspace_roots)?),
+                approval_policy: Some(serde_json::to_string(&settings.approval_policy)?),
+                approvals_reviewer: Some(serde_json::to_string(&settings.approvals_reviewer)?),
+                permission_profile: Some(serde_json::to_string(&settings.permission_profile)?),
+            },
+            None => Self {
+                workspace_roots: None,
+                approval_policy: None,
+                approvals_reviewer: None,
+                permission_profile: None,
+            },
+        })
+    }
+}
+
+#[cfg(test)]
+#[path = "automations_tests.rs"]
+mod tests;

codex-rs/state/src/runtime/automations_tests.rs

@@ -0,0 +1,132 @@
+use super::*;
+use crate::AutomationCreateParams;
+use crate::AutomationStatus;
+use crate::AutomationTarget;
+use crate::AutomationUpdateParams;
+use crate::runtime::test_support::unique_temp_dir;
+use codex_protocol::ThreadId;
+use pretty_assertions::assert_eq;
+use std::path::PathBuf;
+use uuid::Uuid;
+
+fn thread_id() -> ThreadId {
+    ThreadId::from_string(&Uuid::new_v4().to_string()).expect("valid thread id")
+}
+
+fn cron_create_params(owner_thread_id: ThreadId, cwd: PathBuf) -> AutomationCreateParams {
+    AutomationCreateParams {
+        owner_thread_id,
+        name: "Nightly sweep".to_string(),
+        prompt: "Summarize what changed.".to_string(),
+        status: AutomationStatus::Active,
+        rrule: Some("FREQ=DAILY;BYHOUR=9;BYMINUTE=30".to_string()),
+        model: Some("gpt-5".to_string()),
+        reasoning_effort: None,
+        target: AutomationTarget::Cron { cwds: vec![cwd] },
+        dispatch_settings: None,
+    }
+}
+
+#[tokio::test]
+async fn create_update_delete_cron_automation_round_trips() {
+    let codex_home = unique_temp_dir();
+    let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
+        .await
+        .expect("state runtime should initialize");
+    let owner_thread_id = thread_id();
+    let cwd = codex_home.join("workspace");
+    let created = runtime
+        .create_automation(&cron_create_params(owner_thread_id, cwd.clone()))
+        .await
+        .expect("create automation");
+
+    assert_eq!(created.owner_thread_id, owner_thread_id);
+    assert_eq!(created.name, "Nightly sweep");
+    assert_eq!(created.prompt, "Summarize what changed.");
+    assert_eq!(created.status, AutomationStatus::Active);
+    assert_eq!(
+        created.target,
+        AutomationTarget::Cron {
+            cwds: vec![cwd.clone()]
+        }
+    );
+    assert!(created.next_run_at.is_some());
+
+    let updated = runtime
+        .update_automation(&AutomationUpdateParams {
+            id: created.id.clone(),
+            owner_thread_id,
+            name: "Paused sweep".to_string(),
+            prompt: "Wait for now.".to_string(),
+            status: AutomationStatus::Paused,
+            rrule: Some("FREQ=DAILY;BYHOUR=10;BYMINUTE=0".to_string()),
+            model: None,
+            reasoning_effort: None,
+            target: AutomationTarget::Cron { cwds: vec![cwd] },
+            dispatch_settings: None,
+        })
+        .await
+        .expect("update automation")
+        .expect("automation should exist");
+
+    assert_eq!(updated.name, "Paused sweep");
+    assert_eq!(updated.status, AutomationStatus::Paused);
+    assert_eq!(updated.next_run_at, None);
+    assert_eq!(runtime.list_automations().await.expect("list").len(), 1);
+    assert!(
+        runtime
+            .delete_automation(created.id.as_str())
+            .await
+            .expect("delete")
+    );
+    assert_eq!(
+        runtime
+            .get_automation(created.id.as_str())
+            .await
+            .expect("get"),
+        None
+    );
+
+    let _ = tokio::fs::remove_dir_all(codex_home).await;
+}
+
+#[tokio::test]
+async fn active_heartbeat_targets_are_unique() {
+    let codex_home = unique_temp_dir();
+    let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
+        .await
+        .expect("state runtime should initialize");
+    let owner_thread_id = thread_id();
+    let target_thread_id = thread_id();
+    let params = AutomationCreateParams {
+        owner_thread_id,
+        name: "Heartbeat".to_string(),
+        prompt: "Check whether to continue.".to_string(),
+        status: AutomationStatus::Active,
+        rrule: Some("FREQ=HOURLY;INTERVAL=1".to_string()),
+        model: None,
+        reasoning_effort: None,
+        target: AutomationTarget::Heartbeat {
+            thread_id: target_thread_id,
+        },
+        dispatch_settings: None,
+    };
+
+    runtime
+        .create_automation(&params)
+        .await
+        .expect("create first heartbeat");
+    let duplicate = runtime
+        .create_automation(&params)
+        .await
+        .expect_err("duplicate active heartbeat should fail");
+
+    assert!(
+        duplicate
+            .to_string()
+            .contains("active heartbeat already exists"),
+        "{duplicate:#}"
+    );
+
+    let _ = tokio::fs::remove_dir_all(codex_home).await;
+}