fix(cron): prevent scheduler loop from blocking when a job hangs (#820)

* fix(cron): prevent scheduler loop from blocking when a job hangs

The cron scheduler's runLoop calls checkAndRunDueJobs() every second,
which previously used wg.Wait() to block until ALL claimed jobs complete.
If any single job hung (LLM provider timeout, agent loop stuck, network
issue), wg.Wait() would block indefinitely, preventing the scheduler
from ever checking for new due jobs — effectively killing all cron
scheduling until a container restart.

Changes:
- Remove wg.Wait() from both PG and SQLite cron schedulers — jobs now
  run as independent goroutines that don't block the check loop
- Add panic recovery to PG runLoop (safeCheckAndRunDueJobs wrapper)
  and per-job goroutines, matching the existing safego.Recover pattern
  in the SQLite scheduler
- Add 10-minute context timeout to the cron job handler so a hung
  agent run is cancelled instead of blocking forever
- Use select with context.Done() in the handler to respect the timeout
  when waiting for the scheduler outcome
- Invalidate PG job cache per-job on completion instead of after the
  (now-removed) batch wait

The SQLite scheduler already had safego.Recover on job goroutines but
still used wg.Wait() — this commit removes that blocking wait as well.

* fix(cron): make job timeout configurable + add SQLite runLoop panic recovery

- Add `cron.job_timeout` config field (Go duration string, default "10m")
  so operators can tune the per-job timeout for complex agent workloads
  without code changes
- Add `safeCheckJobs` panic recovery wrapper to SQLite cron runLoop,
  matching the PG scheduler's `safeCheckAndRunDueJobs` for consistency
- Use dynamic timeout string in error message for better diagnostics

* fix: remove unused "time" import from gateway_cron.go

* fix(cron): apply same fixes to SQLite DB scheduler (sqlitestore)

The SQLite DB-backed scheduler (used by desktop edition with SQLite
backend) had the exact same wg.Wait() blocking issue and missing
panic recovery as the PG scheduler. Apply identical fixes:

- Remove wg.Wait() — jobs run as independent goroutines
- Add safeCheckAndRunDueJobs panic recovery wrapper for runLoop
- Add per-job panic recovery and cache invalidation

---------

Co-authored-by: Luvu182 <208665161+Luvu182@users.noreply.github.com>

cmd/gateway_cron.go

@@ -78,8 +78,12 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
 			)
 		}
 
-		// Build context with tenant scope so agent loop events are scoped correctly.
-		cronCtx := store.WithTenantID(context.Background(), job.TenantID)
+		// Build context with tenant scope and timeout so agent loop events are
+		// scoped correctly and a hung agent can't block the cron scheduler forever.
+		jobTimeout := cfg.Cron.JobTimeoutDuration()
+		cronCtx, cancelCron := context.WithTimeout(context.Background(), jobTimeout)
+		defer cancelCron()
+		cronCtx = store.WithTenantID(cronCtx, job.TenantID)
 
 		// Reset session before each cron run to prevent tool errors from previous
 		// runs from polluting the context and blocking future executions (#294).
@@ -106,8 +110,13 @@ func makeCronJobHandler(sched *scheduler.Scheduler, msgBus *bus.MessageBus, cfg
 			TraceTags:         []string{"cron"},
 		})
 
-		// Block until the scheduled run completes
-		outcome := <-outCh
+		// Block until the scheduled run completes or the timeout fires.
+		var outcome scheduler.RunOutcome
+		select {
+		case outcome = <-outCh:
+		case <-cronCtx.Done():
+			return nil, fmt.Errorf("cron job %s timed out after %s", job.Name, jobTimeout)
+		}
 		if outcome.Err != nil {
 			return nil, outcome.Err
 		}

internal/config/config.go

