fix(cron): make cron_run non-blocking, enqueue async execution (#3127) (#3142)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>

app/src/components/intelligence/memoryGraphLayout.test.ts

@@ -43,7 +43,7 @@ describe('memoryGraphLayout', () => {
 
   it('nodeRadius grows with summary level and is fixed for chunk/contact', () => {
     expect(nodeRadius(summary({ level: 0 }))).toBe(5);
-    expect(nodeRadius(summary({ level: 3 }))).toBeCloseTo(12.5);
+    expect(nodeRadius(summary({ level: 3 }))).toBe(12.5);
     expect(nodeRadius(summary({ level: 99 }))).toBe(252.5);
     expect(nodeRadius(contact())).toBe(9);
     expect(nodeRadius(chunk())).toBe(3);

app/src/lib/i18n/ar.ts

@@ -214,6 +214,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'لم يهرب بعد',
   'routines.runNow': 'اركض الآن',
   'routines.running': 'تشغيل...',
+  'routines.runNowTimedOut': 'انتهت مهلة التشغيل — يرجى التحديث والمحاولة مرة أخرى.',
   'routines.viewHistory': 'تاريخ الرؤية',
   'routines.loadingHistory': 'التعبئة...',
   'routines.noHistory': 'لا يوجد تاريخ بعد',

app/src/lib/i18n/bn.ts

@@ -216,6 +216,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'এখনো চালাও নি',
   'routines.runNow': 'এখন সঞ্চালন করুন',
   'routines.running': 'চলমান...',
+  'routines.runNowTimedOut': 'রান টাইমআউট হয়েছে — অনুগ্রহ করে রিফ্রেশ করুন এবং আবার চেষ্টা করুন।',
   'routines.viewHistory': 'পূর্ববর্তী তথ্য প্রদর্শন',
   'routines.loadingHistory': 'লোড করা হচ্ছে...',
   'routines.noHistory': 'এখনো কোন ইতিহাস নেই।',

app/src/lib/i18n/de.ts

@@ -225,6 +225,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'Noch nicht ausgeführt',
   'routines.runNow': 'Ausführung starten',
   'routines.running': 'Läuft…',
+  'routines.runNowTimedOut': 'Ausführung abgebrochen – bitte Seite neu laden und erneut versuchen.',
   'routines.viewHistory': 'Verlauf ansehen',
   'routines.loadingHistory': 'Laden…',
   'routines.noHistory': 'Noch kein Verlauf',

app/src/lib/i18n/en.ts

@@ -237,6 +237,7 @@ const en: TranslationMap = {
   'routines.notRunYet': 'Not run yet',
   'routines.runNow': 'Run Now',
   'routines.running': 'Running…',
+  'routines.runNowTimedOut': 'Run timed out — please refresh and try again.',
   'routines.viewHistory': 'View history',
   'routines.loadingHistory': 'Loading…',
   'routines.noHistory': 'No run history yet.',

app/src/lib/i18n/es.ts

@@ -226,6 +226,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'Aún no se ha ejecutado',
   'routines.runNow': 'Corre ahora',
   'routines.running': 'Corriendo…',
+  'routines.runNowTimedOut': 'Tiempo de espera agotado — actualiza e intenta de nuevo.',
   'routines.viewHistory': 'Ver historial',
   'routines.loadingHistory': 'Cargando…',
   'routines.noHistory': 'Aún no hay historial de ejecuciones.',

app/src/lib/i18n/fr.ts

@@ -224,6 +224,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'Pas encore exécuté',
   'routines.runNow': 'Cours maintenant',
   'routines.running': 'Course…',
+  'routines.runNowTimedOut': 'Délai dépassé — veuillez rafraîchir et réessayer.',
   'routines.viewHistory': "Afficher l'historique",
   'routines.loadingHistory': 'Chargement…',
   'routines.noHistory': "Pas encore d'historique d'exécution.",

app/src/lib/i18n/hi.ts

@@ -216,6 +216,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'अभी तक नहीं चला',
   'routines.runNow': 'अब चलायें',
   'routines.running': 'दौड़ना...',
+  'routines.runNowTimedOut': 'चलाने में समय समाप्त — कृपया रिफ्रेश करें और पुनः प्रयास करें।',
   'routines.viewHistory': 'इतिहास देखें',
   'routines.loadingHistory': 'लोड...',
   'routines.noHistory': 'अभी तक कोई रन इतिहास नहीं है।',

app/src/lib/i18n/id.ts

@@ -218,6 +218,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'Belum pernah dijalankan',
   'routines.runNow': 'Jalankan Sekarang',
   'routines.running': 'Berjalan...',
+  'routines.runNowTimedOut': 'Waktu habis — silakan muat ulang dan coba lagi.',
   'routines.viewHistory': 'Lihat riwayat',
   'routines.loadingHistory': 'Memuat…',
   'routines.noHistory': 'Belum ada riwayat eksekusi.',

app/src/lib/i18n/it.ts

@@ -222,6 +222,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'Non ancora eseguito',
   'routines.runNow': 'Corri adesso',
   'routines.running': 'Correndo…',
+  'routines.runNowTimedOut': "Timeout durante l'esecuzione — aggiorna e riprova.",
   'routines.viewHistory': 'Visualizza cronologia',
   'routines.loadingHistory': 'Caricamento…',
   'routines.noHistory': 'Ancora nessuna cronologia delle esecuzioni.',

app/src/lib/i18n/ko.ts

@@ -216,6 +216,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': '아직 실행되지 않음',
   'routines.runNow': '지금 실행',
   'routines.running': '실행 중…',
+  'routines.runNowTimedOut': '실행 시간 초과 — 새로고침 후 다시 시도하세요.',
   'routines.viewHistory': '기록 보기',
   'routines.loadingHistory': '로딩 중…',
   'routines.noHistory': '아직 실행 기록이 없습니다.',

app/src/lib/i18n/pl.ts

@@ -220,6 +220,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'Jeszcze nie uruchomiono',
   'routines.runNow': 'Uruchom teraz',
   'routines.running': 'Uruchamianie…',
+  'routines.runNowTimedOut': 'Przekroczono limit czasu — odśwież stronę i spróbuj ponownie.',
   'routines.viewHistory': 'Wyświetl historię',
   'routines.loadingHistory': 'Ładowanie…',
   'routines.noHistory': 'Brak historii uruchomień.',

app/src/lib/i18n/pt.ts

@@ -224,6 +224,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'Ainda não executado',
   'routines.runNow': 'Corra Agora',
   'routines.running': 'Executando…',
+  'routines.runNowTimedOut': 'Execução expirou — atualize e tente novamente.',
   'routines.viewHistory': 'Ver histórico',
   'routines.loadingHistory': 'Carregando…',
   'routines.noHistory': 'Ainda não há histórico de execuções.',

app/src/lib/i18n/ru.ts

@@ -218,6 +218,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': 'Еще не запущен',
   'routines.runNow': 'Беги сейчас',
   'routines.running': 'Бег…',
+  'routines.runNowTimedOut': 'Время выполнения истекло — обновите страницу и попробуйте снова.',
   'routines.viewHistory': 'Посмотреть историю',
   'routines.loadingHistory': 'Загрузка…',
   'routines.noHistory': 'Истории запусков пока нет.',

app/src/lib/i18n/zh-CN.ts

@@ -206,6 +206,7 @@ const messages: TranslationMap = {
   'routines.notRunYet': '尚未运行',
   'routines.runNow': '立即运行',
   'routines.running': '运行中…',
+  'routines.runNowTimedOut': '运行超时，请刷新后重试。',
   'routines.viewHistory': '查看历史',
   'routines.loadingHistory': '加载中…',
   'routines.noHistory': '暂无运行历史。',

app/src/pages/Routines.tsx

@@ -1,5 +1,5 @@
 import createDebug from 'debug';
-import { useCallback, useEffect, useState } from 'react';
+import { useCallback, useEffect, useRef, useState } from 'react';
 import { useNavigate } from 'react-router-dom';
 
 import RoutineCard from '../components/routines/RoutineCard';
@@ -25,6 +25,14 @@ const Routines = () => {
   const [runsByJob, setRunsByJob] = useState<Record<string, CoreCronRun[]>>({});
   const [busyKeys, setBusyKeys] = useState<Set<string>>(new Set());
 
+  const isMountedRef = useRef(true);
+
+  useEffect(() => {
+    return () => {
+      isMountedRef.current = false;
+    };
+  }, []);
+
   const addBusy = (key: string) => setBusyKeys(prev => new Set(prev).add(key));
   const removeBusy = (key: string) =>
     setBusyKeys(prev => {
@@ -80,21 +88,58 @@ const Routines = () => {
     addBusy(key);
     setError(null);
     try {
-      await openhumanCronRun(jobId);
-      const runs = await openhumanCronRuns(jobId, 10);
-      setRunsByJob(prev => ({ ...prev, [jobId]: runs.result }));
-      // Refresh job list to update last_status
+      const runResponse = await openhumanCronRun(jobId);
+
+      if (runResponse.result.status === 'queued') {
+        // Job was enqueued asynchronously — poll until a new run record appears.
+        // Compare by the latest run's id (not list length) so this works correctly
+        // when the job already has >= 10 runs and the list stays at the fetch limit.
+        const previousLatestId = (runsByJob[jobId] ?? [])[0]?.id;
+        const POLL_INTERVAL_MS = 2000;
+        const MAX_WAIT_MS = 120_000;
+        let elapsed = 0;
+
+        while (elapsed < MAX_WAIT_MS) {
+          await new Promise<void>(resolve => setTimeout(resolve, POLL_INTERVAL_MS));
+          if (!isMountedRef.current) return;
+          elapsed += POLL_INTERVAL_MS;
+          const runs = await openhumanCronRuns(jobId, 10);
+          if (!isMountedRef.current) return;
+          setRunsByJob(prev => ({ ...prev, [jobId]: runs.result }));
+          const latest = runs.result[0];
+          if (
+            latest?.id !== undefined &&
+            latest.id !== previousLatestId &&
+            latest.status !== 'queued'
+          ) {
+            break;
+          }
+        }
+        if (elapsed >= MAX_WAIT_MS && isMountedRef.current) {
+          setError(t('routines.runNowTimedOut'));
+        }
+      } else {
+        // Synchronous response (legacy path — kept for backward compatibility).
+        const runs = await openhumanCronRuns(jobId, 10);
+        setRunsByJob(prev => ({ ...prev, [jobId]: runs.result }));
+      }
+
+      if (!isMountedRef.current) return;
+
+      // Refresh job list to update last_status regardless of path.
       const response = await openhumanCronList();
+      if (!isMountedRef.current) return;
       setJobs(
         [...response.result].sort(
           (a, b) => new Date(a.next_run).getTime() - new Date(b.next_run).getTime()
         )
       );
     } catch (err) {
+      if (!isMountedRef.current) return;
       const message = err instanceof Error ? err.message : String(err);
       setError(message);
     } finally {
-      removeBusy(key);
+      if (isMountedRef.current) removeBusy(key);
     }
   };
 

app/src/utils/tauriCommands/cron.ts

@@ -111,9 +111,9 @@ export async function openhumanCronRun(
 ): Promise<
   CommandResponse<{
     job_id: string;
-    status: 'ok' | 'error' | string;
-    duration_ms: number;
-    output: string;
+    status: 'queued' | 'ok' | 'error' | string;
+    duration_ms?: number;
+    output?: string;
   }>
 > {
   if (!isTauri()) {
@@ -122,9 +122,9 @@ export async function openhumanCronRun(
   return await callCoreRpc<
     CommandResponse<{
       job_id: string;
-      status: 'ok' | 'error' | string;
-      duration_ms: number;
-      output: string;
+      status: 'queued' | 'ok' | 'error' | string;
+      duration_ms?: number;
+      output?: string;
     }>
   >({ method: 'openhuman.cron_run', params: { job_id: jobId } });
 }

src/openhuman/cron/mod.rs

@@ -21,9 +21,9 @@ 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, due_jobs,
-    get_job, list_jobs, list_runs, record_last_run, record_run, remove_job, reschedule_after_run,
-    update_job,
+    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,
 };
 pub use types::{
     ActiveHours, CronJob, CronJobPatch, CronRun, DeliveryConfig, JobType, Schedule, SessionTarget,

src/openhuman/cron/ops.rs

@@ -5,7 +5,30 @@ use crate::openhuman::cron::{
 use crate::openhuman::security::SecurityPolicy;
 use crate::rpc::RpcOutcome;
 use anyhow::Result;
+use once_cell::sync::Lazy;
 use serde_json::json;
+use std::collections::HashSet;
+use std::sync::Mutex;
+
+/// Guard against duplicate concurrent "Run Now" executions per job.
+/// Entries are inserted before spawning and removed when the task completes.
+static ACTIVE_RUNS: Lazy<Mutex<HashSet<String>>> = Lazy::new(|| Mutex::new(HashSet::new()));
+
+/// RAII guard that removes a job ID from `ACTIVE_RUNS` when dropped.
+///
+/// Ensures cleanup runs on normal completion, panic, or future cancellation —
+/// so a hung or aborted background task can never permanently lock a job_id.
+struct ActiveRunGuard {
+    job_id: String,
+}
+
+impl Drop for ActiveRunGuard {
+    fn drop(&mut self) {
+        if let Ok(mut active) = ACTIVE_RUNS.lock() {
+            active.remove(&self.job_id);
+        }
+    }
+}
 
 pub fn add_once(config: &Config, delay: &str, command: &str) -> Result<CronJob> {
     let duration = parse_human_delay(delay)?;
@@ -186,43 +209,97 @@ pub async fn cron_run(
     config: &Config,
     job_id: &str,
 ) -> Result<RpcOutcome<serde_json::Value>, String> {
-    if job_id.trim().is_empty() {
+    let job_id = job_id.trim();
+    if job_id.is_empty() {
         return Err("Missing 'job_id' parameter".to_string());
     }
     if !config.cron.enabled {
         return Err("cron is disabled by config (cron.enabled=false)".to_string());
     }
 
-    let job = cron::get_job(config, job_id.trim()).map_err(|e| e.to_string())?;
-    let started_at = chrono::Utc::now();
-    let (success, output) = cron::scheduler::execute_job_now(config, &job).await;
-    let finished_at = chrono::Utc::now();
-    let duration_ms = (finished_at - started_at).num_milliseconds();
-    let status = if success { "ok" } else { "error" };
+    // Guard against duplicate concurrent executions for the same job.
+    {
+        let mut active = ACTIVE_RUNS
+            .lock()
+            .map_err(|_| "ACTIVE_RUNS lock poisoned".to_string())?;
+        if active.contains(job_id) {
+            tracing::debug!(job_id, "[cron_run] rejected duplicate concurrent execution");
+            return Err(format!("cron job '{job_id}' is already running"));
+        }
+        active.insert(job_id.to_string());
+    }
 
-    let _ = cron::record_run(
-        config,
-        &job.id,
-        started_at,
-        finished_at,
-        status,
-        Some(&output),
-        duration_ms,
-    );
-    let _ = cron::record_last_run(config, &job.id, finished_at, success, &output);
+    let job = match cron::get_job(config, job_id) {
+        Ok(j) => j,
+        Err(e) => {
+            if let Ok(mut active) = ACTIVE_RUNS.lock() {
+                active.remove(job_id);
+            }
+            return Err(e.to_string());
+        }
+    };
 
-    // Deliver via the same path as the scheduler loop so proactive
-    // messages and alerts are sent on "Run Now" too.
-    cron::scheduler::deliver_job(config, &job, &output).await;
+    // Insert a "queued" placeholder run record immediately so the frontend
+    // poller can observe the run as soon as the RPC returns — otherwise the
+    // run list stays unchanged until execute_job_now finishes.
+    let queued_at = chrono::Utc::now();
+    let _ = cron::record_run(config, &job.id, queued_at, queued_at, "queued", None, 0);
+
+    let config_owned = config.clone();
+    let job_id_owned = job_id.to_string();
+
+    tracing::debug!(job_id, "[cron_run] enqueuing background execution");
+
+    // Spawn the execution as a background task so the RPC handler returns
+    // immediately instead of blocking for the full job duration.
+    tokio::spawn(async move {
+        // Drop guard ensures job_id is removed from ACTIVE_RUNS even on panic
+        // or future cancellation, not just on the normal completion path.
+        let _guard = ActiveRunGuard {
+            job_id: job_id_owned.clone(),
+        };
+
+        tracing::debug!(job_id = %job_id_owned, "[cron_run] background task started");
+
+        let started_at = chrono::Utc::now();
+        let (success, output) = cron::scheduler::execute_job_now(&config_owned, &job).await;
+        let finished_at = chrono::Utc::now();
+        let duration_ms = (finished_at - started_at).num_milliseconds();
+        let status = if success { "ok" } else { "error" };
+
+        tracing::debug!(
+            job_id = %job_id_owned,
+            status,
+            duration_ms,
+            "[cron_run] background task finished"
+        );
+
+        // Remove the "queued" placeholder before inserting the real result
+        // so we don't leave orphaned rows in the run history.
+        let _ = cron::delete_queued_runs(&config_owned, &job.id);
+
+        let _ = cron::record_run(
+            &config_owned,
+            &job.id,
+            started_at,
+            finished_at,
+            status,
+            Some(&output),
+            duration_ms,
+        );
+        let _ = cron::record_last_run(&config_owned, &job.id, finished_at, success, &output);
+
+        // Deliver via the same path as the scheduler loop so proactive
+        // messages and alerts are sent on "Run Now" too.
+        cron::scheduler::deliver_job(&config_owned, &job, &output).await;
+    });
 
     Ok(RpcOutcome::new(
         json!({
-            "job_id": job.id,
-            "status": status,
-            "duration_ms": duration_ms,
-            "output": output
+            "job_id": job_id,
+            "status": "queued",
         }),
-        vec![format!("cron job run: {}", job_id.trim())],
+        vec![format!("cron job enqueued: {}", job_id)],
     ))
 }
 

src/openhuman/cron/ops_tests.rs

@@ -418,3 +418,59 @@ async fn cron_runs_returns_empty_history_for_new_job() {
     assert!(out.value.is_empty());
     assert!(out.logs.iter().any(|l| l.contains("cron run history")));
 }
+
+// ── cron_run ────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn cron_run_rejects_empty_job_id() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let err = cron_run(&config, "").await.unwrap_err();
+    assert!(err.contains("Missing 'job_id'"));
+}
+
+#[tokio::test]
+async fn cron_run_rejects_whitespace_job_id() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let err = cron_run(&config, "   ").await.unwrap_err();
+    assert!(err.contains("Missing 'job_id'"));
+}
+
+#[tokio::test]
+async fn cron_run_errors_when_cron_disabled() {
+    let tmp = TempDir::new().unwrap();
+    let config = disabled_cron_config(&tmp);
+    let err = cron_run(&config, "some-id").await.unwrap_err();
+    assert!(err.contains("cron is disabled"));
+}
+
+#[tokio::test]
+async fn cron_run_returns_queued_immediately_for_valid_job() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let job = make_job(&config, "*/5 * * * *", None, "echo hello");
+
+    let out = cron_run(&config, &job.id).await.unwrap();
+    assert_eq!(out.value["job_id"], json!(job.id));
+    assert_eq!(out.value["status"], json!("queued"));
+    assert!(out.logs.iter().any(|l| l.contains("enqueued")));
+}
+
+#[tokio::test]
+async fn cron_run_rejects_duplicate_concurrent_execution() {
+    let tmp = TempDir::new().unwrap();
+    let config = test_config(&tmp);
+    let job = make_job(&config, "*/5 * * * *", None, "echo hello");
+
+    // Seed ACTIVE_RUNS directly to simulate an in-flight execution without
+    // spawning a real long-running subprocess (deterministic, no timing dependency).
+    ACTIVE_RUNS.lock().unwrap().insert(job.id.clone());
+    let err = cron_run(&config, &job.id).await.unwrap_err();
+    ACTIVE_RUNS.lock().unwrap().remove(&job.id);
+
+    assert!(
+        err.contains("already running"),
+        "expected 'already running', got: {err}"
+    );
+}

src/openhuman/cron/store.rs

@@ -394,6 +394,18 @@ pub fn record_run(
     })
 }
 
+/// Remove all "queued" placeholder records for a given job so that only the
+/// real (ok/error) result row remains in the run history.
+pub fn delete_queued_runs(config: &Config, job_id: &str) -> Result<usize> {
+    with_connection(config, |conn| {
+        let deleted = conn.execute(
+            "DELETE FROM cron_runs WHERE job_id = ?1 AND status = 'queued'",
+            params![job_id],
+        )?;
+        Ok(deleted)
+    })
+}
+
 fn truncate_cron_output(output: &str) -> String {
     if output.len() <= MAX_CRON_OUTPUT_BYTES {
         return output.to_string();