fix(cron): harden toggle and scheduler state (#540)

- recompute next_run_at transactionally on toggle/update instead of non-atomic read/compute/write
- keep repeated EnableJob(true) idempotent by preserving existing next_run_at
- reject re-enabling expired one-shot at jobs with ErrCronJobNoFutureRun
- claim due jobs conditionally, reload before execution, conditional writeback
- align PG and SQLite behavior with shared helpers (DRY)
- atomic scheduler writeback via single UPDATE with CASE expression

Closes #378

internal/gateway/methods/cron.go

@@ -3,6 +3,7 @@ package methods
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"log/slog"
 	"regexp"
 
@@ -155,7 +156,11 @@ func (m *CronMethods) handleToggle(ctx context.Context, client *gateway.Client,
 	}
 
 	if err := m.service.EnableJob(ctx, params.JobID, params.Enabled); err != nil {
-		client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrNotFound, err.Error()))
+		code := protocol.ErrInvalidRequest
+		if errors.Is(err, store.ErrCronJobNotFound) {
+			code = protocol.ErrNotFound
+		}
+		client.SendResponse(protocol.NewErrorResponse(req.ID, code, err.Error()))
 		return
 	}
 
@@ -200,7 +205,11 @@ func (m *CronMethods) handleUpdate(ctx context.Context, client *gateway.Client,
 
 	job, err := m.service.UpdateJob(ctx, jobID, params.Patch)
 	if err != nil {
-		client.SendResponse(protocol.NewErrorResponse(req.ID, protocol.ErrInvalidRequest, err.Error()))
+		code := protocol.ErrInvalidRequest
+		if errors.Is(err, store.ErrCronJobNotFound) {
+			code = protocol.ErrNotFound
+		}
+		client.SendResponse(protocol.NewErrorResponse(req.ID, code, err.Error()))
 		return
 	}
 

internal/store/cron_store.go

@@ -2,11 +2,20 @@ package store
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"sort"
 	"time"
 
+	"github.com/adhocore/gronx"
 	"github.com/google/uuid"
 )
 
