feat(cron): add default stagger controls for scheduled jobs

src/cron/service/jobs.ts

@@ -1,6 +1,4 @@
 import crypto from "node:crypto";
-import { parseAbsoluteTimeMs } from "../parse.js";
-import { computeNextRunAtMs } from "../schedule.js";
 import type {
   CronDelivery,
   CronDeliveryPatch,
@@ -10,6 +8,14 @@ import type {
   CronPayload,
   CronPayloadPatch,
 } from "../types.js";
+import type { CronServiceState } from "./state.js";
+import { parseAbsoluteTimeMs } from "../parse.js";
+import { computeNextRunAtMs } from "../schedule.js";
+import {
+  normalizeCronStaggerMs,
+  resolveCronStaggerMs,
+  resolveDefaultCronStaggerMs,
+} from "../stagger.js";
 import { normalizeHttpWebhookUrl } from "../webhook-url.js";
 import {
   normalizeOptionalAgentId,
@@ -18,10 +24,45 @@ import {
   normalizePayloadToSystemText,
   normalizeRequiredName,
 } from "./normalize.js";
-import type { CronServiceState } from "./state.js";
 
 const STUCK_RUN_MS = 2 * 60 * 60 * 1000;
 
+function resolveStableCronOffsetMs(jobId: string, staggerMs: number) {
+  if (staggerMs <= 1) {
+    return 0;
+  }
+  const digest = crypto.createHash("sha256").update(jobId).digest();
+  return digest.readUInt32BE(0) % staggerMs;
+}
+
+function computeStaggeredCronNextRunAtMs(job: CronJob, nowMs: number) {
+  if (job.schedule.kind !== "cron") {
+    return computeNextRunAtMs(job.schedule, nowMs);
+  }
+
+  const staggerMs = resolveCronStaggerMs(job.schedule);
+  const offsetMs = resolveStableCronOffsetMs(job.id, staggerMs);
+  if (offsetMs <= 0) {
+    return computeNextRunAtMs(job.schedule, nowMs);
+  }
+
+  // Shift the schedule cursor backwards by the per-job offset so we can still
+  // target the current schedule window if its staggered slot has not passed yet.
+  let cursorMs = Math.max(0, nowMs - offsetMs);
+  for (let attempt = 0; attempt < 4; attempt += 1) {
+    const baseNext = computeNextRunAtMs(job.schedule, cursorMs);
+    if (baseNext === undefined) {
+      return undefined;
+    }
+    const shifted = baseNext + offsetMs;
+    if (shifted > nowMs) {
+      return shifted;
+    }
+    cursorMs = Math.max(cursorMs + 1, baseNext + 1_000);
+  }
+  return undefined;
+}
+
 function resolveEveryAnchorMs(params: {
   schedule: { everyMs: number; anchorMs?: number };
   fallbackAnchorMs: number;
@@ -97,18 +138,7 @@ export function computeJobNextRunAtMs(job: CronJob, nowMs: number): number | und
             : null;
     return atMs !== null ? atMs : undefined;
   }
-  const next = computeNextRunAtMs(job.schedule, nowMs);
-  // Guard against the scheduler returning a time within the same second as
-  // nowMs.  When a cron job completes within the same wall-clock second it
-  // was scheduled for, some croner versions/timezone combinations may return
-  // the current second (or computeNextRunAtMs may return undefined, which
-  // triggers recomputation).  Advancing to the next second and retrying
-  // ensures we always land on the *next* occurrence.  (See #17821)
-  if (next === undefined && job.schedule.kind === "cron") {
-    const nextSecondMs = (Math.floor(nowMs / 1000) + 1) * 1000;
-    return computeNextRunAtMs(job.schedule, nextSecondMs);
-  }
-  return next;
+  return computeStaggeredCronNextRunAtMs(job, nowMs);
 }
 
 /** Maximum consecutive schedule errors before auto-disabling a job. */
@@ -288,7 +318,18 @@ export function createJob(state: CronServiceState, input: CronJobCreate): CronJo
             fallbackAnchorMs: now,
           }),
         }
