diff --git a/src/openhuman/cron/mod.rs b/src/openhuman/cron/mod.rs index c3728ab63..904e64d18 100644 --- a/src/openhuman/cron/mod.rs +++ b/src/openhuman/cron/mod.rs @@ -22,8 +22,8 @@ pub use schemas::{ #[allow(unused_imports)] pub use store::{ add_agent_job, add_agent_job_with_definition, add_job, add_shell_job, clear_all_jobs, - delete_queued_runs, due_jobs, get_job, list_jobs, list_runs, record_last_run, record_run, - remove_job, reschedule_after_run, update_job, + dedup_named_jobs, delete_queued_runs, due_jobs, get_job, list_jobs, list_runs, record_last_run, + record_run, remove_job, reschedule_after_run, update_job, }; pub use types::{ ActiveHours, CronJob, CronJobPatch, CronRun, DeliveryConfig, JobType, Schedule, SessionTarget, diff --git a/src/openhuman/cron/seed.rs b/src/openhuman/cron/seed.rs index 81e0ff32d..39a7c674a 100644 --- a/src/openhuman/cron/seed.rs +++ b/src/openhuman/cron/seed.rs @@ -18,7 +18,8 @@ use crate::openhuman::config::Config; use crate::openhuman::cron::{ - add_agent_job_with_definition, list_jobs, remove_job, DeliveryConfig, Schedule, SessionTarget, + add_agent_job_with_definition, dedup_named_jobs, list_jobs, remove_job, DeliveryConfig, + Schedule, SessionTarget, }; use anyhow::Result; @@ -49,6 +50,16 @@ fn proactive_delivery() -> DeliveryConfig { /// Also prunes any stale one-shot `welcome` job a prior build might /// have persisted (see [`prune_legacy_welcome`]). pub fn seed_proactive_agents(config: &Config) -> Result<()> { + // Remove any duplicate named jobs left behind by older builds that + // used a non-atomic check-then-insert. Best-effort: log but continue + // on error so a dedup failure never blocks seeding. + if let Err(e) = dedup_named_jobs(config) { + tracing::warn!( + error = %e, + "[cron::seed] dedup_named_jobs failed — continuing without dedup" + ); + } + let existing = list_jobs(config)?; let has = |name: &str| existing.iter().any(|j| j.name.as_deref() == Some(name)); diff --git a/src/openhuman/cron/store.rs b/src/openhuman/cron/store.rs index d4f0adde9..b0947ddfa 100644 --- a/src/openhuman/cron/store.rs +++ b/src/openhuman/cron/store.rs @@ -197,6 +197,80 @@ pub fn clear_all_jobs(config: &Config) -> Result { Ok(removed) } +/// Remove duplicate cron jobs that share the same `name`. +/// +/// Older builds used a non-atomic check-then-insert in `seed_proactive_agents`, +/// which allowed two identical rows (e.g. two `morning_briefing` entries) to +/// land in the database when the function raced or was called twice before the +/// first insert committed. The `cron_jobs` table has no `UNIQUE` constraint on +/// `name`, so both rows persist and the Routines screen renders two cards. +/// +/// For each duplicated name this function keeps the row with the most +/// `cron_runs` history (ties broken by earliest `created_at`) and deletes +/// all others. Returns the total number of rows removed across all names. +/// +/// Idempotent: calling it on a database with no duplicates removes nothing +/// and returns `Ok(0)`. +pub fn dedup_named_jobs(config: &Config) -> Result { + with_connection(config, |conn| { + // 1. Find all names that appear more than once. + let duplicated_names: Vec = { + let mut stmt = conn.prepare( + "SELECT name FROM cron_jobs \ + WHERE name IS NOT NULL \ + GROUP BY name \ + HAVING COUNT(*) > 1", + )?; + let names = stmt.query_map([], |row| row.get::<_, String>(0))?; + let mut out = Vec::new(); + for n in names { + out.push(n?); + } + out + }; + + if duplicated_names.is_empty() { + return Ok(0); + } + + let mut canonical_stmt = conn.prepare( + "SELECT j.id \ + FROM cron_jobs j \ + LEFT JOIN cron_runs r ON r.job_id = j.id \ + WHERE j.name = ?1 \ + GROUP BY j.id \ + ORDER BY COUNT(r.id) DESC, j.created_at ASC, j.id ASC \ + LIMIT 1", + )?; + + let mut total_removed = 0usize; + for name in &duplicated_names { + // 2. Find the canonical id: most run history, tie-break earliest created_at. + let canonical_id: Option = { + let mut rows = canonical_stmt.query(params![name])?; + rows.next()?.map(|r| r.get::<_, String>(0)).transpose()? + }; + + let Some(keep_id) = canonical_id else { + continue; + }; + + // 3. Delete every other row sharing this name. + let deleted = conn.execute( + "DELETE FROM cron_jobs WHERE name = ?1 AND id != ?2", + params![name, keep_id], + )?; + log::info!( + "[cron] dedup_named_jobs: removed {deleted} duplicate(s) of '{name}' \ + (keeping id={keep_id})" + ); + total_removed += deleted; + } + + Ok(total_removed) + }) +} + pub fn due_jobs(config: &Config, now: DateTime) -> Result> { let lim = i64::try_from(config.scheduler.max_tasks.max(1)) .context("Scheduler max_tasks overflows i64")?; diff --git a/src/openhuman/cron/store_tests.rs b/src/openhuman/cron/store_tests.rs index 01d9d87bf..3eb413b97 100644 --- a/src/openhuman/cron/store_tests.rs +++ b/src/openhuman/cron/store_tests.rs @@ -240,3 +240,154 @@ fn reschedule_after_run_truncates_last_output() { assert!(last_output.ends_with(TRUNCATED_OUTPUT_MARKER)); assert!(last_output.len() <= MAX_CRON_OUTPUT_BYTES); } + +// ── dedup_named_jobs ───────────────────────────────────────────── + +#[test] +fn dedup_named_jobs_no_op_on_empty_db() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + let removed = dedup_named_jobs(&config).unwrap(); + assert_eq!(removed, 0); +} + +#[test] +fn dedup_named_jobs_no_op_when_no_duplicates() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + add_shell_job( + &config, + Some("job_a".into()), + Schedule::Cron { + expr: "*/5 * * * *".into(), + tz: None, + active_hours: None, + }, + "echo a", + ) + .unwrap(); + add_shell_job( + &config, + Some("job_b".into()), + Schedule::Cron { + expr: "*/10 * * * *".into(), + tz: None, + active_hours: None, + }, + "echo b", + ) + .unwrap(); + let removed = dedup_named_jobs(&config).unwrap(); + assert_eq!(removed, 0); + assert_eq!(list_jobs(&config).unwrap().len(), 2); +} + +#[test] +fn dedup_named_jobs_removes_duplicates_keeping_history() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + // Insert two jobs with the same name directly — simulating the old double-seed bug. + let job_a = add_shell_job( + &config, + Some("morning_briefing".into()), + Schedule::Cron { + expr: "0 7 * * *".into(), + tz: None, + active_hours: None, + }, + "echo briefing", + ) + .unwrap(); + let job_b = add_shell_job( + &config, + Some("morning_briefing".into()), + Schedule::Cron { + expr: "0 7 * * *".into(), + tz: None, + active_hours: None, + }, + "echo briefing", + ) + .unwrap(); + + // Add run history to job_a — it should survive. + let now = Utc::now(); + record_run( + &config, + &job_a.id, + now, + now + ChronoDuration::seconds(1), + "ok", + Some("output"), + 1000, + ) + .unwrap(); + + let removed = dedup_named_jobs(&config).unwrap(); + assert_eq!(removed, 1); + + let remaining = list_jobs(&config).unwrap(); + assert_eq!(remaining.len(), 1); + assert_eq!( + remaining[0].id, job_a.id, + "job with run history should be kept" + ); + assert!( + get_job(&config, &job_b.id).is_err(), + "duplicate without history should be removed" + ); +} + +#[test] +fn dedup_named_jobs_keeps_earliest_when_history_tied() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + // Both jobs have no run history — tie broken by earliest created_at. + let job_a = add_shell_job( + &config, + Some("routine".into()), + Schedule::Cron { + expr: "0 8 * * *".into(), + tz: None, + active_hours: None, + }, + "echo first", + ) + .unwrap(); + let job_b = add_shell_job( + &config, + Some("routine".into()), + Schedule::Cron { + expr: "0 8 * * *".into(), + tz: None, + active_hours: None, + }, + "echo second", + ) + .unwrap(); + + let removed = dedup_named_jobs(&config).unwrap(); + assert_eq!(removed, 1); + + let remaining = list_jobs(&config).unwrap(); + assert_eq!(remaining.len(), 1); + // job_a was created first — it should win the tie. + assert_eq!(remaining[0].id, job_a.id, "earliest job should be kept"); + assert!(get_job(&config, &job_b.id).is_err()); +} + +#[test] +fn dedup_named_jobs_ignores_unnamed_jobs() { + let tmp = TempDir::new().unwrap(); + let config = test_config(&tmp); + + // Unnamed jobs (name = NULL) — dedup should not touch them. + add_job(&config, "*/5 * * * *", "echo unnamed-1").unwrap(); + add_job(&config, "*/5 * * * *", "echo unnamed-2").unwrap(); + + let removed = dedup_named_jobs(&config).unwrap(); + assert_eq!(removed, 0); + assert_eq!(list_jobs(&config).unwrap().len(), 2); +}