@@ -343,6 +343,20 @@ type CronConfig struct {
 	RetryBaseDelay  string `json:"retry_base_delay,omitempty"` // initial backoff delay (default "2s", Go duration)
 	RetryMaxDelay   string `json:"retry_max_delay,omitempty"`  // maximum backoff delay (default "30s", Go duration)
 	DefaultTimezone string `json:"default_timezone,omitempty"` // IANA timezone for cron expressions when not set per-job (e.g. "Asia/Ho_Chi_Minh")
+	JobTimeout      string `json:"job_timeout,omitempty"`      // max duration per cron job execution (default "10m", Go duration)
+}
+
+// DefaultJobTimeout is the fallback timeout for cron job execution.
+const DefaultJobTimeout = 10 * time.Minute
+
+// JobTimeoutDuration returns the configured job timeout or the default (10m).
+func (cc CronConfig) JobTimeoutDuration() time.Duration {
+	if cc.JobTimeout != "" {
+		if d, err := time.ParseDuration(cc.JobTimeout); err == nil && d > 0 {
+			return d
+		}
+	}
+	return DefaultJobTimeout
 }
 
 // ToRetryConfig converts CronConfig to cron.RetryConfig with defaults applied.

internal/cron/service_execution.go

@@ -6,7 +6,6 @@ import (
 	"log/slog"
 	"os"
 	"path/filepath"
-	"sync"
 	"time"
 
 	"github.com/adhocore/gronx"
@@ -146,11 +145,22 @@ func (cs *Service) runLoop(stopChan chan struct{}) {
 		case <-stopChan:
 			return
 		case <-ticker.C:
-			cs.checkJobs()
+			cs.safeCheckJobs()
 		}
 	}
 }
 
+// safeCheckJobs wraps checkJobs with panic recovery so a panic in any
+// check/claim logic doesn't kill the runLoop goroutine.
+func (cs *Service) safeCheckJobs() {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("cron: checkJobs panicked — runLoop continues", "panic", fmt.Sprint(r))
+		}
+	}()
+	cs.checkJobs()
+}
+
 func (cs *Service) checkJobs() {
 	cs.mu.Lock()
 
@@ -189,17 +199,16 @@ func (cs *Service) checkJobs() {
 	cs.saveUnsafe()
 	cs.mu.Unlock()
 
-	// Execute jobs in parallel — scheduler enforces per-session serialization
-	var wg sync.WaitGroup
+	// Execute jobs in parallel without blocking the runLoop.
+	// Previously wg.Wait() blocked here — if any job hung (e.g. LLM timeout,
+	// agent loop stuck), the entire cron scheduler would stop checking for new
+	// due jobs. Now each job runs independently with panic recovery.
 	for _, dj := range dueJobs {
-		wg.Add(1)
 		go func(id string, scheduledAtMS int64) {
-			defer wg.Done()
 			defer safego.Recover(nil, "job_id", id)
 			cs.executeJobByID(id, scheduledAtMS)
 		}(dj.id, dj.scheduledAtMS)
 	}
-	wg.Wait()
 }
 
 func (cs *Service) executeJobByID(jobID string, scheduledAtMS int64) {

internal/store/pg/cron_scheduler.go

@@ -3,8 +3,8 @@ package pg
 import (
 	"database/sql"
 	"errors"
+	"fmt"
 	"log/slog"
-	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -148,11 +148,22 @@ func (s *PGCronStore) runLoop() {
 		case <-s.stop:
 			return
 		case <-ticker.C:
-			s.checkAndRunDueJobs()
+			s.safeCheckAndRunDueJobs()
 		}
 	}
 }
 
+// safeCheckAndRunDueJobs wraps checkAndRunDueJobs with panic recovery
+// so a panic in any check/claim logic doesn't kill the runLoop goroutine.
+func (s *PGCronStore) safeCheckAndRunDueJobs() {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("cron: checkAndRunDueJobs panicked — runLoop continues", "panic", fmt.Sprint(r))
+		}
+	}()
+	s.checkAndRunDueJobs()
+}
+
 func (s *PGCronStore) checkAndRunDueJobs() {
 	dueJobs := s.GetDueJobs(time.Now())
 	if len(dueJobs) == 0 {
@@ -178,21 +189,22 @@ func (s *PGCronStore) checkAndRunDueJobs() {
 		return
 	}
 
-	// Execute jobs in parallel — scheduler enforces per-session serialization
-	var wg sync.WaitGroup
+	// Execute jobs in parallel without blocking the runLoop.
+	// Previously wg.Wait() blocked here — if any job hung (e.g. LLM timeout,
+	// agent loop stuck), the entire cron scheduler would stop checking for new
+	// due jobs. Now each job runs independently; cache is invalidated per-job.
 	for _, job := range claimedJobs {
-		wg.Add(1)
 		go func(job store.CronJob) {
-			defer wg.Done()
+			defer func() {
+				if r := recover(); r != nil {
+					slog.Error("cron: job execution panicked", "job_id", job.ID, "job_name", job.Name, "panic", fmt.Sprint(r))
+				}
+				// Invalidate cache so the next tick picks up the updated next_run_at.
+				s.InvalidateCache()
+			}()
 			s.executeOneJob(job, handler, true)
 		}(job)
 	}
-	wg.Wait()
-
-	// Invalidate cache after job execution changed next_run_at values
-	s.mu.Lock()
-	s.cacheLoaded = false
-	s.mu.Unlock()
 }
 
 // executeOneJob runs a single cron job with retry, logs the result, and updates next_run_at.

internal/store/sqlitestore/cron_scheduler.go

@@ -5,8 +5,8 @@ package sqlitestore
 import (
 	"database/sql"
 	"errors"
+	"fmt"
 	"log/slog"
-	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -154,11 +154,22 @@ func (s *SQLiteCronStore) runLoop() {
 		case <-s.stop:
 			return
 		case <-ticker.C:
-			s.checkAndRunDueJobs()
+			s.safeCheckAndRunDueJobs()
 		}
 	}
 }
 