-      : input.schedule;
+      : input.schedule.kind === "cron"
+        ? (() => {
+            const explicitStaggerMs = normalizeCronStaggerMs(input.schedule.staggerMs);
+            if (explicitStaggerMs !== undefined) {
+              return { ...input.schedule, staggerMs: explicitStaggerMs };
+            }
+            const defaultStaggerMs = resolveDefaultCronStaggerMs(input.schedule.expr);
+            return defaultStaggerMs !== undefined
+              ? { ...input.schedule, staggerMs: defaultStaggerMs }
+              : input.schedule;
+          })()
+        : input.schedule;
   const deleteAfterRun =
     typeof input.deleteAfterRun === "boolean"
       ? input.deleteAfterRun
@@ -335,7 +376,22 @@ export function applyJobPatch(job: CronJob, patch: CronJobPatch) {
     job.deleteAfterRun = patch.deleteAfterRun;
   }
   if (patch.schedule) {
-    job.schedule = patch.schedule;
+    if (patch.schedule.kind === "cron") {
+      const explicitStaggerMs = normalizeCronStaggerMs(patch.schedule.staggerMs);
+      if (explicitStaggerMs !== undefined) {
+        job.schedule = { ...patch.schedule, staggerMs: explicitStaggerMs };
+      } else if (job.schedule.kind === "cron") {
+        job.schedule = { ...patch.schedule, staggerMs: job.schedule.staggerMs };
+      } else {
+        const defaultStaggerMs = resolveDefaultCronStaggerMs(patch.schedule.expr);
+        job.schedule =
+          defaultStaggerMs !== undefined
+            ? { ...patch.schedule, staggerMs: defaultStaggerMs }
+            : patch.schedule;
+      }
+    } else {
+      job.schedule = patch.schedule;
+    }
   }
   if (patch.sessionTarget) {
     job.sessionTarget = patch.sessionTarget;

src/cron/service/store.ts

@@ -1,4 +1,6 @@
 import fs from "node:fs";
+import type { CronJob } from "../types.js";
+import type { CronServiceState } from "./state.js";
 import {
   buildDeliveryFromLegacyPayload,
   hasLegacyDeliveryHints,
@@ -6,11 +8,10 @@ import {
 } from "../legacy-delivery.js";
 import { parseAbsoluteTimeMs } from "../parse.js";
 import { migrateLegacyCronPayload } from "../payload-migration.js";
+import { normalizeCronStaggerMs, resolveDefaultCronStaggerMs } from "../stagger.js";
 import { loadCronStore, saveCronStore } from "../store.js";
-import type { CronJob } from "../types.js";
 import { recomputeNextRuns } from "./jobs.js";
 import { inferLegacyName, normalizeOptionalText } from "./normalize.js";
-import type { CronServiceState } from "./state.js";
 
 function buildDeliveryPatchFromLegacyPayload(payload: Record<string, unknown>) {
   const deliver = payload.deliver;
@@ -380,6 +381,26 @@ export async function ensureLoaded(
           mutated = true;
         }
       }
+
+      const exprRaw = typeof sched.expr === "string" ? sched.expr.trim() : "";
+      if (typeof sched.expr === "string" && sched.expr !== exprRaw) {
+        sched.expr = exprRaw;
+        mutated = true;
+      }
+      if ((kind === "cron" || sched.kind === "cron") && exprRaw) {
+        const explicitStaggerMs = normalizeCronStaggerMs(sched.staggerMs);
+        const defaultStaggerMs = resolveDefaultCronStaggerMs(exprRaw);
+        const targetStaggerMs = explicitStaggerMs ?? defaultStaggerMs;
+        if (targetStaggerMs === undefined) {
+          if ("staggerMs" in sched) {
+            delete sched.staggerMs;
+            mutated = true;
+          }
+        } else if (sched.staggerMs !== targetStaggerMs) {
+          sched.staggerMs = targetStaggerMs;
+          mutated = true;
+        }
+      }
     }
 
     const delivery = raw.delivery;