fix(desktop): harden automation timezone and restart recovery

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<void> | 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<void> {
+	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<void> {
 	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<void> {
 
 /** 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<Automation[]> {
+	await initAutomations()
 	const configs = listConfigs()
 	const db = getDb()
 	const results: Automation[] = []
@@ -128,6 +146,7 @@ export async function listAutomations(): Promise<Automation[]> {
 }
 
 export async function getAutomation(id: string): Promise<Automation | null> {
+	await initAutomations()
 	const config = readConfig(id)
 	if (!config) return null
 	const db = getDb()
@@ -136,6 +155,7 @@ export async function getAutomation(id: string): Promise<Automation | null> {
 }
 
 export async function createAutomation(input: CreateAutomationInput): Promise<Automation> {
+	await initAutomations()
 	const id = createConfig(input)
 	const db = getDb()
 	const now = Date.now()
@@ -159,6 +179,7 @@ export async function createAutomation(input: CreateAutomationInput): Promise<Au
 }
 
 export async function updateAutomation(input: UpdateAutomationInput): Promise<Automation | null> {
+	await initAutomations()
 	const before = readConfig(input.id)
 	if (!before) return null
 
@@ -191,6 +212,7 @@ export async function updateAutomation(input: UpdateAutomationInput): Promise<Au
 }
 
 export async function deleteAutomation(id: string): Promise<boolean> {
+	await initAutomations()
 	removeTask(id)
 	deleteConfig(id)
 	const db = getDb()
@@ -203,6 +225,7 @@ export async function deleteAutomation(id: string): Promise<boolean> {
 // ============================================================
 
 export async function listRuns(automationId?: string, limit = 50): Promise<AutomationRun[]> {
+	await initAutomations()
 	const db = getDb()
 
 	if (automationId) {
@@ -221,6 +244,7 @@ export async function archiveRun(
 	runId: string,
 	reason: "auto" | "manual" = "manual",
 ): Promise<boolean> {
+	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<boolean> {
+	await initAutomations()
 	const db = getDb()
 	const result = await db
 		.update(automationRuns)
@@ -251,6 +276,7 @@ export async function acceptRun(runId: string): Promise<boolean> {
 
 /** Mark a run as read without changing its status. */
 export async function markRunRead(runId: string): Promise<boolean> {
+	await initAutomations()
 	const db = getDb()
 	const result = await db
 		.update(automationRuns)
@@ -272,6 +298,7 @@ export async function markRunRead(runId: string): Promise<boolean> {
  * block the renderer.
  */
 export async function runNow(id: string): Promise<boolean> {
+	await initAutomations()
 	const config = readConfig(id)
 	if (!config) return false
 
@@ -295,6 +322,7 @@ interface ExecuteOptions {
 }
 
 async function executeAutomation(id: string, opts: ExecuteOptions = {}): Promise<void> {
+	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

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<void>((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 })
+	}
+})

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 })
+		}
+	})
+})

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<typeof getDb>,
+	now = Date.now(),
+): Promise<number> {
+	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<string, number>()
+		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
+	})
+}