+// safeCheckAndRunDueJobs wraps checkAndRunDueJobs with panic recovery
+// so a panic in any check/claim logic doesn't kill the runLoop goroutine.
+func (s *SQLiteCronStore) safeCheckAndRunDueJobs() {
+	defer func() {
+		if r := recover(); r != nil {
+			slog.Error("cron: checkAndRunDueJobs panicked — runLoop continues", "panic", fmt.Sprint(r))
+		}
+	}()
+	s.checkAndRunDueJobs()
+}
+
 func (s *SQLiteCronStore) checkAndRunDueJobs() {
 	dueJobs := s.GetDueJobs(time.Now())
 	if len(dueJobs) == 0 {
@@ -184,20 +195,22 @@ func (s *SQLiteCronStore) checkAndRunDueJobs() {
 		return
 	}
 
-	// Execute jobs in parallel — scheduler enforces per-session serialization.
-	var wg sync.WaitGroup
+	// Execute jobs in parallel without blocking the runLoop.
+	// Previously wg.Wait() blocked here — if any job hung (e.g. LLM timeout,
+	// agent loop stuck), the entire cron scheduler would stop checking for new
+	// due jobs. Now each job runs independently; cache is invalidated per-job.
 	for _, job := range claimedJobs {
-		wg.Add(1)
 		go func(job store.CronJob) {
-			defer wg.Done()
+			defer func() {
+				if r := recover(); r != nil {
+					slog.Error("cron: job execution panicked", "job_id", job.ID, "job_name", job.Name, "panic", fmt.Sprint(r))
+				}
+				// Invalidate cache so the next tick picks up the updated next_run_at.
+				s.InvalidateCache()
+			}()
 			s.executeOneJob(job, handler, true)
 		}(job)
 	}
-	wg.Wait()
-
-	s.mu.Lock()
-	s.cacheLoaded = false
-	s.mu.Unlock()
 }
 
 // executeOneJob runs a claimed job. When reloadClaimed is true (scheduler path),