From 1ec2ba7ce196988e35c1bbeb6dcdd97d646b4460 Mon Sep 17 00:00:00 2001 From: wangtsiao Date: Sun, 8 Nov 2026 23:43:59 +0800 Subject: [PATCH] fix(desktop): harden automation scheduling and restart recovery --- apps/desktop/src/main/automation/index.ts | 31 ++ .../src/main/automation/lifecycle.test.ts | 81 ++++ .../src/main/automation/recovery.test.ts | 88 +++++ apps/desktop/src/main/automation/recovery.ts | 51 +++ .../src/main/automation/scheduler.test.ts | 177 +++++++++ apps/desktop/src/main/automation/scheduler.ts | 346 +++++++----------- apps/desktop/src/renderer/lib/rrule-ui.ts | 22 +- .../src/shared/automation-schedule.test.ts | 114 ++++++ .../desktop/src/shared/automation-schedule.ts | 105 ++++++ docs/desktop-automation-lifecycle.md | 117 ++++++ 10 files changed, 904 insertions(+), 228 deletions(-) create mode 100644 apps/desktop/src/main/automation/lifecycle.test.ts create mode 100644 apps/desktop/src/main/automation/recovery.test.ts create mode 100644 apps/desktop/src/main/automation/recovery.ts create mode 100644 apps/desktop/src/main/automation/scheduler.test.ts create mode 100644 apps/desktop/src/shared/automation-schedule.test.ts create mode 100644 apps/desktop/src/shared/automation-schedule.ts create mode 100644 docs/desktop-automation-lifecycle.md diff --git a/apps/desktop/src/main/automation/index.ts b/apps/desktop/src/main/automation/index.ts index 2a358cbd..201e548b 100644 --- a/apps/desktop/src/main/automation/index.ts +++ b/apps/desktop/src/main/automation/index.ts @@ -11,6 +11,7 @@ import { BrowserWindow } from "electron" import { createLogger } from "../logger" import { closeDb, ensureDb, getDb } from "./database" import { executeRun } from "./executor" +import { recoverInterruptedRuns } from "./recovery" import { createConfig, deleteConfig, listConfigs, readConfig, updateConfig } from "./registry" import { addTask, getNextRunTime, previewSchedule, removeTask, stopAll } from "./scheduler" import { automationRuns, automations } from "./schema" @@ -25,6 +26,8 @@ import type { const log = createLogger("automation") const semaphore = new Semaphore(5) +let initialization: Promise | null = null +let shuttingDown = false // ============================================================ // Broadcast helper @@ -43,12 +46,22 @@ function broadcastRunsUpdated(): void { /** Initialize the automation subsystem. Call once at app startup. */ export async function initAutomations(): Promise { + if (shuttingDown) throw new Error("Automation subsystem is shutting down") + initialization ??= initializeAutomations() + await initialization + if (shuttingDown) throw new Error("Automation subsystem is shutting down") +} + +async function initializeAutomations(): Promise { log.info("Initializing automation subsystem") const db = await ensureDb() + const interruptedCount = await recoverInterruptedRuns(db) + if (interruptedCount > 0) log.warn("Recovered interrupted automation records", { interruptedCount }) // Load all active automations and schedule them const configs = listConfigs() for (const config of configs) { + if (shuttingDown) return if (config.status !== "active") continue // Ensure timing row exists in SQLite @@ -72,6 +85,7 @@ export async function initAutomations(): Promise { /** Shut down the automation subsystem. Call on app quit. */ export function shutdownAutomations(): void { + shuttingDown = true stopAll() closeDb() log.info("Automation subsystem shut down") @@ -95,6 +109,9 @@ async function scheduleAndPersistNextRun( await executeAutomation(id) }) + // A quit during asynchronous scheduling must not reopen the closed database. + if (shuttingDown) return + // Persist the computed nextRunAt const db = getDb() await db @@ -117,6 +134,7 @@ async function scheduleAndPersistNextRun( // ============================================================ export async function listAutomations(): Promise { + await initAutomations() const configs = listConfigs() const db = getDb() const results: Automation[] = [] @@ -128,6 +146,7 @@ export async function listAutomations(): Promise { } export async function getAutomation(id: string): Promise { + await initAutomations() const config = readConfig(id) if (!config) return null const db = getDb() @@ -136,6 +155,7 @@ export async function getAutomation(id: string): Promise { } export async function createAutomation(input: CreateAutomationInput): Promise { + await initAutomations() const id = createConfig(input) const db = getDb() const now = Date.now() @@ -159,6 +179,7 @@ export async function createAutomation(input: CreateAutomationInput): Promise { + await initAutomations() const before = readConfig(input.id) if (!before) return null @@ -191,6 +212,7 @@ export async function updateAutomation(input: UpdateAutomationInput): Promise { + await initAutomations() removeTask(id) deleteConfig(id) const db = getDb() @@ -203,6 +225,7 @@ export async function deleteAutomation(id: string): Promise { // ============================================================ export async function listRuns(automationId?: string, limit = 50): Promise { + await initAutomations() const db = getDb() if (automationId) { @@ -221,6 +244,7 @@ export async function archiveRun( runId: string, reason: "auto" | "manual" = "manual", ): Promise { + await initAutomations() const db = getDb() const result = await db .update(automationRuns) @@ -236,6 +260,7 @@ export async function archiveRun( } export async function acceptRun(runId: string): Promise { + await initAutomations() const db = getDb() const result = await db .update(automationRuns) @@ -251,6 +276,7 @@ export async function acceptRun(runId: string): Promise { /** Mark a run as read without changing its status. */ export async function markRunRead(runId: string): Promise { + await initAutomations() const db = getDb() const result = await db .update(automationRuns) @@ -272,6 +298,7 @@ export async function markRunRead(runId: string): Promise { * block the renderer. */ export async function runNow(id: string): Promise { + await initAutomations() const config = readConfig(id) if (!config) return false @@ -295,6 +322,7 @@ interface ExecuteOptions { } async function executeAutomation(id: string, opts: ExecuteOptions = {}): Promise { + await initAutomations() const config = readConfig(id) if (!config) { log.warn("Automation config not found, skipping execution", { id }) @@ -325,6 +353,7 @@ async function executeAutomation(id: string, opts: ExecuteOptions = {}): Promise }) try { + if (shuttingDown) return // Run once per workspace, or once with empty workspace if none configured const targets = config.workspaces.length > 0 ? config.workspaces : [""] log.info("Starting automation execution", { @@ -335,6 +364,7 @@ async function executeAutomation(id: string, opts: ExecuteOptions = {}): Promise timeoutSec: config.execution.timeout, }) for (const workspace of targets) { + if (shuttingDown) return const runId = crypto.randomUUID() const now = Date.now() @@ -384,6 +414,7 @@ async function executeAutomation(id: string, opts: ExecuteOptions = {}): Promise await new Promise((resolve) => setTimeout(resolve, config.execution.retryDelay * 1000)) } + if (shuttingDown) return lastResult = await executeRun(config, workspace, async (info) => { // Persist sessionId as soon as it is available so the // renderer can show the live session view immediately diff --git a/apps/desktop/src/main/automation/lifecycle.test.ts b/apps/desktop/src/main/automation/lifecycle.test.ts new file mode 100644 index 00000000..1c5b534c --- /dev/null +++ b/apps/desktop/src/main/automation/lifecycle.test.ts @@ -0,0 +1,81 @@ +import { expect, mock, test } from "bun:test" +import { mkdtempSync, rmSync } from "node:fs" +import os from "node:os" +import path from "node:path" +import { fileURLToPath } from "node:url" +import { createClient } from "@libsql/client" +import { drizzle } from "drizzle-orm/libsql" +import { migrate } from "drizzle-orm/libsql/migrator" +import * as schema from "./schema" + +test("startup is single-flight, API reads wait for recovery, and quit rejects new work", async () => { + const dir = mkdtempSync(path.join(os.tmpdir(), "devo-lifecycle-")) + const client = createClient({ url: `file:${path.join(dir, "test.db")}` }) + try { + const db = drizzle({ client, schema }) + await migrate(db, { + migrationsFolder: fileURLToPath(new URL("../../../drizzle", import.meta.url)), + }) + await db.insert(schema.automations).values({ id: "a", createdAt: 1, updatedAt: 1 }).run() + await db + .insert(schema.automationRuns) + .values({ + id: "orphan", + automationId: "a", + workspace: "", + status: "running", + createdAt: 1, + updatedAt: 1, + }) + .run() + let release!: () => void + const ready = new Promise((resolve) => { + release = resolve + }) + let initializations = 0 + mock.module("./database", () => ({ + ensureDb: async () => { + initializations++ + await ready + return db + }, + getDb: () => db, + closeDb: () => client.close(), + })) + mock.module("./registry", () => ({ + listConfigs: () => [], + readConfig: () => null, + createConfig: () => "a", + updateConfig: () => false, + deleteConfig: () => false, + })) + mock.module("./executor", () => ({ + executeRun: async () => { + throw new Error("Must not replay") + }, + })) + mock.module("electron", () => ({ BrowserWindow: { getAllWindows: () => [] } })) + const { initAutomations, listRuns, runNow, shutdownAutomations } = await import("./index") + const startup = initAutomations() + let readCompleted = false + const read = listRuns().then((runs) => { + readCompleted = true + return runs + }) + await Promise.resolve() + expect(initializations).toBe(1) + expect(readCompleted).toBe(false) + release() + await startup + const runs = await read + expect(runs.map(({ id, status }) => ({ id, status }))).toEqual([ + { id: "orphan", status: "failed" }, + ]) + shutdownAutomations() + await expect(runNow("a")).rejects.toThrow("shutting down") + await expect(listRuns()).rejects.toThrow("shutting down") + } finally { + client.close() + rmSync(dir, { recursive: true, force: true }) + } +}) diff --git a/apps/desktop/src/main/automation/recovery.test.ts b/apps/desktop/src/main/automation/recovery.test.ts new file mode 100644 index 00000000..fcf48549 --- /dev/null +++ b/apps/desktop/src/main/automation/recovery.test.ts @@ -0,0 +1,88 @@ +import { describe, expect, test } from "bun:test" +import { mkdtempSync, rmSync } from "node:fs" +import os from "node:os" +import path from "node:path" +import { fileURLToPath } from "node:url" +import { createClient } from "@libsql/client" +import { drizzle } from "drizzle-orm/libsql" +import { migrate } from "drizzle-orm/libsql/migrator" +import { INTERRUPTED_RUN_MESSAGE, recoverInterruptedRuns } from "./recovery" +import * as schema from "./schema" + +describe("automation restart recovery", () => { + test("closes interrupted records once, preserves evidence and never replays work", async () => { + const dir = mkdtempSync(path.join(os.tmpdir(), "devo-recovery-")) + const client = createClient({ url: `file:${path.join(dir, "test.db")}` }) + try { + const db = drizzle({ client, schema }) + await migrate(db, { + migrationsFolder: fileURLToPath(new URL("../../../drizzle", import.meta.url)), + }) + await db.insert(schema.automations).values({ id: "a", createdAt: 1, updatedAt: 1 }).run() + await db + .insert(schema.automationRuns) + .values([ + { + id: "running", + automationId: "a", + workspace: "project", + status: "running", + sessionId: "session", + worktreePath: "worktree", + createdAt: 1, + updatedAt: 1, + }, + { + id: "queued", + automationId: "a", + workspace: "project", + status: "queued", + createdAt: 1, + updatedAt: 1, + }, + { + id: "done", + automationId: "a", + workspace: "project", + status: "pending_review", + completedAt: 2, + createdAt: 1, + updatedAt: 2, + }, + ]) + .run() + const before = await db.select().from(schema.automationRuns).orderBy(schema.automationRuns.id) + expect(await recoverInterruptedRuns(db, 100)).toBe(2) + expect( + await db.select().from(schema.automationRuns).orderBy(schema.automationRuns.id), + ).toEqual( + before.map((run) => + run.status === "pending_review" + ? run + : { + ...run, + status: "failed", + errorMessage: INTERRUPTED_RUN_MESSAGE, + completedAt: 100, + updatedAt: 100, + }, + ), + ) + expect(await recoverInterruptedRuns(db, 200)).toBe(0) + expect(await db.select().from(schema.automations)).toEqual([ + { + id: "a", + nextRunAt: null, + lastRunAt: null, + runCount: 2, + consecutiveFailures: 2, + createdAt: 1, + updatedAt: 100, + }, + ]) + } finally { + client.close() + rmSync(dir, { recursive: true, force: true }) + } + }) +}) diff --git a/apps/desktop/src/main/automation/recovery.ts b/apps/desktop/src/main/automation/recovery.ts new file mode 100644 index 00000000..e71e5182 --- /dev/null +++ b/apps/desktop/src/main/automation/recovery.ts @@ -0,0 +1,51 @@ +import { eq, inArray, sql } from "drizzle-orm" +import type { getDb } from "./database" +import { automationRuns, automations } from "./schema" + +export const INTERRUPTED_RUN_MESSAGE = + "Devo stopped before this run completed. Its outcome is unknown; inspect the session and workspace before retrying." + +/** + * A new Desktop process cannot safely resume or replay a previous process's + * work. Close stale records atomically and retain session/worktree evidence. + * Only invoke before this process admits any new automation execution. + */ +export async function recoverInterruptedRuns( + db: ReturnType, + now = Date.now(), +): Promise { + return db.transaction(async (tx) => { + const interrupted = await tx + .select() + .from(automationRuns) + .where(inArray(automationRuns.status, ["queued", "running"])) + if (interrupted.length === 0) return 0 + + await tx + .update(automationRuns) + .set({ + status: "failed", + errorMessage: INTERRUPTED_RUN_MESSAGE, + completedAt: now, + updatedAt: now, + }) + .where(inArray(automationRuns.status, ["queued", "running"])) + .run() + + const counts = new Map() + for (const run of interrupted) + counts.set(run.automationId, (counts.get(run.automationId) ?? 0) + 1) + for (const [id, count] of counts) { + await tx + .update(automations) + .set({ + runCount: sql`${automations.runCount} + ${count}`, + consecutiveFailures: sql`${automations.consecutiveFailures} + ${count}`, + updatedAt: now, + }) + .where(eq(automations.id, id)) + .run() + } + return interrupted.length + }) +} diff --git a/apps/desktop/src/main/automation/scheduler.test.ts b/apps/desktop/src/main/automation/scheduler.test.ts new file mode 100644 index 00000000..313f45f9 --- /dev/null +++ b/apps/desktop/src/main/automation/scheduler.test.ts @@ -0,0 +1,177 @@ +import { describe, expect, test } from "bun:test" +import { AutomationScheduler, MAX_TIMER_DELAY_MS } from "./scheduler" + +function deferred() { + let resolve!: (value: T) => void + const promise = new Promise((done) => { + resolve = done + }) + return { promise, resolve } +} + +async function settle() { + for (let i = 0; i < 12; i++) await Promise.resolve() +} + +function harness(nextRunAt: number) { + let now = 0 + const timers = new Map, { callback: () => void; delay: number }>() + let occurrence: () => Promise = async () => [new Date(nextRunAt)] + const scheduler = new AutomationScheduler({ + now: () => now, + occurrences: () => occurrence(), + setTimer: (callback, delay) => { + const timer = { unref() {} } as ReturnType + timers.set(timer, { callback, delay }) + return timer + }, + clearTimer: (timer) => { + timers.delete(timer) + }, + }) + return { + scheduler, + timers, + setNow(value: number) { + now = value + }, + setOccurrence(value: () => Promise) { + occurrence = value + }, + async fire() { + const scheduled = timers.entries().next().value + if (!scheduled) throw new Error("No timer is armed") + const [timer, entry] = scheduled + timers.delete(timer) + entry.callback() + await settle() + }, + } +} + +describe("automation timer ownership", () => { + test("chunks a delay beyond Node's timer limit and never fires early", async () => { + const due = 40 * 24 * 60 * 60 * 1000 + const h = harness(due) + let runs = 0 + await h.scheduler.addTask("a", "rule", "UTC", async () => { + runs++ + }) + expect([...h.timers.values()].map((timer) => timer.delay)).toEqual([MAX_TIMER_DELAY_MS]) + h.setNow(MAX_TIMER_DELAY_MS) + await h.fire() + expect(runs).toBe(0) + expect(h.scheduler.getNextRunTime("a")).toEqual(new Date(due)) + h.setOccurrence(async () => []) + h.setNow(due) + await h.fire() + expect(runs).toBe(1) + expect(h.timers.size).toBe(0) + }) + + test("clock rollback re-arms without early execution", async () => { + const h = harness(1000) + let runs = 0 + await h.scheduler.addTask("a", "rule", "UTC", async () => { + runs++ + }) + h.setNow(-5000) + await h.fire() + expect(runs).toBe(0) + expect([...h.timers.values()].map((timer) => timer.delay)).toEqual([6000]) + }) + + test("sleep past several occurrences causes one run, then computes a future occurrence", async () => { + const h = harness(1000) + let runs = 0 + await h.scheduler.addTask("a", "rule", "UTC", async () => { + runs++ + }) + h.setNow(10_000) + h.setOccurrence(async () => [new Date(11_000)]) + await h.fire() + expect(runs).toBe(1) + expect(h.scheduler.getNextRunTime("a")).toEqual(new Date(11_000)) + }) + + for (const action of ["remove", "pause", "stop"] as const) { + test(`${action} during asynchronous date calculation cannot resurrect a timer`, async () => { + const h = harness(1000) + const pending = deferred() + h.setOccurrence(() => pending.promise) + const added = h.scheduler.addTask("a", "rule", "UTC", async () => {}) + if (action === "remove") h.scheduler.removeTask("a") + if (action === "pause") h.scheduler.pauseTask("a") + if (action === "stop") h.scheduler.stopAll() + pending.resolve([new Date(1000)]) + expect(await added).toBeNull() + expect(h.timers.size).toBe(0) + expect(h.scheduler.getNextRunTime("a")).toBeNull() + }) + } + + test("pause/resume invalidates the first pending computation", async () => { + const h = harness(1000) + const pending = deferred() + h.setOccurrence(() => pending.promise) + const added = h.scheduler.addTask("a", "rule", "UTC", async () => {}) + h.scheduler.pauseTask("a") + h.setOccurrence(async () => [new Date(2000)]) + h.scheduler.resumeTask("a") + await settle() + pending.resolve([new Date(1000)]) + await added + expect(h.timers.size).toBe(1) + expect(h.scheduler.getNextRunTime("a")).toEqual(new Date(2000)) + }) + + test("replacing an in-flight task prevents overlapping callbacks", async () => { + const h = harness(1000) + const running = deferred() + let replacementRuns = 0 + await h.scheduler.addTask("a", "rule", "UTC", () => running.promise) + h.setNow(1000) + await h.fire() + expect(h.scheduler.isRunning("a")).toBe(true) + h.setOccurrence(async () => [new Date(2000)]) + await h.scheduler.addTask("a", "new rule", "UTC", async () => { + replacementRuns++ + }) + h.setNow(2000) + h.setOccurrence(async () => [new Date(3000)]) + await h.fire() + expect(replacementRuns).toBe(0) + running.resolve() + await settle() + expect(h.timers.size).toBe(1) + h.setNow(3000) + h.setOccurrence(async () => []) + await h.fire() + expect(replacementRuns).toBe(1) + }) + + test("callback failure keeps the recurring timer chain alive", async () => { + const h = harness(1000) + await h.scheduler.addTask("a", "rule", "UTC", async () => { + throw new Error("execution failed") + }) + h.setNow(1000) + h.setOccurrence(async () => [new Date(2000)]) + await h.fire() + expect(h.scheduler.isRunning("a")).toBe(false) + expect(h.scheduler.getNextRunTime("a")).toEqual(new Date(2000)) + }) + + test("stop during an in-flight callback prevents rescheduling", async () => { + const h = harness(1000) + const running = deferred() + await h.scheduler.addTask("a", "rule", "UTC", () => running.promise) + h.setNow(1000) + await h.fire() + h.scheduler.stopAll() + running.resolve() + await settle() + expect(h.timers.size).toBe(0) + expect(h.scheduler.isRunning("a")).toBe(false) + }) +}) diff --git a/apps/desktop/src/main/automation/scheduler.ts b/apps/desktop/src/main/automation/scheduler.ts index 286e3e04..cb7102a5 100644 --- a/apps/desktop/src/main/automation/scheduler.ts +++ b/apps/desktop/src/main/automation/scheduler.ts @@ -1,267 +1,185 @@ -/** - * Automation scheduler service. - * - * Uses the rrule library for date math and setTimeout for precise - * per-automation timers. No polling -- each automation gets a single - * timer set to fire at its exact next run time. - */ - +/** In-process automation scheduler. Explicit app quit always stops scheduling. */ +import { scheduleOccurrences } from "../../shared/automation-schedule" import { createLogger } from "../logger" const log = createLogger("automation-scheduler") +// Bounded checkpoints avoid Node's ~24.8-day timeout overflow and recheck the +// wall clock after sleep or clock adjustments. They never execute early. +export const MAX_TIMER_DELAY_MS = 60_000 +type Timer = ReturnType interface ScheduledTask { - timer: ReturnType | null + timer: Timer | null rruleStr: string timezone: string + startAt: Date paused: boolean - running: boolean + generation: number callback: () => Promise - /** The computed next run time, updated after each scheduling cycle. */ nextRunAt: Date | null } -const tasks = new Map() - -// ============================================================ -// Internal helpers -// ============================================================ - -/** Human-readable delay string for log messages. */ -function formatDelay(ms: number): string { - if (ms < 1000) return `${ms}ms` - const seconds = Math.round(ms / 1000) - if (seconds < 120) return `${seconds}s` - const minutes = Math.round(seconds / 60) - if (minutes < 120) return `${minutes}m` - const hours = Math.round(minutes / 60) - return `${hours}h` +interface SchedulerDependencies { + now: () => number + setTimer: (callback: () => void, delay: number) => Timer + clearTimer: (timer: Timer) => void + occurrences: typeof scheduleOccurrences } -async function computeNextRun( - id: string, - rruleStr: string, - _timezone: string, -): Promise { - const rruleModule = await import("rrule") - // Handle CJS/ESM interop: named export in CJS, nested under default in ESM - const RRule = rruleModule.RRule ?? rruleModule.default?.RRule - try { - const rule = RRule.fromString(rruleStr) - const now = new Date() - const next = rule.after(now, false) - if (next) { - const delayMs = next.getTime() - now.getTime() - log.debug("Computed next run", { - id, - rruleStr, - now: now.toISOString(), - next: next.toISOString(), - delayMs, - }) - } else { - log.warn("RRULE produced no future occurrence", { id, rruleStr, now: now.toISOString() }) - } - return next - } catch (err) { - log.error("Failed to parse RRULE", { id, rruleStr }, err) - return null +/** Owns timer lifetimes; injected clock/timers make cancellation races testable. */ +export class AutomationScheduler { + private readonly tasks = new Map() + // Kept separately so replacing a schedule cannot overlap its in-flight run. + private readonly running = new Set() + + constructor( + private readonly dependencies: SchedulerDependencies = { + now: Date.now, + setTimer: setTimeout, + clearTimer: clearTimeout, + occurrences: scheduleOccurrences, + }, + ) {} + + private isCurrent(id: string, task: ScheduledTask, generation: number): boolean { + return this.tasks.get(id) === task && !task.paused && task.generation === generation } -} -async function scheduleNext(id: string, task: ScheduledTask): Promise { - if (task.timer) { - clearTimeout(task.timer) + private async scheduleNext(id: string, task: ScheduledTask): Promise { + const generation = ++task.generation + if (task.timer !== null) this.dependencies.clearTimer(task.timer) task.timer = null - log.debug("Cleared previous timer", { id }) - } - - if (task.paused) { - log.debug("Task is paused, skipping schedule", { id }) task.nextRunAt = null - return null - } + if (task.paused || this.tasks.get(id) !== task) return null - const next = await computeNextRun(id, task.rruleStr, task.timezone) - if (!next) { - log.warn("No next occurrence for automation, timer chain stopped", { - id, - rruleStr: task.rruleStr, - }) - task.nextRunAt = null - return null + try { + const [next] = await this.dependencies.occurrences(task.rruleStr, task.timezone, { + after: new Date(this.dependencies.now()), + count: 1, + startAt: task.startAt, + }) + // Removing, pausing, replacing or stopping while date math awaits must + // not resurrect a timer, even if a pause was followed by a resume. + if (!this.isCurrent(id, task, generation)) return null + task.nextRunAt = next ?? null + if (next) this.armTimer(id, task, generation, next) + return next ?? null + } catch (err) { + log.error("Unable to schedule automation", { id, timezone: task.timezone }, err) + return null + } } - task.nextRunAt = next - const delay = Math.max(0, next.getTime() - Date.now()) - log.info("Timer set", { - id, - nextRun: next.toISOString(), - delayMs: delay, - delayHuman: formatDelay(delay), - }) + private armTimer(id: string, task: ScheduledTask, generation: number, next: Date): void { + const delay = Math.min( + MAX_TIMER_DELAY_MS, + Math.max(0, next.getTime() - this.dependencies.now()), + ) + task.timer = this.dependencies.setTimer(() => { + if (!this.isCurrent(id, task, generation)) return + task.timer = null + if (this.dependencies.now() < next.getTime()) { + this.armTimer(id, task, generation, next) + return + } + void this.fire(id, task, generation) + }, delay) + task.timer?.unref?.() + } - task.timer = setTimeout(async () => { - // Verify this task is still the active one (guards against stale closures - // after addTask replaces the task while a callback is in-flight) - if (tasks.get(id) !== task) { - log.warn("Timer fired for stale task, ignoring", { id }) - return - } - if (task.paused) { - log.warn("Timer fired but task is paused, ignoring", { id }) - return - } - if (task.running) { - log.warn("Timer fired but task is already running, ignoring", { id }) + private async fire(id: string, task: ScheduledTask, generation: number): Promise { + task.nextRunAt = null + if (this.running.has(id)) { + // Skip an overlapping occurrence, never queue a backlog of side effects. + await this.scheduleNext(id, task) return } - - log.info("Timer fired, starting automation callback", { id, scheduledFor: next.toISOString() }) - task.running = true - const startTime = Date.now() + this.running.add(id) try { await task.callback() - log.info("Automation callback completed", { - id, - durationMs: Date.now() - startTime, - }) } catch (err) { - log.error("Automation callback threw", { id, durationMs: Date.now() - startTime }, err) + log.error("Automation callback threw", { id }, err) } finally { - task.running = false - // Schedule the next run. Await and catch so a rejection doesn't - // silently kill the recurring timer chain. - if (tasks.get(id) === task && !task.paused) { - log.debug("Rescheduling after run", { id }) - try { - await scheduleNext(id, task) - } catch (err) { - log.error("Failed to reschedule automation after run -- timer chain broken", { id }, err) - } - } else { - log.warn("Skipping reschedule", { - id, - reason: tasks.get(id) !== task ? "task replaced" : "task paused", - }) - } + this.running.delete(id) + if (this.isCurrent(id, task, generation)) await this.scheduleNext(id, task) } - }, delay) - - // Don't prevent app exit - if (task.timer && typeof task.timer === "object" && "unref" in task.timer) { - task.timer.unref() } - return next -} - -// ============================================================ -// Public API -// ============================================================ - -export async function addTask( - id: string, - rruleStr: string, - timezone: string, - callback: () => Promise, -): Promise { - log.info("Adding task", { id, rruleStr, timezone, hadExisting: tasks.has(id) }) - removeTask(id) - const task: ScheduledTask = { - timer: null, - rruleStr, - timezone, - paused: false, - running: false, - callback, - nextRunAt: null, + async addTask( + id: string, + rruleStr: string, + timezone: string, + callback: () => Promise, + ): Promise { + this.removeTask(id) + const task: ScheduledTask = { + timer: null, + rruleStr, + timezone, + callback, + startAt: new Date(this.dependencies.now()), + paused: false, + generation: 0, + nextRunAt: null, + } + this.tasks.set(id, task) + return this.scheduleNext(id, task) } - tasks.set(id, task) - const next = await scheduleNext(id, task) - log.info("Task added", { - id, - nextRun: next?.toISOString() ?? "none", - totalTasks: tasks.size, - }) - return next -} -export function removeTask(id: string): void { - const task = tasks.get(id) - if (task) { - if (task.timer) clearTimeout(task.timer) - tasks.delete(id) - log.info("Task removed", { id, wasRunning: task.running, totalTasks: tasks.size }) + removeTask(id: string): void { + const task = this.tasks.get(id) + if (task?.timer !== null && task?.timer !== undefined) this.dependencies.clearTimer(task.timer) + this.tasks.delete(id) } -} -export function pauseTask(id: string): void { - const task = tasks.get(id) - if (task) { + pauseTask(id: string): void { + const task = this.tasks.get(id) + if (!task) return task.paused = true - if (task.timer) { - clearTimeout(task.timer) - task.timer = null - } - log.info("Task paused", { id }) + task.generation++ + task.nextRunAt = null + if (task.timer !== null) this.dependencies.clearTimer(task.timer) + task.timer = null } -} -export function resumeTask(id: string): void { - const task = tasks.get(id) - if (task?.paused) { + resumeTask(id: string): void { + const task = this.tasks.get(id) + if (!task?.paused) return task.paused = false - log.info("Task resumed, rescheduling", { id }) - scheduleNext(id, task) + void this.scheduleNext(id, task) } -} -export function isRunning(id: string): boolean { - return tasks.get(id)?.running ?? false -} + isRunning(id: string): boolean { + return this.running.has(id) + } -/** - * Returns the computed next run time for a scheduled task, or null if - * the task doesn't exist, is paused, or has no future occurrences. - */ -export function getNextRunTime(id: string): Date | null { - return tasks.get(id)?.nextRunAt ?? null -} + getNextRunTime(id: string): Date | null { + return this.tasks.get(id)?.nextRunAt ?? null + } -export function stopAll(): void { - log.info("Stopping all scheduled automations", { count: tasks.size }) - for (const [id, task] of tasks) { - if (task.timer) clearTimeout(task.timer) - log.debug("Stopped task", { id, wasRunning: task.running }) + stopAll(): void { + for (const id of this.tasks.keys()) this.removeTask(id) } - tasks.clear() - log.info("All scheduled automations stopped") } -/** - * Compute the next N occurrences for an RRULE (for UI preview). - */ +const scheduler = new AutomationScheduler() +export const addTask = scheduler.addTask.bind(scheduler) +export const removeTask = scheduler.removeTask.bind(scheduler) +export const pauseTask = scheduler.pauseTask.bind(scheduler) +export const resumeTask = scheduler.resumeTask.bind(scheduler) +export const isRunning = scheduler.isRunning.bind(scheduler) +export const getNextRunTime = scheduler.getNextRunTime.bind(scheduler) +export const stopAll = scheduler.stopAll.bind(scheduler) + export async function previewSchedule( rruleStr: string, - _timezone: string, - count: number = 5, + timezone: string, + count = 5, ): Promise { - const rruleModule = await import("rrule") - // Handle CJS/ESM interop: named export in CJS, nested under default in ESM - const RRule = rruleModule.RRule ?? rruleModule.default?.RRule try { - const rule = RRule.fromString(rruleStr) - const dates: Date[] = [] - let current = new Date() - for (let i = 0; i < count; i++) { - const next = rule.after(current, false) - if (!next) break - dates.push(next) - current = next - } - return dates - } catch { + return await scheduleOccurrences(rruleStr, timezone, { count }) + } catch (err) { + log.warn("Invalid automation schedule", { rruleStr, timezone }, err) return [] } } diff --git a/apps/desktop/src/renderer/lib/rrule-ui.ts b/apps/desktop/src/renderer/lib/rrule-ui.ts index 637f13b9..dc15c59b 100644 --- a/apps/desktop/src/renderer/lib/rrule-ui.ts +++ b/apps/desktop/src/renderer/lib/rrule-ui.ts @@ -8,6 +8,8 @@ * next-runs preview using the rrule library. */ +import { scheduleOccurrences } from "../../shared/automation-schedule" + // ============================================================ // Types // ============================================================ @@ -250,21 +252,13 @@ export function formatScheduleSummary(config: ScheduleConfig): string { * Compute the next N occurrences for an RRULE string, entirely client-side. * Returns ISO date strings. Falls back to empty array on parse errors. */ -export async function computeNextRuns(rruleStr: string, count = 3): Promise { +export async function computeNextRuns( + rruleStr: string, + count = 3, + timezone = Intl.DateTimeFormat().resolvedOptions().timeZone, +): Promise { try { - const rruleModule = await import("rrule") - // Handle CJS/ESM interop: named export in CJS, nested under default in ESM - const RRule = rruleModule.RRule ?? rruleModule.default?.RRule - const rule = RRule.fromString(rruleStr) - const dates: Date[] = [] - let current = new Date() - for (let i = 0; i < count; i++) { - const next = rule.after(current, false) - if (!next) break - dates.push(next) - current = next - } - return dates + return await scheduleOccurrences(rruleStr, timezone, { count }) } catch { return [] } diff --git a/apps/desktop/src/shared/automation-schedule.test.ts b/apps/desktop/src/shared/automation-schedule.test.ts new file mode 100644 index 00000000..f3e202c0 --- /dev/null +++ b/apps/desktop/src/shared/automation-schedule.test.ts @@ -0,0 +1,114 @@ +import { describe, expect, test } from "bun:test" +import { scheduleOccurrences } from "./automation-schedule" + +const daily = "FREQ=DAILY;BYHOUR=9;BYMINUTE=0;BYSECOND=0" +const iso = (dates: Date[]) => dates.map((date) => date.toISOString()) + +describe("automation calendar timezone", () => { + test("uses the configured timezone, including fractional offsets and date rollover", async () => { + const after = new Date("2026-11-08T00:00:00Z") + expect(iso(await scheduleOccurrences(daily, "Asia/Shanghai", { after, count: 2 }))).toEqual([ + "2026-11-08T01:00:00.000Z", + "2026-11-09T01:00:00.000Z", + ]) + expect(iso(await scheduleOccurrences(daily, "Asia/Kathmandu", { after, count: 1 }))).toEqual([ + "2026-11-08T03:15:00.000Z", + ]) + expect(iso(await scheduleOccurrences(daily, "Pacific/Auckland", { after, count: 1 }))).toEqual([ + "2026-11-08T20:00:00.000Z", + ]) + }) + + test("keeps 9am wall time across daylight saving changes", async () => { + expect( + iso( + await scheduleOccurrences(daily, "America/New_York", { + after: new Date("2026-03-07T00:00:00Z"), + count: 3, + }), + ), + ).toEqual(["2026-03-07T14:00:00.000Z", "2026-03-08T13:00:00.000Z", "2026-03-09T13:00:00.000Z"]) + }) + + test("skips nonexistent spring-forward times", async () => { + expect( + iso( + await scheduleOccurrences( + "FREQ=DAILY;BYHOUR=2;BYMINUTE=30;BYSECOND=0", + "America/New_York", + { + after: new Date("2026-03-08T00:00:00Z"), + count: 1, + }, + ), + ), + ).toEqual(["2026-03-09T06:30:00.000Z"]) + }) + + test("fires repeated fall-back times once at the earlier instant", async () => { + const rule = "FREQ=DAILY;BYHOUR=1;BYMINUTE=30;BYSECOND=0" + const zone = "America/New_York" + expect( + iso( + await scheduleOccurrences(rule, zone, { + after: new Date("2026-11-01T00:00:00Z"), + count: 2, + }), + ), + ).toEqual(["2026-11-01T05:30:00.000Z", "2026-11-02T06:30:00.000Z"]) + expect( + iso( + await scheduleOccurrences(rule, zone, { + after: new Date("2026-11-01T06:15:00Z"), + count: 1, + }), + ), + ).toEqual(["2026-11-02T06:30:00.000Z"]) + }) + + test("handles half-hour DST transitions", async () => { + expect( + iso( + await scheduleOccurrences( + "FREQ=DAILY;BYHOUR=2;BYMINUTE=15;BYSECOND=0", + "Australia/Lord_Howe", + { + after: new Date("2026-10-03T16:00:00Z"), + count: 1, + }, + ), + ), + ).toEqual(["2026-10-04T15:15:00.000Z"]) + }) + + test("honors UTC UNTIL and returns no exhausted occurrences", async () => { + expect( + iso( + await scheduleOccurrences(`${daily};UNTIL=20261109T010000Z`, "Asia/Shanghai", { + after: new Date("2026-11-08T00:00:00Z"), + count: 3, + }), + ), + ).toEqual(["2026-11-08T01:00:00.000Z", "2026-11-09T01:00:00.000Z"]) + }) + + test("keeps an interval anchor across recalculations", async () => { + expect( + iso( + await scheduleOccurrences("FREQ=MINUTELY;INTERVAL=30", "Asia/Shanghai", { + startAt: new Date("2026-11-08T00:00:00Z"), + after: new Date("2026-11-08T00:32:00Z"), + count: 1, + }), + ), + ).toEqual(["2026-11-08T01:00:00.000Z"]) + }) + + test("rejects invalid zones, rules and conflicting embedded TZID", async () => { + await expect(scheduleOccurrences(daily, "Invalid/Zone")).rejects.toThrow() + await expect(scheduleOccurrences("invalid", "UTC")).rejects.toThrow() + await expect( + scheduleOccurrences("DTSTART;TZID=Asia/Tokyo:20261108T090000\nRRULE:FREQ=DAILY", "UTC"), + ).rejects.toThrow() + }) +}) diff --git a/apps/desktop/src/shared/automation-schedule.ts b/apps/desktop/src/shared/automation-schedule.ts new file mode 100644 index 00000000..3314d5e9 --- /dev/null +++ b/apps/desktop/src/shared/automation-schedule.ts @@ -0,0 +1,105 @@ +/** Calendar recurrence calculation shared by the scheduler and its preview. */ + +const HOUR_MS = 60 * 60 * 1000 + +function wallTime(date: Date, formatter: Intl.DateTimeFormat): Date { + const parts = Object.fromEntries( + formatter.formatToParts(date).map(({ type, value }) => [type, value]), + ) + return new Date( + Date.UTC( + Number(parts.year), + Number(parts.month) - 1, + Number(parts.day), + Number(parts.hour), + Number(parts.minute), + Number(parts.second), + ), + ) +} + +/** + * Resolve a calendar time without depending on the host process's timezone. + * Missing spring-forward times are skipped; repeated fall-back times run once, + * at the earlier instant. Check both sides of transitions, including half-hour DST. + */ +function resolveWallTime(wall: Date, formatter: Intl.DateTimeFormat): Date | null { + const timestamp = wall.getTime() + const offsets = new Set() + for (const hours of [-36, 0, 36]) { + const sample = new Date(timestamp + hours * HOUR_MS) + offsets.add(wallTime(sample, formatter).getTime() - sample.getTime()) + } + const matches: number[] = [] + for (const offset of offsets) { + const candidate = timestamp - offset + if (wallTime(new Date(candidate), formatter).getTime() === timestamp) { + matches.push(candidate) + } + } + return matches.length > 0 ? new Date(Math.min(...matches)) : null +} + +export async function scheduleOccurrences( + rruleStr: string, + timezone: string, + { + after = new Date(), + count = 5, + startAt = after, + }: { + after?: Date + count?: number + startAt?: Date + } = {}, +): Promise { + if (!Number.isInteger(count) || count < 0 || count > 100) { + throw new RangeError("Schedule preview count must be an integer between 0 and 100") + } + // Validate even when count is zero. Invalid zones must never fall back to UTC. + const formatter = new Intl.DateTimeFormat("en-GB", { + timeZone: timezone, + calendar: "gregory", + numberingSystem: "latn", + hourCycle: "h23", + year: "numeric", + month: "2-digit", + day: "2-digit", + hour: "2-digit", + minute: "2-digit", + second: "2-digit", + }) + const rruleModule = await import("rrule") + const RRule = rruleModule.RRule ?? rruleModule.default?.RRule + const options = RRule.parseString(rruleStr) + if (options.tzid && options.tzid !== timezone) { + throw new Error("RRULE TZID must match the automation timezone") + } + const until = options.until + // rrule's TZID conversion depends on the host timezone. Generate floating + // wall-clock fields instead and convert them to real instants ourselves. + const rule = new RRule({ + ...options, + tzid: null, + until: null, + dtstart: options.dtstart + ? /DTSTART:\d{8}T\d{6}Z/i.test(rruleStr) + ? wallTime(options.dtstart, formatter) + : options.dtstart + : wallTime(startAt, formatter), + }) + const dates: Date[] = [] + let cursor = wallTime(after, formatter) + // Include the current wall second, then filter by the actual instant below. + let inclusive = true + while (dates.length < count) { + const next = rule.after(cursor, inclusive) + if (!next) break + inclusive = false + cursor = next + const instant = resolveWallTime(next, formatter) + if (instant && until && instant > until) break + if (instant && instant > after) dates.push(instant) + } + return dates +} diff --git a/docs/desktop-automation-lifecycle.md b/docs/desktop-automation-lifecycle.md new file mode 100644 index 00000000..0a65c2e7 --- /dev/null +++ b/docs/desktop-automation-lifecycle.md @@ -0,0 +1,117 @@ +# Desktop automation lifecycle and background-runtime roadmap + +## What this implementation guarantees + +Desktop automation currently runs inside the Electron main process. Keeping a +window hidden or closed is not the same as quitting the application. This change +hardens that in-process scheduler; it does **not** introduce a daemon, an OS +service, or a 24/7 execution guarantee. + +### Calendar and clock behavior + +- The saved IANA `schedule.timezone` controls the schedule's wall-clock hours and + weekdays, independently of the machine's current timezone. Preview and execution + share the same date calculation. The creation dialog previews in the local zone + it will save. Invalid zones and invalid rules produce no scheduled execution. +- Daily 09:00 remains 09:00 across DST. A nonexistent spring-forward local time is + skipped. A repeated fall-back local time fires once at the earlier instant. +- An embedded RRULE `TZID` must match the saved timezone. UTC `DTSTART` is converted + to that zone; a floating/TZID `DTSTART` supplies wall-clock fields. `UNTIL` is an + absolute cutoff. The supported input is one RRULE, not a calendar/RRULE set. +- A live timer retains its recurrence anchor across executions. Rules without an + explicit `DTSTART` are re-anchored on registration/restart; durable interval + phase across restarts remains future work. Interval previews are estimates + based on preview time rather than the live task's anchor. +- Timers use at most 60-second checkpoints. Long delays cannot overflow Node's + signed 32-bit timeout limit. Each checkpoint rechecks the actual clock before + execution, so a backward clock adjustment cannot trigger early work. +- If the process survives sleep or a forward clock jump, one overdue pending + occurrence may run on wake. The next occurrence is then calculated after the + callback completes. Missed occurrences are not replayed as a backlog. +- A replacement schedule cannot overlap the same automation's scheduled callback + already in flight. This is an in-process scheduled-run guard, not a durable + exactly-once guarantee or a manual-Run-now deduplication policy. + +### Stop and restart behavior + +- macOS: closing the last window leaves Electron running. Explicit Quit stops it. +- Windows: the normal main-window close action hides to the tray. Explicit Quit + stops the process, local server, and automation scheduling. +- Linux: closing the last window follows the existing application quit behavior. +- Explicit quit never silently becomes “keep running in the background.” The + scheduler stops and admits no new automatic work. In-flight actions may already + have changed files or remote systems; stopping the process cannot undo them. +- On startup, before API calls or new executions are admitted, old `queued` and + `running` rows become `failed` with an explicit **outcome unknown** interruption + message. Their session ID, worktree, and result evidence are preserved. Failure + counters and records are updated in one SQLite transaction, idempotently. +- Interrupted runs are **not automatically replayed**. Inspect the session and + workspace before choosing Run now. Automations active in config are scheduled + for their next future occurrence; downtime is not backfilled. +- Remove, pause, replacement, and stop invalidate pending date calculations. An + asynchronous calculation cannot resurrect a cancelled timer. A cancelled or + replaced in-flight callback cannot re-arm its former schedule on completion. + +Startup recovery assumes the existing single Desktop owner of this database. +Multiple independent processes sharing the database, manual/scheduled execution +coordination, durable claims, and crash-safe side-effect deduplication are not +provided by this slice. + +## Verification + +Automated tests cover configured zones, fractional offsets, date rollover, both +DST transitions, half-hour DST, UTC cutoffs, interval phase within one process, +invalid input, long timers, clock rollback, sleep catch-up, asynchronous +cancellation, replacement during execution, callback failure, shutdown, and +transactional restart recovery against a real temporary SQLite database. + +Run the calendar tests in separate processes with multiple host zones, for example: + +```sh +cd apps/desktop +TZ=UTC bun test src/shared/automation-schedule.test.ts +TZ=America/Los_Angeles bun test src/shared/automation-schedule.test.ts +bun test --isolate +bunx tsc --noEmit +``` + +OS-level sleep/wake, tray, logout, reboot, and forced-process-termination tests +still require packaged applications on supported Windows/macOS/Linux hosts. +Unit tests do not establish those operating-system guarantees. + +## Roadmap to a persistent background runtime + +1. **Define ownership and the service contract.** Move execution ownership to one + user-scoped runtime using the Native protocol. Desktop becomes an attachable + client. Define process identity, authenticated local transport, version + negotiation, status/health, cancellation, and explicit stop versus detach. + Acceptance: detach/reattach does not restart a turn or lose its event cursor; + incompatible clients fail visibly without spawning a competing owner. +2. **Persist schedule claims and run state.** Give every occurrence a durable + identity, timezone, anchor, claim lease and fencing token. Persist intent before + execution, transitions during work, and results before acknowledging completion. + Define per-automation skip/coalesce/catch-up policy and concurrency limits. + Acceptance: crash at each transition never creates two owners of one run; + expired leases recover without blindly repeating an uncertain external action. +3. **Supervise the process explicitly.** Provide opt-in per-user launchd, Windows + service/task, and systemd integrations with bounded restart backoff, log + rotation, health checks, and upgrade handover. No silent service installation or + expansion of unattended permissions. Acceptance: sign-in/reboot/upgrade tests, + explicit Stop remains stopped, restart storms are bounded, and secrets remain + within the existing credential boundary. +4. **Recover sessions and observe outcomes.** Reattach to Native sessions with + durable event cursors, preserve artifact/worktree references, show interrupted, + blocked, stale and unknown outcomes, and expose queue/lease/scheduler health. + Approval requests must remain blocked until an authorized decision arrives. + Acceptance: a disconnected Desktop can reconnect, find every child session and + artifact, and distinguish a finished action from one requiring reconciliation. +5. **Prove long-duration operation.** Run multi-day soak and fault-injection tests: + suspend, wall-clock jumps/DST, network loss, provider throttling, process kill, + disk full, corrupt configuration, lease contention and upgrades. Publish + measured recovery latency, missed/duplicate occurrence counts, resource usage, + and supported host conditions. Only then describe the supported always-on + behavior, with power/network/permission limits stated explicitly. + +The daemon work must align with the subagent supervisor and Factory lifecycle. +An idle kernel cell is not an authorized background host lifetime: do not create +unowned async tasks merely to simulate durable execution.