fix(schedules): Harden runner recovery and manual runs

Author: MathurAditya724

apps/server/src/client/lib/api.ts

@@ -47,6 +47,7 @@ export type ScheduleInput = {
 export type Schedule = ScheduleInput & {
   id: string
   revision: number
+  armedRevision: number | null
   nextDueAt: number | null
   createdAt?: number
   updatedAt?: number

apps/server/src/client/pages/schedules.tsx

@@ -54,9 +54,9 @@ function cadenceLabel(schedule: Pick<ScheduleInput, "cadence" | "time" | "timezo
   return `Monthly on day ${schedule.dayOfMonth ?? 1} at ${schedule.time}`
 }
 
-function formatAt(value: number | null | undefined) {
+function formatAt(value: number | null | undefined, timeZone?: string) {
   return value
-    ? new Date(value).toLocaleString(undefined, { dateStyle: "medium", timeStyle: "short" })
+    ? new Date(value).toLocaleString(undefined, { dateStyle: "medium", timeStyle: "short", timeZone })
     : "Not scheduled"
 }
 
@@ -269,7 +269,15 @@ function ScheduleEditor({
             </CardHeader>
             <CardContent className="space-y-1 text-xs text-muted-foreground">
               {preview.length ? (
-                preview.map((time) => <p key={time}>{new Date(time).toLocaleString()}</p>)
+                preview.map((time) => (
+                  <p key={time}>
+                    {new Date(time).toLocaleString(undefined, {
+                      dateStyle: "medium",
+                      timeStyle: "short",
+                      timeZone: input.timezone,
+                    })}
+                  </p>
+                ))
               ) : (
                 <p>Enter a valid timezone and recurrence.</p>
               )}
@@ -380,6 +388,7 @@ export default function SchedulesPage() {
   const [editor, setEditor] = useState<Schedule | null | undefined>(undefined)
   const [historyId, setHistoryId] = useState<string | null>(null)
   const [confirm, setConfirm] = useState<{ kind: "run" | "archive"; schedule: Schedule } | null>(null)
+  const actionError = [run, setEnabled, archive].find((mutation) => mutation.isError)?.error
 
   return (
     <div className="space-y-6">
@@ -418,6 +427,13 @@ export default function SchedulesPage() {
         </Card>
       ) : (
         <div className="grid gap-3">
+          {actionError && (
+            <Card>
+              <CardContent role="alert" className="py-3 text-xs text-destructive">
+                {actionError instanceof Error ? actionError.message : "Could not update schedule"}
+              </CardContent>
+            </Card>
+          )}
           {schedules.data.data.map((schedule) => (
             <Card key={schedule.id}>
               <CardHeader className="pb-3">
@@ -434,9 +450,12 @@ export default function SchedulesPage() {
               <CardContent className="flex flex-wrap items-center gap-x-5 gap-y-2 text-xs text-muted-foreground">
                 <span className="inline-flex items-center gap-1">
                   <Clock className="size-3.5" />
-                  Next: {formatAt(schedule.nextDueAt)}
+                  Next: {formatAt(schedule.nextDueAt, schedule.timezone)}
                 </span>
                 <span>Last: {schedule.lastRun ? schedule.lastRun.status : "Never"}</span>
+                {schedule.enabled && schedule.armedRevision !== schedule.revision && (
+                  <span className="text-amber-700 dark:text-amber-400">Scheduler recovery pending</span>
+                )}
                 {schedule.lastRun?.artifact_url && (
                   <a
                     className="underline underline-offset-2"
@@ -477,7 +496,7 @@ export default function SchedulesPage() {
                     variant="outline"
                     size="xs"
                     onClick={() => setConfirm({ kind: "run", schedule })}
-                    disabled={!schedule.enabled || !!schedule.activeRun || run.isPending}
+                    disabled={!!schedule.activeRun || run.isPending}
                   >
                     <Play data-icon="inline-start" />
                     Run now
@@ -522,6 +541,7 @@ export default function SchedulesPage() {
           <AlertDialogFooter>
             <AlertDialogCancel>Cancel</AlertDialogCancel>
             <AlertDialogAction
+              disabled={confirm?.kind === "run" ? run.isPending : archive.isPending}
               onClick={(event) => {
                 event.preventDefault()
                 if (!confirm) return

apps/server/src/cloudflare.ts

@@ -102,19 +102,37 @@ const handlers = {
         } catch (_err) {
           console.warn("maintenance_runs.record.failed", { failure_class: "contained" })
         }
-        // Schedule configuration is authoritative in D1. Re-arm only schedules
-        // whose runner never confirmed the current revision; avoid disturbing
-        // healthy runner monitor alarms.
+        // Schedule configuration is authoritative in D1. Re-arm only runners
+        // that did not confirm their current revision. Active runs are included
+        // even when a recurrence was paused or archived: their monitor alarm is
+        // for settlement only and must not strand a leased capacity slot.
         try {
           const unarmed = await env.DB.prepare(
-            "SELECT id, revision, next_due_at FROM scheduled_jobs WHERE enabled = 1 AND archived_at IS NULL AND next_due_at IS NOT NULL AND (armed_revision IS NULL OR armed_revision != revision) LIMIT 100",
-          ).all<{ id: string; revision: number; next_due_at: number }>()
+            `SELECT job.id, job.revision, job.next_due_at,
+              EXISTS (
+                SELECT 1 FROM scheduled_job_runs run
+                WHERE run.schedule_id = job.id
+                  AND run.status IN ('preparing', 'admitting', 'admitted', 'unknown_admission')
+              ) AS has_active_run
+            FROM scheduled_jobs job
+            WHERE (job.armed_revision IS NULL OR job.armed_revision != job.revision)
+              AND (
+                (job.enabled = 1 AND job.archived_at IS NULL AND job.next_due_at IS NOT NULL)
+                OR EXISTS (
+                  SELECT 1 FROM scheduled_job_runs run
+                  WHERE run.schedule_id = job.id
+                    AND run.status IN ('preparing', 'admitting', 'admitted', 'unknown_admission')
+                )
+              )
+            LIMIT 100`,
+          ).all<{ id: string; revision: number; next_due_at: number | null; has_active_run: number }>()
           await Promise.all(
             (unarmed.results ?? []).map((job) =>
               env.ScheduleRunner.get(env.ScheduleRunner.idFromName(job.id)).arm({
                 scheduleId: job.id,
                 revision: job.revision,
                 nextDueAt: job.next_due_at,
+                wakeAt: job.has_active_run ? controller.scheduledTime : undefined,
               } satisfies ScheduleArm),
             ),
           )

apps/server/src/lib/schedules/__tests__/runner-state.test.ts

@@ -1,4 +1,5 @@
-import { describe, expect, it } from "vitest"
+import { describe, expect, it, vi } from "vitest"
+import { ScheduleRunner } from "../runner"
 import { nextRunnerWakeAt } from "../runner-state"
 
 describe("nextRunnerWakeAt", () => {
@@ -10,3 +11,23 @@ describe("nextRunnerWakeAt", () => {
     expect(nextRunnerWakeAt({ scheduleDueAt: 4_000, hasActiveRun: true, now: 3_000 })).toBe(4_000)
   })
 })
+
+describe("ScheduleRunner arm", () => {
+  it("keeps an alarm for active work even when its recurrence has no next due time", async () => {
+    const storage = {
+      put: vi.fn(),
+      setAlarm: vi.fn(),
+      deleteAlarm: vi.fn(),
+    }
+    const run = vi.fn().mockResolvedValue({ meta: { changes: 1 } })
+    const durableObject = new ScheduleRunner(
+      { storage } as never,
+      { DB: { prepare: () => ({ bind: () => ({ run }) }) } } as never,
+    )
+
+    await durableObject.arm({ scheduleId: "schedule-1", revision: 2, nextDueAt: null, wakeAt: 123_456 })
+
+    expect(storage.setAlarm).toHaveBeenCalledWith(123_456)
+    expect(storage.deleteAlarm).not.toHaveBeenCalled()
+  })
+})

apps/server/src/lib/schedules/__tests__/service.test.ts

@@ -0,0 +1,42 @@
+import { describe, expect, it, vi } from "vitest"
+import { settleScheduleRuns } from "../service"
+
+function fakeDatabase(rows: Array<Record<string, unknown>>) {
+  const calls: Array<{ sql: string; values: unknown[] }> = []
+  const prepare = vi.fn((sql: string) => ({
+    bind: (...values: unknown[]) => {
+      calls.push({ sql, values })
+      return {
+        all: async () => ({ results: rows }),
+        run: async () => ({ meta: { changes: 1 } }),
+      }
+    },
+  }))
+  return { database: { prepare }, calls }
+}
+
+describe("settleScheduleRuns", () => {
+  it("releases a run stranded in preparation instead of blocking later occurrences", async () => {
+    const now = Date.UTC(2026, 8, 9, 0, 0, 0)
+    const clock = vi.spyOn(Date, "now").mockReturnValue(now)
+    const { database, calls } = fakeDatabase([
+      {
+        id: "run-1",
+        entity_key: null,
+        status: "preparing",
+        created_at: now - 3 * 60 * 60 * 1000,
+        updated_at: now - 3 * 60 * 60 * 1000,
+      },
+    ])
+
+    await settleScheduleRuns({ DB: database } as never, "schedule-1")
+
+    expect(calls.map((call) => call.sql)).toContain(
+      "UPDATE scheduled_job_runs SET status = 'failed', failure_reason = COALESCE(failure_reason, 'Scheduled run did not finish preparation'), updated_at = ? WHERE id = ? AND status = 'preparing'",
+    )
+    expect(calls.map((call) => call.sql)).toContain(
+      "UPDATE scheduled_run_slots SET run_id = NULL, lease_expires_at = NULL WHERE run_id = ?",
+    )
+    clock.mockRestore()
+  })
+})

apps/server/src/lib/schedules/__tests__/validation.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "vitest"
-import { parseScheduleInput } from "../validation"
+import { parseManualRunInput, parseScheduleInput } from "../validation"
 
 const base = {
   name: "Weekly dependency maintenance",
@@ -25,3 +25,17 @@ describe("parseScheduleInput", () => {
     expect(() => parseScheduleInput({ ...base, timezone: "Mars/Olympus" })).toThrow("timezone")
   })
 })
+
+describe("parseManualRunInput", () => {
+  it("requires an explicit confirmation and bounded string idempotency key", () => {
+    expect(parseManualRunInput({ confirm: true, idempotencyKey: "  run-123  " })).toEqual({
+      confirm: true,
+      idempotencyKey: "run-123",
+    })
+  })
+
+  it("rejects malformed replay fences", () => {
+    expect(() => parseManualRunInput({ confirm: true, idempotencyKey: {} })).toThrow()
+    expect(() => parseManualRunInput({ confirm: true, idempotencyKey: "x".repeat(121) })).toThrow()
+  })
+})

apps/server/src/lib/schedules/runner.ts

@@ -8,11 +8,18 @@ export type ScheduleArm = { scheduleId: string; revision: number; nextDueAt: num
 
 /** One isolated alarm per configured schedule. D1 remains the source of truth. */
 export class ScheduleRunner extends DurableObject<Env> {
+  /** Reconcile this runner with D1 after an operator mutation. */
+  async sync(scheduleId: string): Promise<void> {
+    await this.armNext(scheduleId)
+  }
+
   async arm(input: ScheduleArm): Promise<void> {
-    const wakeAt = input.nextDueAt === null ? null : (input.wakeAt ?? input.nextDueAt)
+    // A paused/archived schedule has no due occurrence, but an already-admitted
+    // run still needs a short-lived monitor wake-up to settle its history.
+    const wakeAt = input.wakeAt ?? input.nextDueAt
     const arm = { ...input, wakeAt }
     await this.ctx.storage.put("arm", arm)
-    if (input.nextDueAt === null) {
+    if (wakeAt === null) {
       await this.ctx.storage.deleteAlarm()
       if (input.scheduleId) {
         await this.env.DB.prepare("UPDATE scheduled_jobs SET armed_revision = ? WHERE id = ?")
@@ -27,10 +34,9 @@ export class ScheduleRunner extends DurableObject<Env> {
       .run()
   }
 
-  async runNow(idempotencyKey: string): Promise<{ runId: string; status: string } | null> {
-    const arm = await this.ctx.storage.get<ScheduleArm>("arm")
-    const schedule = arm ? await readSchedule(this.env, arm.scheduleId) : null
-    if (!schedule?.enabled || schedule.archivedAt) return null
+  async runNow(scheduleId: string, idempotencyKey: string): Promise<{ runId: string; status: string } | null> {
+    const schedule = await readSchedule(this.env, scheduleId)
+    if (!schedule || schedule.archivedAt) return null
     const result = await startScheduleRun(this.env, schedule.id, {
       trigger: "manual",
       intendedAt: Date.now(),
@@ -43,27 +49,34 @@ export class ScheduleRunner extends DurableObject<Env> {
   async alarm(): Promise<void> {
     const arm = await this.ctx.storage.get<ScheduleArm>("arm")
     const schedule = arm ? await readSchedule(this.env, arm.scheduleId) : null
-    if (!arm || !schedule?.enabled || schedule.archivedAt || schedule.nextDueAt === null) {
+    if (!arm || !schedule) {
       await this.arm({ scheduleId: arm?.scheduleId ?? "", revision: schedule?.revision ?? 0, nextDueAt: null })
       return
     }
     if (arm.revision !== schedule.revision || arm.nextDueAt !== schedule.nextDueAt) {
-      await this.arm({ scheduleId: schedule.id, revision: schedule.revision, nextDueAt: schedule.nextDueAt })
+      await this.armNext(schedule.id)
       return
     }
 
+    // Durable Object alarms retry at least once but can eventually exhaust
+    // their retry budget. Clear the confirmation before external work so the
+    // maintenance reconciler can re-arm this schedule after such a failure.
+    await this.env.DB.prepare("UPDATE scheduled_jobs SET armed_revision = NULL WHERE id = ? AND revision = ?")
+      .bind(schedule.id, schedule.revision)
+      .run()
     await settleScheduleRuns(this.env, schedule.id)
-    if (schedule.nextDueAt <= Date.now()) await startScheduleRun(this.env, schedule.id)
+    if (schedule.enabled && schedule.nextDueAt !== null && schedule.nextDueAt <= Date.now())
+      await startScheduleRun(this.env, schedule.id)
     await this.armNext(schedule.id)
   }
 
   private async armNext(scheduleId: string): Promise<void> {
     const schedule = await readSchedule(this.env, scheduleId)
-    if (!schedule?.enabled || schedule.archivedAt || schedule.nextDueAt === null) {
+    if (!schedule) {
       const arm = await this.ctx.storage.get<ScheduleArm>("arm")
       await this.arm({
-        scheduleId: schedule?.id ?? arm?.scheduleId ?? "",
-        revision: schedule?.revision ?? 0,
+        scheduleId: arm?.scheduleId ?? "",
+        revision: 0,
         nextDueAt: null,
       })
       return
@@ -73,7 +86,21 @@ export class ScheduleRunner extends DurableObject<Env> {
     )
       .bind(schedule.id)
       .first<{ id: string }>()
-    if (active?.id) await renewSlot(this.env, active.id)
+    if (active?.id) {
+      await renewSlot(this.env, active.id)
+      if (!schedule.enabled || schedule.archivedAt || schedule.nextDueAt === null) {
+        await this.arm({
+          scheduleId: schedule.id,
+          revision: schedule.revision,
+          nextDueAt: null,
+          wakeAt: Date.now() + 2 * 60 * 1000,
+        })
+        return
+      }
+    } else if (!schedule.enabled || schedule.archivedAt || schedule.nextDueAt === null) {
+      await this.arm({ scheduleId: schedule.id, revision: schedule.revision, nextDueAt: null })
+      return
+    }
     await this.arm({
       scheduleId: schedule.id,
       revision: schedule.revision,