+var (
+	ErrCronJobNotFound    = errors.New("cron job not found")
+	ErrCronJobNoFutureRun = errors.New("cron job has no future run")
+)
+
 // CronJob represents a scheduled job.
 type CronJob struct {
 	ID             string       `json:"id"`
@@ -124,3 +133,202 @@ type CronStore interface {
 type CacheInvalidatable interface {
 	InvalidateCache()
 }
+
+// CronJobMutableState holds the mutable fields of a cron job loaded within a
+// transaction for read-compute-write operations (EnableJob, UpdateJob).
+type CronJobMutableState struct {
+	Enabled   bool
+	Schedule  CronSchedule
+	NextRunAt *time.Time
+	Payload   CronPayload
+}
+
+// ComputeNextRun calculates the next run time for a cron schedule.
+// defaultTZ is used for cron expressions that do not specify a per-job timezone.
+func ComputeNextRun(schedule *CronSchedule, now time.Time, defaultTZ string) *time.Time {
+	switch schedule.Kind {
+	case "at":
+		if schedule.AtMS != nil {
+			t := time.UnixMilli(*schedule.AtMS)
+			if t.After(now) {
+				return &t
+			}
+		}
+		return nil
+	case "every":
+		if schedule.EveryMS != nil && *schedule.EveryMS > 0 {
+			t := now.Add(time.Duration(*schedule.EveryMS) * time.Millisecond)
+			return &t
+		}
+		return nil
+	case "cron":
+		if schedule.Expr == "" {
+			return nil
+		}
+		tz := schedule.TZ
+		if tz == "" {
+			tz = defaultTZ
+		}
+		evalTime := now
+		if tz != "" {
+			if loc, err := time.LoadLocation(tz); err == nil {
+				evalTime = now.In(loc)
+			}
+		}
+		nextTime, err := gronx.NextTickAfter(schedule.Expr, evalTime, false)
+		if err != nil {
+			return nil
+		}
+		utcNext := nextTime.UTC()
+		return &utcNext
+	default:
+		return nil
+	}
+}
+
+// NextRunForSchedule resolves the persisted next_run_at for a given schedule state.
+func NextRunForSchedule(schedule *CronSchedule, enabled bool, now time.Time, defaultTZ string) (*time.Time, error) {
+	if !enabled {
+		return nil, nil
+	}
+
+	next := ComputeNextRun(schedule, now, defaultTZ)
+	if next != nil {
+		return next, nil
+	}
+
+	switch schedule.Kind {
+	case "at":
+		return nil, fmt.Errorf("%w: at schedule is already in the past", ErrCronJobNoFutureRun)
+	case "cron":
+		return nil, fmt.Errorf("%w: cron schedule has no valid next execution", ErrCronJobNoFutureRun)
+	case "every":
+		return nil, fmt.Errorf("%w: every schedule has no valid interval", ErrCronJobNoFutureRun)
+	default:
+		return nil, fmt.Errorf("%w: unsupported schedule kind %q", ErrCronJobNoFutureRun, schedule.Kind)
+	}
+}
+
+// NextRunForToggle returns the next run state after explicitly enabling or
+// disabling a cron job. Disabling clears next_run_at immediately so the
+// scheduler stops seeing the job as runnable.
+func NextRunForToggle(schedule *CronSchedule, enabled, currentlyEnabled bool, currentNextRunAt *time.Time, now time.Time, defaultTZ string) (*time.Time, error) {
+	if !enabled {
+		return nil, nil
+	}
+	if currentlyEnabled && currentNextRunAt != nil {
+		next := *currentNextRunAt
+		return &next, nil
+	}
+	return NextRunForSchedule(schedule, true, now, defaultTZ)
+}
+
+// MergeCronSchedule applies a partial schedule patch on top of the current schedule.
+func MergeCronSchedule(current CronSchedule, patch *CronSchedule) CronSchedule {
+	if patch == nil {
+		return current
+	}
+
+	newKind := patch.Kind
+	if newKind == "" {
+		newKind = current.Kind
+	}
+
+	merged := CronSchedule{Kind: newKind}
+	switch newKind {
+	case "cron":
+		if patch.Expr != "" {
+			merged.Expr = patch.Expr
+		} else if current.Kind == newKind {
+			merged.Expr = current.Expr
+		}
+		if patch.TZ != "" {
+			merged.TZ = patch.TZ
+		} else if current.Kind == newKind {
+			merged.TZ = current.TZ
+		}
+	case "every":
+		if patch.EveryMS != nil {
+			merged.EveryMS = patch.EveryMS
+		} else if current.Kind == newKind {
+			merged.EveryMS = current.EveryMS
+		}
+	case "at":
+		if patch.AtMS != nil {
+			merged.AtMS = patch.AtMS
+		} else if current.Kind == newKind {
+			merged.AtMS = current.AtMS
+		}
+	}
+
+	return merged
+}
+
+// ValidateCronSchedule checks structural schedule validity without evaluating future run existence.
+func ValidateCronSchedule(schedule *CronSchedule) error {
+	switch schedule.Kind {
+	case "cron":
+		if schedule.Expr == "" {
+			return fmt.Errorf("cron schedule requires expr")
+		}
+		if !gronx.New().IsValid(schedule.Expr) {
+			return fmt.Errorf("invalid cron expression: %s", schedule.Expr)
+		}
+		if schedule.TZ != "" {
+			if _, err := time.LoadLocation(schedule.TZ); err != nil {
+				return fmt.Errorf("invalid timezone: %s", schedule.TZ)
+			}
+		}
+	case "every":
+		if schedule.EveryMS == nil || *schedule.EveryMS <= 0 {
+			return fmt.Errorf("every schedule requires positive everyMs")
+		}
+	case "at":
+		if schedule.AtMS == nil {
+			return fmt.Errorf("at schedule requires atMs")
+		}
+	default:
+		return fmt.Errorf("invalid schedule kind: %s", schedule.Kind)
+	}
+	return nil
+}
+
+// ApplyCronScheduleUpdates populates the update map with the column values
+// for a fully-resolved cron schedule (after merge + validation).
+func ApplyCronScheduleUpdates(updates map[string]any, schedule CronSchedule) {
+	updates["schedule_kind"] = schedule.Kind
+
+	switch schedule.Kind {
+	case "cron":
+		updates["cron_expression"] = schedule.Expr
+		if schedule.TZ != "" {
+			updates["timezone"] = schedule.TZ
+		} else {
+			updates["timezone"] = nil
+		}
+		updates["interval_ms"] = nil
+		updates["run_at"] = nil
+	case "every":
+		updates["cron_expression"] = nil
+		updates["timezone"] = nil
+		updates["interval_ms"] = *schedule.EveryMS
+		updates["run_at"] = nil
+	case "at":
+		runAt := time.UnixMilli(*schedule.AtMS)
+		updates["cron_expression"] = nil
+		updates["timezone"] = nil
+		updates["interval_ms"] = nil
+		updates["run_at"] = runAt
+	}
+}
+
+// SortedUpdateColumns returns the map keys in sorted order for deterministic
+// SQL generation.
+func SortedUpdateColumns(updates map[string]any) []string {
+	cols := make([]string, 0, len(updates))
+	for col := range updates {
+		cols = append(cols, col)
+	}
+	sort.Strings(cols)
+	return cols
+}

internal/store/cron_store_test.go

@@ -0,0 +1,113 @@
+package store
+
+import (
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestNextRunForToggle_DisableClearsNextRun(t *testing.T) {
+	now := time.Date(2026, time.March, 28, 12, 0, 0, 0, time.UTC)
+	schedule := &CronSchedule{
+		Kind:    "every",
+		EveryMS: int64Ptr(60_000),
+	}
+
+	next, err := NextRunForToggle(schedule, false, true, timePtr(now.Add(time.Minute)), now, "")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if next != nil {
+		t.Fatalf("expected disable toggle to clear next_run_at, got %v", next)
+	}
+}
+
+func TestNextRunForToggle_EnableRecomputesEverySchedule(t *testing.T) {
+	now := time.Date(2026, time.March, 28, 12, 0, 0, 0, time.UTC)
+	schedule := &CronSchedule{
+		Kind:    "every",
+		EveryMS: int64Ptr(60_000),
+	}
+
+	next, err := NextRunForToggle(schedule, true, false, nil, now, "")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if next == nil {
+		t.Fatal("expected enable toggle to recompute next_run_at")
+	}
+
+	want := now.Add(time.Minute)
+	if !next.Equal(want) {
+		t.Fatalf("got %v, want %v", next, want)
+	}
+}
+
+func TestNextRunForToggle_EnableUsesDefaultTimezoneForCronSchedule(t *testing.T) {
+	now := time.Date(2026, time.March, 28, 12, 0, 0, 0, time.UTC)
+	schedule := &CronSchedule{
+		Kind: "cron",
+		Expr: "0 9 * * *",
+	}
+
+	next, err := NextRunForToggle(schedule, true, false, nil, now, "America/Toronto")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if next == nil {
+		t.Fatal("expected enable toggle to compute next_run_at for cron schedule")
+	}
+
+	want := time.Date(2026, time.March, 28, 13, 0, 0, 0, time.UTC)
+	if !next.Equal(want) {
+		t.Fatalf("got %v, want %v", next, want)
+	}
+}
+
+func TestNextRunForToggle_AlreadyEnabledPreservesCurrentNextRun(t *testing.T) {
+	now := time.Date(2026, time.March, 28, 12, 0, 0, 0, time.UTC)
+	currentNextRun := now.Add(5 * time.Minute)
+	schedule := &CronSchedule{
+		Kind:    "every",
+		EveryMS: int64Ptr(60_000),
+	}
+
+	next, err := NextRunForToggle(schedule, true, true, &currentNextRun, now.Add(time.Minute), "")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if next == nil {
+		t.Fatal("expected preserved next run")
+	}
+	if !next.Equal(currentNextRun) {
+		t.Fatalf("got %v, want %v", next, currentNextRun)
+	}
+}
+
+func TestNextRunForToggle_ExpiredAtReturnsError(t *testing.T) {
+	now := time.Date(2026, time.March, 28, 12, 0, 0, 0, time.UTC)
+	past := now.Add(-time.Minute).UnixMilli()
+	schedule := &CronSchedule{
+		Kind: "at",
+		AtMS: &past,
+	}
+
+	next, err := NextRunForToggle(schedule, true, false, nil, now, "")
+	if next != nil {
+		t.Fatalf("expected nil next run, got %v", next)
+	}
+	if err == nil {
+		t.Fatal("expected error for expired at schedule")
+	}
+	if !errors.Is(err, ErrCronJobNoFutureRun) {
+		t.Fatalf("got %v, want ErrCronJobNoFutureRun", err)
+	}
+}
+
+func int64Ptr(v int64) *int64 {
+	return &v
+}
+
+func timePtr(v time.Time) *time.Time {
+	return &v
+}

internal/store/pg/cron_crud.go

@@ -185,25 +185,35 @@ func (s *PGCronStore) EnableJob(ctx context.Context, jobID string, enabled bool)
 		return fmt.Errorf("invalid job ID: %s", jobID)
 	}
 
-	q := "UPDATE cron_jobs SET enabled = $1, updated_at = $2 WHERE id = $3"
-	args := []any{enabled, time.Now(), id}
-
-	if !store.IsCrossTenant(ctx) {
-		tid := store.TenantIDFromContext(ctx)
-		if tid == uuid.Nil {
-			return fmt.Errorf("tenant_id required")
-		}
-		q += fmt.Sprintf(" AND tenant_id = $%d", len(args)+1)
-		args = append(args, tid)
-	}
-
-	res, err := s.db.ExecContext(ctx, q, args...)
+	tx, err := s.db.BeginTx(ctx, nil)
 	if err != nil {
 		return err
 	}
-	if n, _ := res.RowsAffected(); n == 0 {
-		return fmt.Errorf("job not found")
+	defer tx.Rollback()
+
+	current, err := s.lockCronJobForMutation(ctx, tx, id, false)
+	if err != nil {
+		return err
 	}
-	s.cacheLoaded = false
+
+	now := time.Now()
+	nextRun, err := store.NextRunForToggle(&current.Schedule, enabled, current.Enabled, current.NextRunAt, now, s.defaultTZ)
+	if err != nil {
+		return err
+	}
+
+	if err := execCronJobUpdateTx(ctx, tx, id, map[string]any{
+		"enabled":     enabled,
+		"next_run_at": nextRun,
+		"updated_at":  now,
+	}); err != nil {
+		return err
+	}
+
+	if err := tx.Commit(); err != nil {
+		return err
+	}
+
+	s.InvalidateCache()
 	return nil
 }

internal/store/pg/cron_scan.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/adhocore/gronx"
 	"github.com/google/uuid"
 
 	"github.com/nextlevelbuilder/goclaw/internal/store"
@@ -127,42 +126,5 @@ func scanCronSingleRow(row *sql.Row) (*store.CronJob, error) {
 // defaultTZ is the gateway-level fallback IANA timezone used when the
 // schedule itself does not specify a timezone (existing jobs with TZ = NULL).
 func computeNextRun(schedule *store.CronSchedule, now time.Time, defaultTZ string) *time.Time {
-	switch schedule.Kind {
-	case "at":
-		if schedule.AtMS != nil {
-			t := time.UnixMilli(*schedule.AtMS)
-			if t.After(now) {
-				return &t
-			}
-		}
-		return nil
-	case "every":
-		if schedule.EveryMS != nil && *schedule.EveryMS > 0 {
-			t := now.Add(time.Duration(*schedule.EveryMS) * time.Millisecond)
-			return &t
-		}
-		return nil
-	case "cron":
-		if schedule.Expr == "" {
-			return nil
-		}
-		tz := schedule.TZ
-		if tz == "" {
-			tz = defaultTZ
-		}
-		evalTime := now
-		if tz != "" {
-			if loc, err := time.LoadLocation(tz); err == nil {
-				evalTime = now.In(loc)
-			}
-		}
-		nextTime, err := gronx.NextTickAfter(schedule.Expr, evalTime, false)
-		if err != nil {
-			return nil
-		}
-		utcNext := nextTime.UTC()
-		return &utcNext
-	default:
-		return nil
-	}
+	return store.ComputeNextRun(schedule, now, defaultTZ)
 }

internal/store/pg/cron_scheduler.go

@@ -1,6 +1,8 @@
 package pg
 
 import (
+	"database/sql"
+	"errors"
 	"log/slog"
 	"sync"
 	"time"
@@ -146,16 +148,20 @@ func (s *PGCronStore) checkAndRunDueJobs() {
 		return
 	}
 
-	// Clear next_run for all due jobs first to prevent duplicate fires
+	now := time.Now()
+	var claimedJobs []store.CronJob
 	for _, job := range dueJobs {
-		if id, parseErr := uuid.Parse(job.ID); parseErr == nil {
-			s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = NULL WHERE id = $1", id)
+		if id, parseErr := uuid.Parse(job.ID); parseErr == nil && s.claimDueJob(id, now) {
+			claimedJobs = append(claimedJobs, job)
 		}
 	}
+	if len(claimedJobs) == 0 {
+		return
+	}
 
 	// Execute jobs in parallel — scheduler enforces per-session serialization
 	var wg sync.WaitGroup
-	for _, job := range dueJobs {
+	for _, job := range claimedJobs {
 		wg.Add(1)
 		go func(job store.CronJob) {
 			defer wg.Done()
@@ -172,6 +178,15 @@ func (s *PGCronStore) checkAndRunDueJobs() {
 
 // executeOneJob runs a single cron job with retry, logs the result, and updates next_run_at.
 func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.CronJob) (*store.CronJobResult, error)) {
+	if id, parseErr := uuid.Parse(job.ID); parseErr == nil {
+		freshJob, ok := s.loadClaimedJob(id)
+		if !ok {
+			slog.Info("cron job skipped after claim state changed", "id", job.ID)
+			return
+		}
+		job = *freshJob
+	}
+
 	startTime := time.Now()
 
 	// Wrap handler to fit ExecuteWithRetry's (string, error) signature
@@ -237,9 +252,16 @@ func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.C
 	} else if id, parseErr := uuid.Parse(job.ID); parseErr == nil {
 		schedule := job.Schedule
 		next := computeNextRun(&schedule, now, s.defaultTZ)
+		var nextRunValue any
+		if next != nil {
+			nextRunValue = *next
+		}
 		s.db.ExecContext(s.baseCtx,
-			"UPDATE cron_jobs SET last_run_at = $1, last_status = $2, last_error = $3, next_run_at = $4, updated_at = $5 WHERE id = $6",
-			now, status, lastError, next, now, id,
+			`UPDATE cron_jobs SET
+			 last_run_at = $1, last_status = $2, last_error = $3, updated_at = $4,
+			 next_run_at = CASE WHEN enabled = true AND next_run_at IS NULL THEN $5 ELSE next_run_at END
+			 WHERE id = $6`,
+			now, status, lastError, now, nextRunValue, id,
 		)
 	}
 
@@ -251,3 +273,42 @@ func (s *PGCronStore) executeOneJob(job store.CronJob, handler func(job *store.C
 	}
 	s.emitEvent(evt)
 }
+
+func (s *PGCronStore) claimDueJob(id uuid.UUID, now time.Time) bool {
+	res, err := s.db.ExecContext(
+		s.baseCtx,
+		`UPDATE cron_jobs
+		 SET next_run_at = NULL
+		 WHERE id = $1 AND enabled = true AND next_run_at IS NOT NULL AND next_run_at <= $2`,
+		id,
+		now,
+	)
+	if err != nil {
+		slog.Warn("cron: failed to claim due job", "id", id, "error", err)
+		return false
+	}
+
+	n, _ := res.RowsAffected()
+	return n == 1
+}
+
+func (s *PGCronStore) loadClaimedJob(id uuid.UUID) (*store.CronJob, bool) {
+	row := s.db.QueryRowContext(
+		s.baseCtx,
+		`SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
+		 interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error,
+		 created_at, updated_at
+		 FROM cron_jobs
+		 WHERE id = $1 AND enabled = true AND next_run_at IS NULL`,
+		id,
+	)
+	job, err := scanCronSingleRow(row)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, false
+	}
+	if err != nil {
+		slog.Warn("cron: failed to reload claimed job", "id", id, "error", err)
+		return nil, false
+	}
+	return job, true
+}

internal/store/pg/cron_update.go

@@ -2,11 +2,13 @@ package pg
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
+	"errors"
 	"fmt"
+	"strings"
 	"time"
 
-	"github.com/adhocore/gronx"
 	"github.com/google/uuid"
 
 	"github.com/nextlevelbuilder/goclaw/internal/store"
@@ -18,140 +20,31 @@ func (s *PGCronStore) UpdateJob(ctx context.Context, jobID string, patch store.C
 		return nil, fmt.Errorf("invalid job ID: %s", jobID)
 	}
 
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	current, err := s.lockCronJobForMutation(ctx, tx, id, true)
+	if err != nil {
+		return nil, err
+	}
+
+	now := time.Now()
 	updates := make(map[string]any)
+	effectiveEnabled := current.Enabled
+	if patch.Enabled != nil {
+		effectiveEnabled = *patch.Enabled
+		updates["enabled"] = effectiveEnabled
+	}
+
 	if patch.Name != "" {
 		updates["name"] = patch.Name
 	}
-	if patch.Enabled != nil {
-		updates["enabled"] = *patch.Enabled
-	}
-	// Build tenant WHERE suffix for internal reads (defense-in-depth).
-	tenantSuffix := ""
-	var tenantArg []any
-	if !store.IsCrossTenant(ctx) {
-		if tid := store.TenantIDFromContext(ctx); tid != uuid.Nil {
-			tenantSuffix = " AND tenant_id = $2"
-			tenantArg = append(tenantArg, tid)
-		}
-	}
-
-	if patch.Schedule != nil {
-		// Fetch current schedule to merge with patch (partial updates)
-		var curKind string
-		var curExpr, curTZ *string
-		var curIntervalMS *int64
-		var curRunAt *time.Time
-		if err := s.db.QueryRowContext(ctx,
-			"SELECT schedule_kind, cron_expression, timezone, interval_ms, run_at FROM cron_jobs WHERE id = $1"+tenantSuffix,
-			append([]any{id}, tenantArg...)...,
-		).Scan(&curKind, &curExpr, &curTZ, &curIntervalMS, &curRunAt); err != nil {
-			return nil, fmt.Errorf("failed to fetch current schedule for job %s: %w", jobID, err)
-		}
-
-		// Resolve the effective schedule kind
-		newKind := patch.Schedule.Kind
-		if newKind == "" {
-			newKind = curKind // keep current kind if not specified
-		}
-		updates["schedule_kind"] = newKind
-
-		// Set type-specific fields and clear others
-		switch newKind {
-		case "cron":
-			if patch.Schedule.Expr != "" {
-				updates["cron_expression"] = patch.Schedule.Expr
-			}
-			if patch.Schedule.TZ != "" {
-				if _, err := time.LoadLocation(patch.Schedule.TZ); err != nil {
-					return nil, fmt.Errorf("invalid timezone: %s", patch.Schedule.TZ)
-				}
-				updates["timezone"] = patch.Schedule.TZ
-			}
-			// Clear other type fields when switching to cron
-			if curKind != "cron" {
-				updates["interval_ms"] = nil
-				updates["run_at"] = nil
-			}
-		case "every":
-			if patch.Schedule.EveryMS != nil {
-				updates["interval_ms"] = *patch.Schedule.EveryMS
-			}
-			// Clear other type fields when switching to every
-			if curKind != "every" {
-				updates["cron_expression"] = nil
-				updates["timezone"] = nil
-				updates["run_at"] = nil
-			}
-		case "at":
-			if patch.Schedule.AtMS != nil {
-				t := time.UnixMilli(*patch.Schedule.AtMS)
-				updates["run_at"] = t
-			}
-			// Clear other type fields when switching to at
-			if curKind != "at" {
-				updates["cron_expression"] = nil
-				updates["timezone"] = nil
-				updates["interval_ms"] = nil
-			}
-		}
-
-		// Build merged schedule for recomputing next_run_at
-		merged := store.CronSchedule{Kind: newKind}
-		switch newKind {
-		case "cron":
-			if patch.Schedule.Expr != "" {
-				merged.Expr = patch.Schedule.Expr
-			} else if curExpr != nil {
-				merged.Expr = *curExpr
-			}
-			if patch.Schedule.TZ != "" {
-				merged.TZ = patch.Schedule.TZ
-			} else if curTZ != nil && newKind == curKind {
-				merged.TZ = *curTZ
-			}
-		case "every":
-			if patch.Schedule.EveryMS != nil {
-				merged.EveryMS = patch.Schedule.EveryMS
-			} else if curIntervalMS != nil {
-				merged.EveryMS = curIntervalMS
-			}
-		case "at":
-			if patch.Schedule.AtMS != nil {
-				merged.AtMS = patch.Schedule.AtMS
-			} else if curRunAt != nil {
-				ms := curRunAt.UnixMilli()
-				merged.AtMS = &ms
-			}
-		}
-
-		// Validate the merged schedule before applying
-		switch merged.Kind {
-		case "cron":
-			if merged.Expr == "" {
-				return nil, fmt.Errorf("cron schedule requires expr")
-			}
-			gx := gronx.New()
-			if !gx.IsValid(merged.Expr) {
-				return nil, fmt.Errorf("invalid cron expression: %s", merged.Expr)
-			}
-		case "every":
-			if merged.EveryMS == nil || *merged.EveryMS <= 0 {
-				return nil, fmt.Errorf("every schedule requires positive everyMs")
-			}
-		case "at":
-			if merged.AtMS == nil {
-				return nil, fmt.Errorf("at schedule requires atMs")
-			}
-		}
-
-		next := computeNextRun(&merged, time.Now(), s.defaultTZ)
-		updates["next_run_at"] = next
-	}
 	if patch.DeleteAfterRun != nil {
 		updates["delete_after_run"] = *patch.DeleteAfterRun
 	}
-
-	// Update agent_id column
 	if patch.AgentID != nil {
 		if *patch.AgentID == "" {
 			updates["agent_id"] = nil
@@ -160,60 +53,170 @@ func (s *PGCronStore) UpdateJob(ctx context.Context, jobID string, patch store.C
 		}
 	}
 
-	// Update payload JSONB — fetch current, merge patch fields, re-serialize
+	if patch.Schedule != nil {
+		merged := store.MergeCronSchedule(current.Schedule, patch.Schedule)
+		if err := store.ValidateCronSchedule(&merged); err != nil {
+			return nil, err
+		}
+
+		store.ApplyCronScheduleUpdates(updates, merged)
+
+		nextRun, err := store.NextRunForSchedule(&merged, effectiveEnabled, now, s.defaultTZ)
+		if err != nil {
+			return nil, err
+		}
+		updates["next_run_at"] = nextRun
+	} else if patch.Enabled != nil {
+		nextRun, err := store.NextRunForToggle(&current.Schedule, effectiveEnabled, current.Enabled, current.NextRunAt, now, s.defaultTZ)
+		if err != nil {
+			return nil, err
+		}
+		updates["next_run_at"] = nextRun
+	}
+
 	needsPayloadUpdate := patch.Message != "" || patch.Deliver != nil || patch.Channel != nil || patch.To != nil || patch.WakeHeartbeat != nil
 	if needsPayloadUpdate {
-		var payloadJSON []byte
-		if scanErr := s.db.QueryRowContext(ctx, "SELECT payload FROM cron_jobs WHERE id = $1"+tenantSuffix, append([]any{id}, tenantArg...)...).Scan(&payloadJSON); scanErr == nil {
-			var payload store.CronPayload
-			if len(payloadJSON) > 0 {
-				if err := json.Unmarshal(payloadJSON, &payload); err != nil {
-					return nil, fmt.Errorf("failed to parse existing payload for job %s: %w", jobID, err)
-				}
-			}
-
-			if patch.Message != "" {
-				payload.Message = patch.Message
-			}
-			if patch.Deliver != nil {
-				payload.Deliver = *patch.Deliver
-			}
-			if patch.Channel != nil {
-				payload.Channel = *patch.Channel
-			}
-			if patch.To != nil {
-				payload.To = *patch.To
-			}
-			if patch.WakeHeartbeat != nil {
-				payload.WakeHeartbeat = *patch.WakeHeartbeat
-			}
-
-			merged, err := json.Marshal(payload)
-			if err != nil {
-				return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err)
-			}
-			updates["payload"] = merged
+		payload := current.Payload
+		if patch.Message != "" {
+			payload.Message = patch.Message
 		}
+		if patch.Deliver != nil {
+			payload.Deliver = *patch.Deliver
+		}
+		if patch.Channel != nil {
+			payload.Channel = *patch.Channel
+		}
+		if patch.To != nil {
+			payload.To = *patch.To
+		}
+		if patch.WakeHeartbeat != nil {
+			payload.WakeHeartbeat = *patch.WakeHeartbeat
+		}
+
+		mergedPayload, err := json.Marshal(payload)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err)
+		}
+		updates["payload"] = mergedPayload
 	}
 
-	updates["updated_at"] = time.Now()
-
-	var execErr error
-	if store.IsCrossTenant(ctx) {
-		execErr = execMapUpdate(ctx, s.db, "cron_jobs", id, updates)
-	} else {
-		tid := store.TenantIDFromContext(ctx)
-		if tid == uuid.Nil {
-			execErr = fmt.Errorf("tenant_id required for update")
-		} else {
-			execErr = execMapUpdateWhereTenant(ctx, s.db, "cron_jobs", updates, id, tid)
-		}
-	}
-	if err := execErr; err != nil {
+	updates["updated_at"] = now
+	if err := execCronJobUpdateTx(ctx, tx, id, updates); err != nil {
 		return nil, err
 	}
 
-	s.cacheLoaded = false
+	if err := tx.Commit(); err != nil {
+		return nil, err
+	}
+
+	s.InvalidateCache()
 	job, _ := s.scanJob(ctx, id)
 	return job, nil
 }
+
+func (s *PGCronStore) lockCronJobForMutation(ctx context.Context, tx *sql.Tx, id uuid.UUID, loadPayload bool) (*store.CronJobMutableState, error) {
+	q := `SELECT enabled, schedule_kind, cron_expression, run_at, timezone, interval_ms, next_run_at, payload
+		FROM cron_jobs WHERE id = $1`
+	args := []any{id}
+
+	if !store.IsCrossTenant(ctx) {
+		tid := store.TenantIDFromContext(ctx)
+		if tid == uuid.Nil {
+			return nil, fmt.Errorf("tenant_id required")
+		}
+		q += fmt.Sprintf(" AND tenant_id = $%d", len(args)+1)
+		args = append(args, tid)
+	}
+	q += " FOR UPDATE"
+
+	var (
+		state        store.CronJobMutableState
+		scheduleKind string
+		cronExpr     *string
+		runAt        *time.Time
+		tz           *string
+		intervalMS   *int64
+		nextRunAt    *time.Time
+		payloadJSON  []byte
+	)
+
+	if err := tx.QueryRowContext(ctx, q, args...).Scan(
+		&state.Enabled,
+		&scheduleKind,
+		&cronExpr,
+		&runAt,
+		&tz,
+		&intervalMS,
+		&nextRunAt,
+		&payloadJSON,
+	); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, store.ErrCronJobNotFound
+		}
+		return nil, err
+	}
+
+	state.Schedule = store.CronSchedule{Kind: scheduleKind}
+	if cronExpr != nil {
+		state.Schedule.Expr = *cronExpr
+	}
+	if runAt != nil {
+		ms := runAt.UnixMilli()
+		state.Schedule.AtMS = &ms
+	}
+	if tz != nil {
+		state.Schedule.TZ = *tz
+	}
+	if intervalMS != nil {
+		state.Schedule.EveryMS = intervalMS
+	}
+	state.NextRunAt = nextRunAt
+
+	if loadPayload && len(payloadJSON) > 0 {
+		if err := json.Unmarshal(payloadJSON, &state.Payload); err != nil {
+			return nil, fmt.Errorf("failed to parse existing payload for job %s: %w", id, err)
+		}
+	}
+
+	return &state, nil
+}
+
+
+func execCronJobUpdateTx(ctx context.Context, tx *sql.Tx, id uuid.UUID, updates map[string]any) error {
+	if len(updates) == 0 {
+		return nil
+	}
+
+	var (
+		setClauses []string
+		args       []any
+	)
+	for idx, col := range store.SortedUpdateColumns(updates) {
+		if !validColumnName.MatchString(col) {
+			return fmt.Errorf("invalid column name: %q", col)
+		}
+		setClauses = append(setClauses, fmt.Sprintf("%s = $%d", col, idx+1))
+		args = append(args, updates[col])
+	}
+
+	args = append(args, id)
+	q := fmt.Sprintf("UPDATE cron_jobs SET %s WHERE id = $%d", strings.Join(setClauses, ", "), len(args))
+	if !store.IsCrossTenant(ctx) {
+		tid := store.TenantIDFromContext(ctx)
+		if tid == uuid.Nil {
+			return fmt.Errorf("tenant_id required")
+		}
+		args = append(args, tid)
+		q += fmt.Sprintf(" AND tenant_id = $%d", len(args))
+	}
+
+	res, err := tx.ExecContext(ctx, q, args...)
+	if err != nil {
+		return err
+	}
+	if n, _ := res.RowsAffected(); n == 0 {
+		return store.ErrCronJobNotFound
+	}
+	return nil
+}
+

internal/store/sqlitestore/cron.go

@@ -11,7 +11,6 @@ import (
 	"sync"
 	"time"
 
-	"github.com/adhocore/gronx"
 	"github.com/google/uuid"
 
 	"github.com/nextlevelbuilder/goclaw/internal/cron"
@@ -24,6 +23,7 @@ const defaultCronCacheTTL = 2 * time.Minute
 type SQLiteCronStore struct {
 	db        *sql.DB
 	mu        sync.Mutex
+	writeMu   sync.Mutex
 	baseCtx   context.Context
 	cancelCtx context.CancelFunc
 	onJob     func(job *store.CronJob) (*store.CronJobResult, error)
@@ -119,14 +119,14 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
 	var name, scheduleKind string
 	var enabled, deleteAfterRun bool
 	var cronExpr, tz, lastStatus, lastError *string
-	var runAt, nextRunAt, lastRunAt *time.Time
+	var runAt, nextRunAt, lastRunAt nullSqliteTime
 	var intervalMS *int64
 	var payloadJSON []byte
-	var createdAt, updatedAt time.Time
+	createdAt, updatedAt := scanTimePair()
 
 	err := row.Scan(&id, &tenantID, &agentID, &userID, &name, &enabled, &scheduleKind, &cronExpr, &runAt, &tz,
 		&intervalMS, &payloadJSON, &deleteAfterRun, &nextRunAt, &lastRunAt, &lastStatus, &lastError,
-		&createdAt, &updatedAt)
+		createdAt, updatedAt)
 	if err != nil {
 		return nil, err
 	}
@@ -145,8 +145,8 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
 		Enabled:        enabled,
 		Schedule:       store.CronSchedule{Kind: scheduleKind},
 		Payload:        payload,
-		CreatedAtMS:    createdAt.UnixMilli(),
-		UpdatedAtMS:    updatedAt.UnixMilli(),
+		CreatedAtMS:    createdAt.Time.UnixMilli(),
+		UpdatedAtMS:    updatedAt.Time.UnixMilli(),
 		DeleteAfterRun: deleteAfterRun,
 	}
 
@@ -159,8 +159,8 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
 	if cronExpr != nil {
 		job.Schedule.Expr = *cronExpr
 	}
-	if runAt != nil {
-		ms := runAt.UnixMilli()
+	if runAt.Valid {
+		ms := runAt.Time.UnixMilli()
 		job.Schedule.AtMS = &ms
 	}
 	if intervalMS != nil {
@@ -169,12 +169,12 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
 	if tz != nil {
 		job.Schedule.TZ = *tz
 	}
-	if nextRunAt != nil {
-		ms := nextRunAt.UnixMilli()
+	if nextRunAt.Valid {
+		ms := nextRunAt.Time.UnixMilli()
 		job.State.NextRunAtMS = &ms
 	}
-	if lastRunAt != nil {
-		ms := lastRunAt.UnixMilli()
+	if lastRunAt.Valid {
+		ms := lastRunAt.Time.UnixMilli()
 		job.State.LastRunAtMS = &ms
 	}
 	if lastStatus != nil {
@@ -189,44 +189,7 @@ func scanCronRow(row cronRowScanner) (*store.CronJob, error) {
 
 // computeNextRun calculates the next run time for a schedule.
 func computeNextRun(schedule *store.CronSchedule, now time.Time, defaultTZ string) *time.Time {
-	switch schedule.Kind {
-	case "at":
-		if schedule.AtMS != nil {
-			t := time.UnixMilli(*schedule.AtMS)
-			if t.After(now) {
-				return &t
-			}
-		}
-		return nil
-	case "every":
-		if schedule.EveryMS != nil && *schedule.EveryMS > 0 {
-			t := now.Add(time.Duration(*schedule.EveryMS) * time.Millisecond)
-			return &t
-		}
-		return nil
-	case "cron":
-		if schedule.Expr == "" {
-			return nil
-		}
-		tz := schedule.TZ
-		if tz == "" {
-			tz = defaultTZ
-		}
-		evalTime := now
-		if tz != "" {
-			if loc, err := time.LoadLocation(tz); err == nil {
-				evalTime = now.In(loc)
-			}
-		}
-		nextTime, err := gronx.NextTickAfter(schedule.Expr, evalTime, false)
-		if err != nil {
-			return nil
-		}
-		utcNext := nextTime.UTC()
-		return &utcNext
-	default:
-		return nil
-	}
+	return store.ComputeNextRun(schedule, now, defaultTZ)
 }
 
 func (s *SQLiteCronStore) scanJob(ctx context.Context, id uuid.UUID) (*store.CronJob, error) {

internal/store/sqlitestore/cron_crud.go

@@ -4,12 +4,14 @@ package sqlitestore
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log/slog"
+	"strings"
 	"time"
 
-	"github.com/adhocore/gronx"
 	"github.com/google/uuid"
 
 	"github.com/nextlevelbuilder/goclaw/internal/store"
@@ -184,25 +186,38 @@ func (s *SQLiteCronStore) EnableJob(ctx context.Context, jobID string, enabled b
 		return fmt.Errorf("invalid job ID: %s", jobID)
 	}
 
-	q := "UPDATE cron_jobs SET enabled = ?, updated_at = ? WHERE id = ?"
-	args := []any{enabled, time.Now(), id}
+	s.writeMu.Lock()
+	defer s.writeMu.Unlock()
 
-	if !store.IsCrossTenant(ctx) {
-		tid := store.TenantIDFromContext(ctx)
-		if tid == uuid.Nil {
-			return fmt.Errorf("tenant_id required")
-		}
-		q += " AND tenant_id = ?"
-		args = append(args, tid)
-	}
-
-	res, err := s.db.ExecContext(ctx, q, args...)
+	tx, err := s.db.BeginTx(ctx, nil)
 	if err != nil {
 		return err
 	}
-	if n, _ := res.RowsAffected(); n == 0 {
-		return fmt.Errorf("job not found")
+	defer tx.Rollback()
+
+	current, err := s.lockCronJobForMutation(ctx, tx, id, false)
+	if err != nil {
+		return err
 	}
+
+	now := time.Now()
+	nextRun, err := store.NextRunForToggle(&current.Schedule, enabled, current.Enabled, current.NextRunAt, now, s.defaultTZ)
+	if err != nil {
+		return err
+	}
+
+	if err := execCronJobUpdateTx(ctx, tx, id, map[string]any{
+		"enabled":     enabled,
+		"next_run_at": nextRun,
+		"updated_at":  now,
+	}); err != nil {
+		return err
+	}
+
+	if err := tx.Commit(); err != nil {
+		return err
+	}
+
 	s.InvalidateCache()
 	return nil
 }
@@ -213,86 +228,31 @@ func (s *SQLiteCronStore) UpdateJob(ctx context.Context, jobID string, patch sto
 		return nil, fmt.Errorf("invalid job ID: %s", jobID)
 	}
 
+	s.writeMu.Lock()
+	defer s.writeMu.Unlock()
+
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	current, err := s.lockCronJobForMutation(ctx, tx, id, true)
+	if err != nil {
+		return nil, err
+	}
+
+	now := time.Now()
 	updates := make(map[string]any)
+	effectiveEnabled := current.Enabled
+	if patch.Enabled != nil {
+		effectiveEnabled = *patch.Enabled
+		updates["enabled"] = effectiveEnabled
+	}
+
 	if patch.Name != "" {
 		updates["name"] = patch.Name
 	}
-	if patch.Enabled != nil {
-		updates["enabled"] = *patch.Enabled
-	}
-
-	// Build tenant WHERE suffix for schedule reads.
-	tenantSuffix := ""
-	var tenantArg []any
-	if !store.IsCrossTenant(ctx) {
-		if tid := store.TenantIDFromContext(ctx); tid != uuid.Nil {
-			tenantSuffix = " AND tenant_id = ?"
-			tenantArg = append(tenantArg, tid)
-		}
-	}
-
-	if patch.Schedule != nil {
-		var curKind string
-		var curExpr, curTZ *string
-		var curIntervalMS *int64
-		var curRunAt *time.Time
-		if err := s.db.QueryRowContext(ctx,
-			"SELECT schedule_kind, cron_expression, timezone, interval_ms, run_at FROM cron_jobs WHERE id = ?"+tenantSuffix,
-			append([]any{id}, tenantArg...)...,
-		).Scan(&curKind, &curExpr, &curTZ, &curIntervalMS, &curRunAt); err != nil {
-			return nil, fmt.Errorf("failed to fetch current schedule for job %s: %w", jobID, err)
-		}
-
-		newKind := patch.Schedule.Kind
-		if newKind == "" {
-			newKind = curKind
-		}
-		updates["schedule_kind"] = newKind
-
-		switch newKind {
-		case "cron":
-			if patch.Schedule.Expr != "" {
-				updates["cron_expression"] = patch.Schedule.Expr
-			}
-			if patch.Schedule.TZ != "" {
-				if _, err := time.LoadLocation(patch.Schedule.TZ); err != nil {
-					return nil, fmt.Errorf("invalid timezone: %s", patch.Schedule.TZ)
-				}
-				updates["timezone"] = patch.Schedule.TZ
-			}
-			if curKind != "cron" {
-				updates["interval_ms"] = nil
-				updates["run_at"] = nil
-			}
-		case "every":
-			if patch.Schedule.EveryMS != nil {
-				updates["interval_ms"] = *patch.Schedule.EveryMS
-			}
-			if curKind != "every" {
-				updates["cron_expression"] = nil
-				updates["timezone"] = nil
-				updates["run_at"] = nil
-			}
-		case "at":
-			if patch.Schedule.AtMS != nil {
-				t := time.UnixMilli(*patch.Schedule.AtMS)
-				updates["run_at"] = t
-			}
-			if curKind != "at" {
-				updates["cron_expression"] = nil
-				updates["timezone"] = nil
-				updates["interval_ms"] = nil
-			}
-		}
-
-		merged := buildMergedSchedule(newKind, curKind, patch.Schedule, curExpr, curTZ, curIntervalMS, curRunAt)
-		if err := validateSchedule(&merged); err != nil {
-			return nil, err
-		}
-		next := computeNextRun(&merged, time.Now(), s.defaultTZ)
-		updates["next_run_at"] = next
-	}
-
 	if patch.DeleteAfterRun != nil {
 		updates["delete_after_run"] = *patch.DeleteAfterRun
 	}
@@ -304,54 +264,60 @@ func (s *SQLiteCronStore) UpdateJob(ctx context.Context, jobID string, patch sto
 		}
 	}
 
+	if patch.Schedule != nil {
+		merged := store.MergeCronSchedule(current.Schedule, patch.Schedule)
+		if err := store.ValidateCronSchedule(&merged); err != nil {
+			return nil, err
+		}
+
+		store.ApplyCronScheduleUpdates(updates, merged)
+
+		nextRun, err := store.NextRunForSchedule(&merged, effectiveEnabled, now, s.defaultTZ)
+		if err != nil {
+			return nil, err
+		}
+		updates["next_run_at"] = nextRun
+	} else if patch.Enabled != nil {
+		nextRun, err := store.NextRunForToggle(&current.Schedule, effectiveEnabled, current.Enabled, current.NextRunAt, now, s.defaultTZ)
+		if err != nil {
+			return nil, err
+		}
+		updates["next_run_at"] = nextRun
+	}
+
 	needsPayloadUpdate := patch.Message != "" || patch.Deliver != nil || patch.Channel != nil || patch.To != nil || patch.WakeHeartbeat != nil
 	if needsPayloadUpdate {
-		var payloadJSON []byte
-		if scanErr := s.db.QueryRowContext(ctx, "SELECT payload FROM cron_jobs WHERE id = ?"+tenantSuffix, append([]any{id}, tenantArg...)...).Scan(&payloadJSON); scanErr == nil {
-			var p store.CronPayload
-			if len(payloadJSON) > 0 {
-				if err := json.Unmarshal(payloadJSON, &p); err != nil {
-					return nil, fmt.Errorf("failed to parse existing payload for job %s: %w", jobID, err)
-				}
-			}
-			if patch.Message != "" {
-				p.Message = patch.Message
-			}
-			if patch.Deliver != nil {
-				p.Deliver = *patch.Deliver
-			}
-			if patch.Channel != nil {
-				p.Channel = *patch.Channel
-			}
-			if patch.To != nil {
-				p.To = *patch.To
-			}
-			if patch.WakeHeartbeat != nil {
-				p.WakeHeartbeat = *patch.WakeHeartbeat
-			}
-			merged, err := json.Marshal(p)
-			if err != nil {
-				return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err)
-			}
-			updates["payload"] = merged
+		payload := current.Payload
+		if patch.Message != "" {
+			payload.Message = patch.Message
 		}
+		if patch.Deliver != nil {
+			payload.Deliver = *patch.Deliver
+		}
+		if patch.Channel != nil {
+			payload.Channel = *patch.Channel
+		}
+		if patch.To != nil {
+			payload.To = *patch.To
+		}
+		if patch.WakeHeartbeat != nil {
+			payload.WakeHeartbeat = *patch.WakeHeartbeat
+		}
+
+		mergedPayload, err := json.Marshal(payload)
+		if err != nil {
+			return nil, fmt.Errorf("failed to marshal payload for job %s: %w", jobID, err)
+		}
+		updates["payload"] = mergedPayload
 	}
 
-	updates["updated_at"] = time.Now()
-
-	var execErr error
-	if store.IsCrossTenant(ctx) {
-		execErr = execMapUpdate(ctx, s.db, "cron_jobs", id, updates)
-	} else {
-		tid := store.TenantIDFromContext(ctx)
-		if tid == uuid.Nil {
-			execErr = fmt.Errorf("tenant_id required for update")
-		} else {
-			execErr = execMapUpdateWhereTenant(ctx, s.db, "cron_jobs", updates, id, tid)
-		}
+	updates["updated_at"] = now
+	if err := execCronJobUpdateTx(ctx, tx, id, updates); err != nil {
+		return nil, err
 	}
-	if execErr != nil {
-		return nil, execErr
+
+	if err := tx.Commit(); err != nil {
+		return nil, err
 	}
 
 	s.InvalidateCache()
@@ -359,55 +325,111 @@ func (s *SQLiteCronStore) UpdateJob(ctx context.Context, jobID string, patch sto
 	return job, nil
 }
 
-func buildMergedSchedule(newKind, curKind string, patch *store.CronSchedule, curExpr, curTZ *string, curIntervalMS *int64, curRunAt *time.Time) store.CronSchedule {
-	merged := store.CronSchedule{Kind: newKind}
-	switch newKind {
-	case "cron":
-		if patch.Expr != "" {
-			merged.Expr = patch.Expr
-		} else if curExpr != nil {
-			merged.Expr = *curExpr
+func (s *SQLiteCronStore) lockCronJobForMutation(ctx context.Context, tx *sql.Tx, id uuid.UUID, loadPayload bool) (*store.CronJobMutableState, error) {
+	q := `SELECT enabled, schedule_kind, cron_expression, run_at, timezone, interval_ms, next_run_at, payload
+		FROM cron_jobs WHERE id = ?`
+	args := []any{id}
+
+	if !store.IsCrossTenant(ctx) {
+		tid := store.TenantIDFromContext(ctx)
+		if tid == uuid.Nil {
+			return nil, fmt.Errorf("tenant_id required")
 		}
-		if patch.TZ != "" {
-			merged.TZ = patch.TZ
-		} else if curTZ != nil && newKind == curKind {
-			merged.TZ = *curTZ
+		q += " AND tenant_id = ?"
+		args = append(args, tid)
+	}
+
+	var (
+		state        store.CronJobMutableState
+		scheduleKind string
+		cronExpr     *string
+		runAt        nullSqliteTime
+		tz           *string
+		intervalMS   *int64
+		nextRunAt    nullSqliteTime
+		payloadJSON  []byte
+	)
+
+	if err := tx.QueryRowContext(ctx, q, args...).Scan(
+		&state.Enabled,
+		&scheduleKind,
+		&cronExpr,
+		&runAt,
+		&tz,
+		&intervalMS,
+		&nextRunAt,
+		&payloadJSON,
+	); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, store.ErrCronJobNotFound
 		}
-	case "every":
-		if patch.EveryMS != nil {
-			merged.EveryMS = patch.EveryMS
-		} else if curIntervalMS != nil {
-			merged.EveryMS = curIntervalMS
-		}
-	case "at":
-		if patch.AtMS != nil {
-			merged.AtMS = patch.AtMS
-		} else if curRunAt != nil {
-			ms := curRunAt.UnixMilli()
-			merged.AtMS = &ms
+		return nil, err
+	}
+
+	state.Schedule = store.CronSchedule{Kind: scheduleKind}
+	if cronExpr != nil {
+		state.Schedule.Expr = *cronExpr
+	}
+	if runAt.Valid {
+		ms := runAt.Time.UnixMilli()
+		state.Schedule.AtMS = &ms
+	}
+	if tz != nil {
+		state.Schedule.TZ = *tz
+	}
+	if intervalMS != nil {
+		state.Schedule.EveryMS = intervalMS
+	}
+	if nextRunAt.Valid {
+		next := nextRunAt.Time
+		state.NextRunAt = &next
+	}
+
+	if loadPayload && len(payloadJSON) > 0 {
+		if err := json.Unmarshal(payloadJSON, &state.Payload); err != nil {
+			return nil, fmt.Errorf("failed to parse existing payload for job %s: %w", id, err)
 		}
 	}
-	return merged
+
+	return &state, nil
 }
 
-func validateSchedule(s *store.CronSchedule) error {
-	switch s.Kind {
-	case "cron":
-		if s.Expr == "" {
-			return fmt.Errorf("cron schedule requires expr")
+
+func execCronJobUpdateTx(ctx context.Context, tx *sql.Tx, id uuid.UUID, updates map[string]any) error {
+	if len(updates) == 0 {
+		return nil
+	}
+
+	var (
+		setClauses []string
+		args       []any
+	)
+	for _, col := range store.SortedUpdateColumns(updates) {
+		if !validColumnName.MatchString(col) {
+			return fmt.Errorf("invalid column name: %q", col)
 		}
-		gx := gronx.New()
-		if !gx.IsValid(s.Expr) {
-			return fmt.Errorf("invalid cron expression: %s", s.Expr)
-		}
-	case "every":
-		if s.EveryMS == nil || *s.EveryMS <= 0 {
-			return fmt.Errorf("every schedule requires positive everyMs")
-		}
-	case "at":
-		if s.AtMS == nil {
-			return fmt.Errorf("at schedule requires atMs")
+		setClauses = append(setClauses, col+" = ?")
+		args = append(args, sqliteVal(updates[col]))
+	}
+
+	args = append(args, id)
+	q := "UPDATE cron_jobs SET " + strings.Join(setClauses, ", ") + " WHERE id = ?"
+	if !store.IsCrossTenant(ctx) {
+		tid := store.TenantIDFromContext(ctx)
+		if tid == uuid.Nil {
+			return fmt.Errorf("tenant_id required")
 		}
+		args = append(args, tid)
+		q += " AND tenant_id = ?"
+	}
+
+	res, err := tx.ExecContext(ctx, q, args...)
+	if err != nil {
+		return err
+	}
+	if n, _ := res.RowsAffected(); n == 0 {
+		return store.ErrCronJobNotFound
 	}
 	return nil
 }
+

internal/store/sqlitestore/cron_crud_test.go

@@ -0,0 +1,337 @@
+//go:build sqlite || sqliteonly
+
+package sqlitestore
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+
+	"github.com/nextlevelbuilder/goclaw/internal/store"
+)
+
+func TestSQLiteCronStore_ReenableRestoresNextRun(t *testing.T) {
+	cronStore, ctx, _ := newTestSQLiteCronStore(t)
+	everyMS := int64(time.Minute / time.Millisecond)
+
+	job, err := cronStore.AddJob(ctx, "job-reenable", store.CronSchedule{
+		Kind:    "every",
+		EveryMS: &everyMS,
+	}, "hello", false, "", "", "", "user-1")
+	if err != nil {
+		t.Fatalf("AddJob error: %v", err)
+	}
+	if job == nil {
+		job = mustOnlyJob(t, cronStore, ctx)
+	}
+
+	if err := cronStore.EnableJob(ctx, job.ID, false); err != nil {
+		t.Fatalf("disable error: %v", err)
+	}
+	disabled, ok := cronStore.GetJob(ctx, job.ID)
+	if !ok {
+		t.Fatal("disabled job not found")
+	}
+	if disabled.Enabled {
+		t.Fatal("expected job to be disabled")
+	}
+	if disabled.State.NextRunAtMS != nil {
+		t.Fatalf("expected disabled job to clear next_run_at, got %v", *disabled.State.NextRunAtMS)
+	}
+
+	if err := cronStore.EnableJob(ctx, job.ID, true); err != nil {
+		t.Fatalf("re-enable error: %v", err)
+	}
+	reenabled, ok := cronStore.GetJob(ctx, job.ID)
+	if !ok {
+		t.Fatal("re-enabled job not found")
+	}
+	if !reenabled.Enabled {
+		t.Fatal("expected job to be enabled")
+	}
+	if reenabled.State.NextRunAtMS == nil {
+		t.Fatal("expected re-enabled job to have next_run_at")
+	}
+
+	due := cronStore.GetDueJobs(time.UnixMilli(*reenabled.State.NextRunAtMS))
+	if len(due) != 1 || due[0].ID != job.ID {
+		t.Fatalf("expected job %s to be due after re-enable, got %#v", job.ID, due)
+	}
+}
+
+func TestSQLiteCronStore_EnableAlreadyEnabledPreservesNextRun(t *testing.T) {
+	cronStore, ctx, _ := newTestSQLiteCronStore(t)
+	everyMS := int64(time.Hour / time.Millisecond)
+
+	job, err := cronStore.AddJob(ctx, "job-idempotent", store.CronSchedule{
+		Kind:    "every",
+		EveryMS: &everyMS,
+	}, "hello", false, "", "", "", "user-1")
+	if err != nil {
+		t.Fatalf("AddJob error: %v", err)
+	}
+	if job == nil {
+		job = mustOnlyJob(t, cronStore, ctx)
+	}
+	if job.State.NextRunAtMS == nil {
+		t.Fatal("expected new job to have next_run_at")
+	}
+
+	originalNextRun := *job.State.NextRunAtMS
+	time.Sleep(20 * time.Millisecond)
+
+	if err := cronStore.EnableJob(ctx, job.ID, true); err != nil {
+		t.Fatalf("enable error: %v", err)
+	}
+
+	current, ok := cronStore.GetJob(ctx, job.ID)
+	if !ok {
+		t.Fatal("job not found after enable")
+	}
+	if current.State.NextRunAtMS == nil {
+		t.Fatal("expected enabled job to keep next_run_at")
+	}
+	if got := *current.State.NextRunAtMS; got != originalNextRun {
+		t.Fatalf("got next_run_at %d, want preserved %d", got, originalNextRun)
+	}
+}
+
+func TestSQLiteCronStore_EnableExpiredAtReturnsError(t *testing.T) {
+	cronStore, ctx, _ := newTestSQLiteCronStore(t)
+	future := time.Now().Add(time.Hour).UnixMilli()
+
+	job, err := cronStore.AddJob(ctx, "job-expired-at", store.CronSchedule{
+		Kind: "at",
+		AtMS: &future,
+	}, "hello", false, "", "", "", "user-1")
+	if err != nil {
+		t.Fatalf("AddJob error: %v", err)
+	}
+	if job == nil {
+		job = mustOnlyJob(t, cronStore, ctx)
+	}
+
+	if err := cronStore.EnableJob(ctx, job.ID, false); err != nil {
+		t.Fatalf("disable error: %v", err)
+	}
+
+	past := time.Now().Add(-time.Hour).UnixMilli()
+	if _, err := cronStore.UpdateJob(ctx, job.ID, store.CronJobPatch{
+		Schedule: &store.CronSchedule{Kind: "at", AtMS: &past},
+	}); err != nil {
+		t.Fatalf("UpdateJob error: %v", err)
+	}
+
+	err = cronStore.EnableJob(ctx, job.ID, true)
+	if !errors.Is(err, store.ErrCronJobNoFutureRun) {
+		t.Fatalf("got %v, want ErrCronJobNoFutureRun", err)
+	}
+
+	current, ok := cronStore.GetJob(ctx, job.ID)
+	if !ok {
+		t.Fatal("job not found after failed enable")
+	}
+	if current.Enabled {
+		t.Fatal("expected failed enable to leave job disabled")
+	}
+	if current.State.NextRunAtMS != nil {
+		t.Fatalf("expected failed enable to leave next_run_at nil, got %v", *current.State.NextRunAtMS)
+	}
+}
+
+func TestSQLiteCronStore_SchedulerSkipsDisabledCachedJob(t *testing.T) {
+	cronStore, ctx, db := newTestSQLiteCronStore(t)
+	everyMS := int64(time.Minute / time.Millisecond)
+
+	job, err := cronStore.AddJob(ctx, "job-stale-cache", store.CronSchedule{
+		Kind:    "every",
+		EveryMS: &everyMS,
+	}, "hello", false, "", "", "", "user-1")
+	if err != nil {
+		t.Fatalf("AddJob error: %v", err)
+	}
+	if job == nil {
+		job = mustOnlyJob(t, cronStore, ctx)
+	}
+
+	jobUUID := uuid.MustParse(job.ID)
+	pastRun := time.Now().Add(-time.Minute)
+	if _, err := db.ExecContext(ctx,
+		"UPDATE cron_jobs SET next_run_at = ?, updated_at = ? WHERE id = ? AND tenant_id = ?",
+		pastRun, time.Now(), jobUUID, store.MasterTenantID,
+	); err != nil {
+		t.Fatalf("mark due error: %v", err)
+	}
+
+	cached, ok := cronStore.GetJob(ctx, job.ID)
+	if !ok {
+		t.Fatal("cached job not found")
+	}
+	pastMS := pastRun.UnixMilli()
+	cached.Enabled = true
+	cached.State.NextRunAtMS = &pastMS
+	cronStore.jobCache = []store.CronJob{*cached}
+	cronStore.cacheLoaded = true
+	cronStore.cacheTime = time.Now()
+	cronStore.cacheTTL = time.Hour
+	cronStore.baseCtx = context.Background()
+
+	var runs atomic.Int32
+	cronStore.onJob = func(job *store.CronJob) (*store.CronJobResult, error) {
+		runs.Add(1)
+		return &store.CronJobResult{Content: "ok"}, nil
+	}
+
+	if _, err := db.ExecContext(ctx,
+		"UPDATE cron_jobs SET enabled = 0, next_run_at = NULL, updated_at = ? WHERE id = ? AND tenant_id = ?",
+		time.Now(), jobUUID, store.MasterTenantID,
+	); err != nil {
+		t.Fatalf("disable error: %v", err)
+	}
+
+	cronStore.checkAndRunDueJobs()
+
+	if got := runs.Load(); got != 0 {
+		t.Fatalf("expected claimed disabled job to be skipped, got %d executions", got)
+	}
+}
+
+func TestSQLiteCronStore_ExecuteOneJob_DoesNotRestoreNextRunAfterDisable(t *testing.T) {
+	cronStore, ctx, db := newTestSQLiteCronStore(t)
+	everyMS := int64(time.Minute / time.Millisecond)
+
+	job, err := cronStore.AddJob(ctx, "job-midrun-disable", store.CronSchedule{
+		Kind:    "every",
+		EveryMS: &everyMS,
+	}, "hello", false, "", "", "", "user-1")
+	if err != nil {
+		t.Fatalf("AddJob error: %v", err)
+	}
+	if job == nil {
+		job = mustOnlyJob(t, cronStore, ctx)
+	}
+
+	jobUUID := uuid.MustParse(job.ID)
+	if _, err := db.ExecContext(ctx,
+		"UPDATE cron_jobs SET next_run_at = NULL, updated_at = ? WHERE id = ? AND tenant_id = ?",
+		time.Now(), jobUUID, store.MasterTenantID,
+	); err != nil {
+		t.Fatalf("claim setup error: %v", err)
+	}
+
+	cronStore.executeOneJob(*job, func(job *store.CronJob) (*store.CronJobResult, error) {
+		if err := cronStore.EnableJob(ctx, job.ID, false); err != nil {
+			t.Fatalf("disable during run error: %v", err)
+		}
+		return &store.CronJobResult{Content: "ok"}, nil
+	})
+
+	current := mustRawJob(t, db, jobUUID)
+	if current.enabled {
+		t.Fatal("expected job to remain disabled after run")
+	}
+	if current.nextRunAt != nil {
+		t.Fatalf("expected disabled job to keep next_run_at nil, got %v", current.nextRunAt)
+	}
+}
+
+func TestSQLiteCronStore_EnableJob_IgnoresMalformedPayload(t *testing.T) {
+	cronStore, ctx, db := newTestSQLiteCronStore(t)
+	everyMS := int64(time.Minute / time.Millisecond)
+
+	job, err := cronStore.AddJob(ctx, "job-bad-payload", store.CronSchedule{
+		Kind:    "every",
+		EveryMS: &everyMS,
+	}, "hello", false, "", "", "", "user-1")
+	if err != nil {
+		t.Fatalf("AddJob error: %v", err)
+	}
+	if job == nil {
+		job = mustOnlyJob(t, cronStore, ctx)
+	}
+
+	jobUUID := uuid.MustParse(job.ID)
+	if _, err := db.ExecContext(ctx,
+		"UPDATE cron_jobs SET payload = ? WHERE id = ? AND tenant_id = ?",
+		"{", jobUUID, store.MasterTenantID,
+	); err != nil {
+		t.Fatalf("corrupt payload error: %v", err)
+	}
+
+	if err := cronStore.EnableJob(ctx, job.ID, false); err != nil {
+		t.Fatalf("toggle with malformed payload error: %v", err)
+	}
+
+	current := mustRawJob(t, db, jobUUID)
+	if current.enabled {
+		t.Fatal("expected malformed-payload job to be disabled")
+	}
+	if current.nextRunAt != nil {
+		t.Fatalf("expected malformed-payload job to clear next_run_at, got %v", current.nextRunAt)
+	}
+}
+
+func newTestSQLiteCronStore(t *testing.T) (*SQLiteCronStore, context.Context, *sql.DB) {
+	t.Helper()
+
+	db, err := OpenDB(filepath.Join(t.TempDir(), "cron.db"))
+	if err != nil {
+		t.Fatalf("OpenDB error: %v", err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+
+	if err := EnsureSchema(db); err != nil {
+		t.Fatalf("EnsureSchema error: %v", err)
+	}
+
+	ctx := store.WithCrossTenant(store.WithTenantID(context.Background(), store.MasterTenantID))
+	cronStore := NewSQLiteCronStore(db)
+	cronStore.baseCtx = context.Background()
+	cronStore.cacheTTL = time.Hour
+	return cronStore, ctx, db
+}
+
+func mustOnlyJob(t *testing.T, cronStore *SQLiteCronStore, ctx context.Context) *store.CronJob {
+	t.Helper()
+
+	jobs := cronStore.ListJobs(ctx, true, "", "")
+	if len(jobs) != 1 {
+		t.Fatalf("expected exactly 1 job, got %#v", jobs)
+	}
+
+	job := jobs[0]
+	return &job
+}
+
+type rawCronJobState struct {
+	enabled   bool
+	nextRunAt *time.Time
+}
+
+func mustRawJob(t *testing.T, db *sql.DB, id uuid.UUID) rawCronJobState {
+	t.Helper()
+
+	var (
+		enabled   bool
+		nextRunAt nullSqliteTime
+	)
+	if err := db.QueryRow(
+		"SELECT enabled, next_run_at FROM cron_jobs WHERE id = ?",
+		id,
+	).Scan(&enabled, &nextRunAt); err != nil {
+		t.Fatalf("raw cron job query error: %v", err)
+	}
+
+	state := rawCronJobState{enabled: enabled}
+	if nextRunAt.Valid {
+		next := nextRunAt.Time
+		state.nextRunAt = &next
+	}
+	return state
+}

internal/store/sqlitestore/cron_scheduler.go

@@ -3,6 +3,8 @@
 package sqlitestore
 
 import (
+	"database/sql"
+	"errors"
 	"log/slog"
 	"sync"
 	"time"
@@ -152,16 +154,20 @@ func (s *SQLiteCronStore) checkAndRunDueJobs() {
 		return
 	}
 
-	// Clear next_run for all due jobs first to prevent duplicate fires.
+	now := time.Now()
+	var claimedJobs []store.CronJob
 	for _, job := range dueJobs {
-		if id, parseErr := uuid.Parse(job.ID); parseErr == nil {
-			s.db.ExecContext(s.baseCtx, "UPDATE cron_jobs SET next_run_at = NULL WHERE id = ?", id)
+		if id, parseErr := uuid.Parse(job.ID); parseErr == nil && s.claimDueJob(id, now) {
+			claimedJobs = append(claimedJobs, job)
 		}
 	}
+	if len(claimedJobs) == 0 {
+		return
+	}
 
 	// Execute jobs in parallel — scheduler enforces per-session serialization.
 	var wg sync.WaitGroup
-	for _, job := range dueJobs {
+	for _, job := range claimedJobs {
 		wg.Add(1)
 		go func(job store.CronJob) {
 			defer wg.Done()
@@ -177,6 +183,15 @@ func (s *SQLiteCronStore) checkAndRunDueJobs() {
 
 // executeOneJob runs a single cron job with retry, logs the result, and updates next_run_at.
 func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *store.CronJob) (*store.CronJobResult, error)) {
+	if id, parseErr := uuid.Parse(job.ID); parseErr == nil {
+		freshJob, ok := s.loadClaimedJob(id)
+		if !ok {
+			slog.Info("cron job skipped after claim state changed", "id", job.ID)
+			return
+		}
+		job = *freshJob
+	}
+
 	startTime := time.Now()
 
 	var lastResult *store.CronJobResult
@@ -238,9 +253,16 @@ func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *sto
 	} else if id, parseErr := uuid.Parse(job.ID); parseErr == nil {
 		schedule := job.Schedule
 		next := computeNextRun(&schedule, now, s.defaultTZ)
+		var nextRunValue any
+		if next != nil {
+			nextRunValue = *next
+		}
 		s.db.ExecContext(s.baseCtx,
-			"UPDATE cron_jobs SET last_run_at = ?, last_status = ?, last_error = ?, next_run_at = ?, updated_at = ? WHERE id = ?",
-			now, status, lastError, next, now, id,
+			`UPDATE cron_jobs SET
+			 last_run_at = ?, last_status = ?, last_error = ?, updated_at = ?,
+			 next_run_at = CASE WHEN enabled = 1 AND next_run_at IS NULL THEN ? ELSE next_run_at END
+			 WHERE id = ?`,
+			now, status, lastError, now, nextRunValue, id,
 		)
 	}
 
@@ -251,3 +273,42 @@ func (s *SQLiteCronStore) executeOneJob(job store.CronJob, handler func(job *sto
 	}
 	s.emitEvent(evt)
 }
+
+func (s *SQLiteCronStore) claimDueJob(id uuid.UUID, now time.Time) bool {
+	res, err := s.db.ExecContext(
+		s.baseCtx,
+		`UPDATE cron_jobs
+		 SET next_run_at = NULL
+		 WHERE id = ? AND enabled = 1 AND next_run_at IS NOT NULL AND next_run_at <= ?`,
+		id,
+		now,
+	)
+	if err != nil {
+		slog.Warn("cron: failed to claim due job", "id", id, "error", err)
+		return false
+	}
+
+	n, _ := res.RowsAffected()
+	return n == 1
+}
+
+func (s *SQLiteCronStore) loadClaimedJob(id uuid.UUID) (*store.CronJob, bool) {
+	row := s.db.QueryRowContext(
+		s.baseCtx,
+		`SELECT id, tenant_id, agent_id, user_id, name, enabled, schedule_kind, cron_expression, run_at, timezone,
+		 interval_ms, payload, delete_after_run, next_run_at, last_run_at, last_status, last_error,
+		 created_at, updated_at
+		 FROM cron_jobs
+		 WHERE id = ? AND enabled = 1 AND next_run_at IS NULL`,
+		id,
+	)
+	job, err := scanCronRow(row)
+	if errors.Is(err, sql.ErrNoRows) {
+		return nil, false
+	}
+	if err != nil {
+		slog.Warn("cron: failed to reload claimed job", "id", id, "error", err)
+		return nil, false
+	}
+	return job, true
+}

internal/store/sqlitestore/helpers.go

@@ -119,9 +119,14 @@ func sqliteVal(v any) any {
 	if v == nil {
 		return nil
 	}
-	switch v.(type) {
+	switch typed := v.(type) {
 	case string, int, int64, float64, bool, time.Time, []byte, json.RawMessage:
 		return v
+	case *time.Time:
+		if typed == nil {
+			return nil
+		}
+		return *typed
 	}
 	// For maps, slices, etc. — marshal to JSON string.
 	b, err := json.Marshal(v)