From 5e8c75401d7e5b58ae2543dd3b32e30aa6232637 Mon Sep 17 00:00:00 2001 From: Marc Warne Date: Thu, 23 Jul 2026 13:07:54 +0100 Subject: [PATCH] fix: coordinate reminders across processes --- README.md | 28 +++ index.ts | 38 ++-- leases.ts | 175 +++++++++++++++ package.json | 1 + scheduler.ts | 290 +++++++++++++++---------- storage.ts | 22 +- test/README.md | 166 +------------- test/fixtures/lease-child.ts | 21 ++ test/fixtures/scheduler-child.ts | 37 ++++ test/leases.test.ts | 169 +++++++++++++++ test/scheduler.test.ts | 360 +++++++++++++++++++++++++++++-- test/storage.test.ts | 28 +++ 12 files changed, 1031 insertions(+), 304 deletions(-) create mode 100644 leases.ts create mode 100644 test/fixtures/lease-child.ts create mode 100644 test/fixtures/scheduler-child.ts create mode 100644 test/leases.test.ts diff --git a/README.md b/README.md index cb34e9b..b52bbe8 100644 --- a/README.md +++ b/README.md @@ -65,6 +65,34 @@ OpenCode will automatically load the plugin when you start the TUI. 3. **Execution** - Sends prompt to session when timer fires 4. **Cleanup** - Removes reminders when session deleted +## Multi-Process Semantics + +When independent OpenCode server processes load the same reminder directory on one host, an +occurrence lease prevents them from submitting the same prompt concurrently. Leases and reminder +JSON updates use same-directory atomic filesystem operations. A losing process keeps an unref'd +reconciliation timer, reloads durable state, and schedules the winner's next recurring occurrence. +Post-prompt recurrence updates and every plugin cancellation/restore cleanup share a short-lived +per-reminder mutation lock, so the durable read-and-transition cannot race a deletion. The lock is +not held while the prompt API runs. + +This is deliberately not a claim of crash-proof exactly-once delivery. Any process, storage, or lock +failure after the prompt API accepts a request but before its durable transition can lead to retry. +That is at-least-once crash semantics; true exactly-once delivery requires idempotency support in the +prompt API. Likewise, cancellation does not hold a lock across prompting: if an owner has already +passed durable validation, a prompt can still be submitted even when cancellation returns first. +The post-prompt durable transition remains fenced, so cancellation cannot intentionally resurrect or +reschedule the reminder after deletion. + +The lease protocol assumes same-host processes and a local filesystem with atomic hard links, +exclusive create, and rename. It is not intended to coordinate hosts or provide NFS/distributed +filesystem safety. Linux detects a dead or reused PID using `/proc//stat` process start time. +Other platforms may acquire leases normally, but only reclaim a lease when `process.kill(pid, 0)` +reports `ESRCH`; a reused PID is conservatively treated as live and can leave a stale lease until +manual cleanup rather than risk a duplicate prompt. + +A stale `.recover` guard is never reclaimed automatically. This avoids a replacement race that could +delete a live guard or lease, at the cost of manual cleanup after a recovery-process crash. + ## Data Storage Reminders stored in: diff --git a/index.ts b/index.ts index 56a1c2e..f81c9f0 100644 --- a/index.ts +++ b/index.ts @@ -2,8 +2,8 @@ import { Plugin } from "@opencode-ai/plugin" import { logger } from "./logger" import type { State, PluginConfig } from "./types" import { ReminderSchema } from "./types" -import { getStorageDir, deleteReminder, listReminders } from "./storage" -import { scheduleTimer, cancelReminder } from "./scheduler" +import { getStorageDir, listReminders } from "./storage" +import { scheduleTimer, cancelReminder, cleanupReminderSnapshot } from "./scheduler" import { createReminderAddTool } from "./tools/reminderadd" import { createReminderListTool } from "./tools/reminderlist" import { createReminderRemoveTool } from "./tools/reminderremove" @@ -40,22 +40,30 @@ const RemindersPlugin: Plugin = async (ctx) => { const storedReminders = await listReminders(ctx) - for (const reminder of storedReminders) { + for (const snapshot of storedReminders) { + const parsed = ReminderSchema.safeParse(snapshot) + // Persisted filenames are derived from IDs. Only generated UUID snapshots with safe + // timestamps may be re-opened or cleaned; malformed files are left for manual repair. + if (!parsed.success || !isSafeRestoredReminder(parsed.data)) { + logger.error(`[RemindersPlugin] Invalid restored reminder left untouched`) + invalidCount++ + continue + } + const reminder = parsed.data try { - ReminderSchema.parse(reminder) - // Skip session validation during startup - it may not be ready yet // Session cleanup will happen via event hook when session is actually deleted if (reminder.time.nextExecution + gracePeriod < now) { - logger.info(`[RemindersPlugin] Reminder ${reminder.id} expired, removing`) - await deleteReminder(reminder.id, ctx) - expiredCount++ + if (await cleanupReminderSnapshot(reminder, ctx, state, config)) { + logger.info(`[RemindersPlugin] Reminder ${reminder.id} expired, removing`) + expiredCount++ + } continue } state.reminders.set(reminder.id, reminder) - await scheduleTimer(reminder, ctx, state, config) + await scheduleTimer(reminder, ctx, state, config, { skipOverdue: true }) // Validate timer was actually created (timer health validation) const isHealthy = state.timers.has(reminder.id) @@ -64,16 +72,14 @@ const RemindersPlugin: Plugin = async (ctx) => { healthyCount++ logger.info(`[RemindersPlugin] Restored and validated reminder ${reminder.id}`) } else { - await deleteReminder(reminder.id, ctx) + await cleanupReminderSnapshot(reminder, ctx, state, config) state.reminders.delete(reminder.id) invalidCount++ logger.error(`[RemindersPlugin] Timer restoration failed for ${reminder.id}, cancelled reminder`) } } catch (error) { logger.error(`[RemindersPlugin] Failed to restore reminder:`, error) - if (reminder.id) { - await deleteReminder(reminder.id, ctx) - } + await cleanupReminderSnapshot(reminder, ctx, state, config) invalidCount++ } } @@ -121,4 +127,10 @@ const RemindersPlugin: Plugin = async (ctx) => { } } +function isSafeRestoredReminder(reminder: { id: string; time: { nextExecution: number; created: number } }): boolean { + return /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i.test(reminder.id) + && Number.isSafeInteger(reminder.time.created) + && Number.isSafeInteger(reminder.time.nextExecution) +} + export default RemindersPlugin diff --git a/leases.ts b/leases.ts new file mode 100644 index 0000000..9bad488 --- /dev/null +++ b/leases.ts @@ -0,0 +1,175 @@ +import { link, open, readFile, rm } from "node:fs/promises" +import path from "node:path" +import type { PluginInput } from "@opencode-ai/plugin" +import { getStorageDir } from "./storage" + +type Owner = { + pid: number + marker: string + start?: string + token: string +} + +export type ExecutionLease = { release: () => Promise } +export type LeaseAcquisition = + | { status: "acquired"; lease: ExecutionLease } + | { status: "contended" } + | { status: "error"; error: unknown } + +const processMarker = crypto.randomUUID() +const MUTATION_LOCK_TIMEOUT_MS = 30_000 +const MUTATION_LOCK_RETRY_MS = 25 + +function ownerText(owner: Owner): string { + return `pid=${owner.pid}\nmarker=${owner.marker}\n${owner.start ? `start=${owner.start}\n` : ""}token=${owner.token}\n` +} + +function parseOwner(contents: string): Owner | null { + const values = new Map( + contents.split("\n").flatMap((line) => { + const index = line.indexOf("=") + return index > 0 ? [[line.slice(0, index), line.slice(index + 1)]] : [] + }), + ) + const pid = Number(values.get("pid")) + const marker = values.get("marker") + const token = values.get("token") + const start = values.get("start") + if (!Number.isSafeInteger(pid) || pid <= 0 || !marker || !token) return null + return { pid, marker, token, ...(start ? { start } : {}) } +} + +async function linuxProcessStart(pid: number): Promise { + try { + const stat = await readFile(`/proc/${pid}/stat`, "utf8") + const fields = stat.slice(stat.lastIndexOf(")") + 2).trim().split(/\s+/) + return fields[19] ?? null + } catch { + return null + } +} + +async function ownerIsDemonstrablyDead(owner: Owner): Promise { + if (process.platform === "linux") { + if (!owner.start) return false + const currentStart = await linuxProcessStart(owner.pid) + if (currentStart !== null) return currentStart !== owner.start + } + + // A missing or unreadable /proc entry is not by itself proof of death: it may + // reflect permissions, descriptor exhaustion, or a transient I/O failure. + // kill(pid, 0) lets us reclaim only when the OS specifically reports ESRCH. + try { + process.kill(owner.pid, 0) + return false + } catch (error: any) { + return error?.code === "ESRCH" + } +} + +async function createOwnerFile(file: string, owner: Owner): Promise { + const candidate = `${file}.${owner.token}.candidate` + let handle + try { + handle = await open(candidate, "wx", 0o600) + await handle.writeFile(ownerText(owner)) + await handle.close() + handle = undefined + await link(candidate, file) + return true + } catch (error: any) { + if (error?.code === "EEXIST") return false + throw error + } finally { + await handle?.close() + await rm(candidate, { force: true }) + } +} + +async function readOwner(file: string): Promise { + try { + return parseOwner(await readFile(file, "utf8")) + } catch { + return null + } +} + +async function recoverDeadOwnerFile(file: string): Promise { + const owner = await readOwner(file) + if (!owner || !(await ownerIsDemonstrablyDead(owner))) return false + await rm(file, { force: true }) + return true +} + +async function removeStaleLease(lease: string, owner: Owner): Promise { + const recovery = `${lease}.recover` + // Never reclaim a recovery guard. A crashed recoverer can require manual cleanup, + // but reclaiming it has a check/unlink race that could delete a live replacement guard. + if (!(await createOwnerFile(recovery, owner))) return false + + try { + return await recoverDeadOwnerFile(lease) + } finally { + const current = await readOwner(recovery) + if (current?.token === owner.token) await rm(recovery, { force: true }) + } +} + +async function acquireOwnerLock(file: string, owner: Owner): Promise { + for (let attempt = 0; attempt < 2; attempt++) { + if (await createOwnerFile(file, owner)) { + let released = false + return { + status: "acquired", + lease: { + async release() { + if (released) return + released = true + const current = await readOwner(file) + if (current?.token === owner.token) await rm(file, { force: true }) + }, + }, + } + } + if (attempt === 0 && await removeStaleLease(file, owner)) continue + return { status: "contended" } + } + return { status: "contended" } +} + +async function newOwner(): Promise { + const start = process.platform === "linux" ? await linuxProcessStart(process.pid) : undefined + return { pid: process.pid, marker: processMarker, token: crypto.randomUUID(), ...(start ? { start } : {}) } +} + +export async function acquireExecutionLease( + reminderID: string, + occurrence: number, + ctx: PluginInput, +): Promise { + try { + const dir = await getStorageDir(ctx) + const leasePath = path.join(dir, `${reminderID}.${occurrence}.lease`) + return await acquireOwnerLock(leasePath, await newOwner()) + } catch (error) { + return { status: "error", error } + } +} + +export async function acquireMutationLock(reminderID: string, ctx: PluginInput): Promise { + const dir = await getStorageDir(ctx) + const lockPath = path.join(dir, `${reminderID}.mutation.lock`) + const owner = await newOwner() + const deadline = Date.now() + MUTATION_LOCK_TIMEOUT_MS + + while (Date.now() < deadline) { + const acquisition = await acquireOwnerLock(lockPath, owner) + if (acquisition.status === "acquired") return acquisition.lease + if (acquisition.status === "error") throw acquisition.error + + // Mutations are short and never encompass prompt execution. Waiting makes cancellation + // serializable with an in-flight transition rather than silently abandoning it. + await new Promise((resolve) => setTimeout(resolve, Math.min(MUTATION_LOCK_RETRY_MS, deadline - Date.now()))) + } + throw new Error(`Timed out acquiring mutation lock for reminder ${reminderID}`) +} diff --git a/package.json b/package.json index 5f9a2d6..df8deb1 100644 --- a/package.json +++ b/package.json @@ -23,6 +23,7 @@ "types.ts", "storage.ts", "scheduler.ts", + "leases.ts", "logger.ts", "tools/", "README.md" diff --git a/scheduler.ts b/scheduler.ts index 94076b8..26f2101 100644 --- a/scheduler.ts +++ b/scheduler.ts @@ -1,147 +1,213 @@ import type { PluginInput } from "@opencode-ai/plugin" import type { Reminder, State, PluginConfig } from "./types" -import { saveReminder, deleteReminder } from "./storage" +import { saveReminder, deleteReminder, loadReminder } from "./storage" +import { acquireExecutionLease, acquireMutationLock } from "./leases" import { logger } from "./logger" -export async function scheduleTimer( - reminder: Reminder, - ctx: PluginInput, - state: State, - config: PluginConfig, -): Promise { - const existingTimer = state.timers.get(reminder.id) - if (existingTimer) { - clearTimeout(existingTimer) - } +const RECONCILIATION_DELAY_MS = 250 +type ScheduleOptions = { skipOverdue?: boolean } +export async function scheduleTimer(reminder: Reminder, ctx: PluginInput, state: State, config: PluginConfig, options: ScheduleOptions = {}): Promise { + const existing = state.timers.get(reminder.id) + if (existing) clearTimeout(existing) const now = Date.now() - let delay = reminder.time.nextExecution - now - - // Handle missed execution windows for recurring reminders - if (delay < 0 && reminder.type === "recurring") { - // Calculate how many intervals were missed - const missedIntervals = Math.ceil(Math.abs(delay) / reminder.interval) - // Schedule for next interval from the original scheduled time to maintain cadence - reminder.time.nextExecution = reminder.time.nextExecution + (missedIntervals * reminder.interval) - delay = reminder.time.nextExecution - now - await saveReminder(reminder, ctx) - logger.info( - `[RemindersPlugin] Skipped ${missedIntervals} missed execution(s) for recurring reminder ${reminder.id}`, - ) - } else { - // For one-time reminders or on-time recurring, use the scheduled time - delay = Math.max(0, delay) - } - + const delay = Math.max(0, reminder.time.nextExecution - now) + const skipOverdue = options.skipOverdue === true && reminder.time.nextExecution < now const timer = setTimeout(async () => { state.timers.delete(reminder.id) - await executeReminder(reminder, ctx, state, config) + try { + await executeReminder(reminder, ctx, state, config, skipOverdue) + } catch (error) { + logger.error(`Unhandled reminder timer failure ${reminder.id}:`, error) + scheduleReconciliation(reminder.id, ctx, state, config) + } }, delay) + timer.unref() + state.timers.set(reminder.id, timer) +} - // CRITICAL: Allow process to exit even with active timers +function scheduleReconciliation(id: string, ctx: PluginInput, state: State, config: PluginConfig): void { + const existing = state.timers.get(id) + if (existing) clearTimeout(existing) + const timer = setTimeout(async () => { + state.timers.delete(id) + await reconcileReminder(id, ctx, state, config) + }, RECONCILIATION_DELAY_MS) timer.unref() + state.timers.set(id, timer) +} - state.timers.set(reminder.id, timer) +async function withMutationLock(id: string, ctx: PluginInput, action: () => Promise): Promise { + const lock = await acquireMutationLock(id, ctx) + try { + return await action() + } finally { + await lock.release() + } +} - logger.info(`Scheduled reminder ${reminder.id} to execute in ${Math.round(delay / 1000)}s`) +async function deleteReminderLocked(id: string, ctx: PluginInput, state: State): Promise { + const timer = state.timers.get(id) + if (timer) clearTimeout(timer) + state.timers.delete(id) + state.reminders.delete(id) + await deleteReminder(id, ctx) } -export async function executeReminder( - reminder: Reminder, - ctx: PluginInput, - state: State, - config: PluginConfig, -): Promise { - logger.info(`Executing reminder ${reminder.id}: ${reminder.userDescription}`) +export async function reconcileReminder(id: string, ctx: PluginInput, state: State, config: PluginConfig): Promise { + try { + const stored = await loadReminder(id, ctx) + if (!stored || stored.status !== "active") { + const timer = state.timers.get(id) + if (timer) clearTimeout(timer) + state.timers.delete(id) + state.reminders.delete(id) + return + } + state.reminders.set(id, stored) + await scheduleTimer(stored, ctx, state, config) + } catch (error) { + logger.error(`Failed to reconcile reminder ${id}:`, error) + // Preserve in-memory state and retry an I/O/parse failure rather than consuming it. + scheduleReconciliation(id, ctx, state, config) + } +} +export async function cleanupReminderSnapshot(snapshot: Reminder, ctx: PluginInput, state: State, config: PluginConfig): Promise { try { - await ctx.client.session.prompt({ - path: { id: reminder.sessionID }, - body: { - parts: [ - { - type: "text", - text: reminder.originalPrompt, - }, - ], - }, + return await withMutationLock(snapshot.id, ctx, async () => { + const current = await loadReminder(snapshot.id, ctx) + if (!current || current.time.nextExecution !== snapshot.time.nextExecution) { + await reconcileReminder(snapshot.id, ctx, state, config) + return false + } + await deleteReminderLocked(snapshot.id, ctx, state) + return true }) + } catch (error) { + logger.error(`Failed conditional cleanup for reminder ${snapshot.id}:`, error) + await reconcileReminder(snapshot.id, ctx, state, config) + return false + } +} - reminder.time.lastExecution = Date.now() +async function skipMissedRecurring(reminder: Reminder, occurrence: number, ctx: PluginInput, state: State, config: PluginConfig, skipOverdue: boolean): Promise { + if (!skipOverdue || reminder.type !== "recurring") return false + let advanced = false + await withMutationLock(reminder.id, ctx, async () => { + const current = await loadReminder(reminder.id, ctx) + if (!current || current.status !== "active" || current.time.nextExecution !== occurrence) return + const missed = Math.ceil(Math.abs(Date.now() - occurrence) / current.interval) + current.time.nextExecution = occurrence + missed * current.interval + state.reminders.set(current.id, current) + await saveReminder(current, ctx) + await scheduleTimer(current, ctx, state, config) + advanced = true + }) + if (!advanced) await reconcileReminder(reminder.id, ctx, state, config) + return true +} - if (reminder.type === "recurring") { - reminder.time.nextExecution = Date.now() + reminder.interval - state.reminders.set(reminder.id, reminder) - await saveReminder(reminder, ctx) - await scheduleTimer(reminder, ctx, state, config) - logger.info(`Recurring reminder ${reminder.id} rescheduled`) - - if (config.notifications.enabled) { - await ctx.client.tui.showToast({ - body: { - message: `Reminder executed: ${reminder.userDescription}`, - variant: "success", - }, - }) +async function handlePromptError(reminder: Reminder, occurrence: number, error: any, ctx: PluginInput, state: State, config: PluginConfig): Promise { + try { + let changed = false + await withMutationLock(reminder.id, ctx, async () => { + const current = await loadReminder(reminder.id, ctx) + if (!current || current.status !== "active" || current.time.nextExecution !== occurrence) return + changed = true + if (error?.name === "MessageAbortedError" && current.type === "recurring") { + current.time.nextExecution = Date.now() + current.interval + state.reminders.set(current.id, current) + await saveReminder(current, ctx) + await scheduleTimer(current, ctx, state, config) + } else { + await deleteReminderLocked(current.id, ctx, state) } - } else { - await cancelReminder(reminder.id, ctx, state) - logger.info(`One-time reminder ${reminder.id} completed and removed`) + }) + if (!changed) await reconcileReminder(reminder.id, ctx, state, config) + } catch (mutationError) { + logger.error(`Failed to handle reminder prompt error ${reminder.id}:`, mutationError) + await reconcileReminder(reminder.id, ctx, state, config) + } +} - if (config.notifications.enabled) { - await ctx.client.tui.showToast({ - body: { - message: `Reminder completed: ${reminder.userDescription}`, - variant: "success", - }, - }) +export async function executeReminder(reminder: Reminder, ctx: PluginInput, state: State, config: PluginConfig, skipOverdue = false): Promise { + const occurrence = reminder.time.nextExecution + const acquisition = await acquireExecutionLease(reminder.id, occurrence, ctx) + if (acquisition.status !== "acquired") { + if (acquisition.status === "error") logger.error(`Could not acquire reminder lease ${reminder.id}:`, acquisition.error) + scheduleReconciliation(reminder.id, ctx, state, config) + return + } + + try { + // Pre-prompt read/validation/skip failures only reconcile; they never delete or prompt. + try { + const stored = await loadReminder(reminder.id, ctx) + if (!stored || stored.status !== "active" || stored.time.nextExecution !== occurrence) { + await reconcileReminder(reminder.id, ctx, state, config) + return } + if (await skipMissedRecurring(stored, occurrence, ctx, state, config, skipOverdue)) return + } catch (error) { + logger.error(`Pre-prompt validation failed for reminder ${reminder.id}:`, error) + await reconcileReminder(reminder.id, ctx, state, config) + return } - } catch (error: any) { - logger.error(`Reminder ${reminder.id} execution failed:`, error) - if (config.notifications.enabled) { - try { - await ctx.client.tui.showToast({ - body: { - title: "Reminder Failed", - message: reminder.userDescription, - variant: "error", - }, - }) - } catch (notificationError) { - logger.error(`Failed to show error notification for reminder ${reminder.id}:`, notificationError) - } + try { + await ctx.client.session.prompt({ path: { id: reminder.sessionID }, body: { parts: [{ type: "text", text: reminder.originalPrompt }] } }) + } catch (error: any) { + logger.error(`Reminder ${reminder.id} prompt failed:`, error) + await handlePromptError(reminder, occurrence, error, ctx, state, config) + return } - if (error?.name === "MessageAbortedError") { - if (reminder.type === "recurring") { - reminder.time.nextExecution = Date.now() + reminder.interval - state.reminders.set(reminder.id, reminder) - await saveReminder(reminder, ctx) - await scheduleTimer(reminder, ctx, state, config) - logger.info(`Recurring reminder ${reminder.id} rescheduled after abort`) - } else { - await cancelReminder(reminder.id, ctx, state) - logger.info(`One-time reminder ${reminder.id} cancelled after abort`) + // Prompt acceptance is separate from its durable transition. Never treat a failure here as a prompt error. + try { + const completed = await withMutationLock(reminder.id, ctx, async (): Promise => { + const current = await loadReminder(reminder.id, ctx) + if (!current || current.status !== "active" || current.time.nextExecution !== occurrence) return null + current.time.lastExecution = Date.now() + reminder.time.lastExecution = current.time.lastExecution + if (current.type === "recurring") { + current.time.nextExecution = Date.now() + current.interval + state.reminders.set(current.id, current) + await saveReminder(current, ctx) + await scheduleTimer(current, ctx, state, config) + } else { + await deleteReminderLocked(current.id, ctx, state) + } + return current + }) + if (!completed) await reconcileReminder(reminder.id, ctx, state, config) + else if (config.notifications.enabled) { + try { + await ctx.client.tui.showToast({ body: { message: completed.type === "recurring" ? `Reminder executed: ${completed.userDescription}` : `Reminder completed: ${completed.userDescription}`, variant: "success" } }) + } catch (error) { + logger.error(`Failed reminder toast ${reminder.id}:`, error) + } } - } else { - await cancelReminder(reminder.id, ctx, state) - logger.info(`Reminder ${reminder.id} cancelled due to error`) + } catch (error) { + logger.error(`Post-prompt transition failed for reminder ${reminder.id}:`, error) + await reconcileReminder(reminder.id, ctx, state, config) + } + } finally { + try { + await acquisition.lease.release() + } catch (error) { + logger.error(`Failed to release reminder lease ${reminder.id}:`, error) + scheduleReconciliation(reminder.id, ctx, state, config) } } } export async function cancelReminder(id: string, ctx: PluginInput, state: State): Promise { - const timer = state.timers.get(id) - if (timer) { - clearTimeout(timer) - state.timers.delete(id) + try { + await withMutationLock(id, ctx, async () => deleteReminderLocked(id, ctx, state)) + logger.info(`Reminder ${id} cancelled`) + } catch (error) { + logger.error(`Failed to cancel reminder ${id}:`, error) + throw error } - - state.reminders.delete(id) - - await deleteReminder(id, ctx) - - logger.info(`Reminder ${id} cancelled`) } diff --git a/storage.ts b/storage.ts index f0754ff..e0662d8 100644 --- a/storage.ts +++ b/storage.ts @@ -1,4 +1,5 @@ import path from "path" +import { open, readFile, rename, rm } from "node:fs/promises" import type { PluginInput } from "@opencode-ai/plugin" import type { Reminder } from "./types" import { logger } from "./logger" @@ -13,7 +14,19 @@ export async function getStorageDir(ctx: PluginInput): Promise { export async function saveReminder(reminder: Reminder, ctx: PluginInput): Promise { const dir = await getStorageDir(ctx) const filePath = path.join(dir, `${reminder.id}.json`) - await Bun.write(filePath, JSON.stringify(reminder, null, 2)) + const temporaryPath = path.join(dir, `.${reminder.id}.${crypto.randomUUID()}.tmp`) + let handle + + try { + handle = await open(temporaryPath, "wx", 0o600) + await handle.writeFile(JSON.stringify(reminder, null, 2)) + await handle.close() + handle = undefined + await rename(temporaryPath, filePath) + } finally { + await handle?.close() + await rm(temporaryPath, { force: true }) + } } export async function loadReminder(id: string, ctx: PluginInput): Promise { @@ -21,9 +34,10 @@ export async function loadReminder(id: string, ctx: PluginInput): Promise { - return { - client: { session: { prompt, get, list } }, - project: { id, worktree, time }, - directory: tmpDir, - worktree: tmpDir, - $: Bun.$, - } -} -``` - -### Cleanup Pattern -```typescript -beforeEach(async () => { - tmpDir = await $`mktemp -d`.text().then((t) => t.trim()) - ctx = await createMockContext(tmpDir) -}) - -afterEach(async () => { - for (const timer of state.timers.values()) { - clearTimeout(timer) - } - await $`rm -rf ${tmpDir}`.quiet() -}) -``` - -## Success Criteria - -- [ ] All type errors resolved -- [ ] 100% test passing rate -- [ ] No crashes or hangs -- [ ] Coverage matches original (460+ tests) -- [ ] Tests run in <1 second -- [ ] No flaky tests +The suite intentionally does not claim exact-once prompt delivery across a crash or storage failure +after prompt acceptance. It tests the plugin's same-host coordination and reconciliation behavior. diff --git a/test/fixtures/lease-child.ts b/test/fixtures/lease-child.ts new file mode 100644 index 0000000..b569991 --- /dev/null +++ b/test/fixtures/lease-child.ts @@ -0,0 +1,21 @@ +import { $ } from "bun" +import type { PluginInput } from "@opencode-ai/plugin" +import { acquireExecutionLease } from "../../leases" + +const [directory, reminderID, occurrence] = process.argv.slice(2) +if (!directory || !reminderID || !occurrence) throw new Error("missing lease child arguments") + +const ctx: PluginInput = { + client: {} as any, + project: { id: "lease-project", worktree: directory, time: { created: Date.now() } }, + directory, + worktree: directory, + $, +} +const lease = await acquireExecutionLease(reminderID, Number(occurrence), ctx) +process.stdout.write(lease.status === "acquired" ? "winner\n" : "loser\n") + +if (lease.status === "acquired") { + await new Response(Bun.stdin.stream()).text() + await lease.lease.release() +} diff --git a/test/fixtures/scheduler-child.ts b/test/fixtures/scheduler-child.ts new file mode 100644 index 0000000..42403eb --- /dev/null +++ b/test/fixtures/scheduler-child.ts @@ -0,0 +1,37 @@ +import { appendFile } from "node:fs/promises" +import { $ } from "bun" +import type { PluginInput } from "@opencode-ai/plugin" +import { executeReminder } from "../../scheduler" +import type { PluginConfig, Reminder, State } from "../../types" + +const [directory, reminderPath, recorder] = process.argv.slice(2) +if (!directory || !reminderPath || !recorder) throw new Error("missing scheduler child arguments") + +const reminder = await Bun.file(reminderPath).json() as Reminder +const ctx: PluginInput = { + client: { + session: { + prompt: async () => { + await appendFile(recorder, "prompt\n") + process.stdout.write("prompted\n") + await new Response(Bun.stdin.stream()).text() + return { data: {} as any, error: undefined, response: {} as any } + }, + }, + tui: { showToast: async () => {} }, + } as any, + project: { id: "lease-project", worktree: directory, time: { created: Date.now() } }, + directory, + worktree: directory, + $, +} +const state: State = { reminders: new Map([[reminder.id, reminder]]), timers: new Map(), projectID: "lease-project" } +const config: PluginConfig = { + enabled: true, + max_reminders_per_project: 50, + min_interval_seconds: 30, + notifications: { enabled: false }, +} + +await executeReminder(reminder, ctx, state, config) +process.stdout.write("finished\n") diff --git a/test/leases.test.ts b/test/leases.test.ts new file mode 100644 index 0000000..275eec6 --- /dev/null +++ b/test/leases.test.ts @@ -0,0 +1,169 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test" +import { $ } from "bun" +import path from "node:path" +import type { PluginInput } from "@opencode-ai/plugin" +import { acquireExecutionLease } from "../leases" +import { getStorageDir, saveReminder } from "../storage" +import type { Reminder } from "../types" + +async function context(directory: string): Promise { + return { + client: {} as any, + project: { id: "lease-project", worktree: directory, time: { created: Date.now() } }, + directory, + worktree: directory, + $, + } +} + +async function firstLine(stream: ReadableStream): Promise { + const reader = stream.getReader() + const decoder = new TextDecoder() + let output = "" + try { + while (true) { + const { done, value } = await reader.read() + if (done) return output.trim() + output += decoder.decode(value, { stream: true }) + const newline = output.indexOf("\n") + if (newline >= 0) return output.slice(0, newline) + } + } finally { + reader.releaseLock() + } +} + +async function within(promise: Promise, label: string, timeout = 5000): Promise { + let timer: ReturnType | undefined + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { timer = setTimeout(() => reject(new Error(`Timed out waiting for ${label}`)), timeout) }), + ]) + } finally { + if (timer) clearTimeout(timer) + } +} + +async function stopChildren(children: Bun.Subprocess[]): Promise { + for (const child of children) { + try { (child.stdin as any)?.end() } catch {} + try { child.kill() } catch {} + } + await within(Promise.all(children.map((child) => child.exited)), "child process cleanup", 1000).catch(() => {}) +} + +describe("execution leases", () => { + let directory: string + let ctx: PluginInput + + beforeEach(async () => { + directory = (await $`mktemp -d`.text()).trim() + ctx = await context(directory) + }) + afterEach(async () => { await $`rm -rf ${directory}`.quiet() }) + + test("concurrent acquisition has exactly one winner", async () => { + const attempts = await Promise.all(Array.from({ length: 20 }, () => acquireExecutionLease("rem-lease", 123, ctx))) + const winners = attempts.filter((lease) => lease.status === "acquired") + expect(winners).toHaveLength(1) + if (winners[0]?.status === "acquired") await winners[0].lease.release() + }) + + test("independent child processes have exactly one winner", async () => { + const fixture = path.join(import.meta.dir, "fixtures", "lease-child.ts") + const children = Array.from({ length: 2 }, () => Bun.spawn({ + cmd: [process.execPath, fixture, directory, "rem-child", "234"], + stdin: "pipe", + stdout: "pipe", + stderr: "inherit", + })) + try { + const results = await within(Promise.all(children.map((child) => firstLine(child.stdout))), "child ownership result") + expect(results.filter((result) => result === "winner")).toHaveLength(1) + expect(results.filter((result) => result === "loser")).toHaveLength(1) + children[results.indexOf("winner")]!.stdin.end() + await within(Promise.all(children.map((child) => child.exited)), "child process exit") + } finally { + await stopChildren(children) + } + }) + + test("independent schedulers submit one prompt for a persisted occurrence", async () => { + const reminder: Reminder = { + id: "rem-multiprocess-scheduler", + sessionID: "ses-child", + projectID: "lease-project", + type: "one-time", + interval: 1000, + originalPrompt: "child prompt", + userDescription: "child scheduler", + time: { created: Date.now(), nextExecution: Date.now() + 1000 }, + status: "active", + } + await saveReminder(reminder, ctx) + const dir = await getStorageDir(ctx) + const reminderPath = `${dir}/${reminder.id}.json` + const recorder = `${directory}/prompts.txt` + const fixture = path.join(import.meta.dir, "fixtures", "scheduler-child.ts") + const children = Array.from({ length: 2 }, () => Bun.spawn({ + cmd: [process.execPath, fixture, directory, reminderPath, recorder], + stdin: "pipe", + stdout: "pipe", + stderr: "inherit", + })) + try { + const results = await within(Promise.all(children.map((child) => firstLine(child.stdout))), "scheduler child result") + expect(results.filter((result) => result === "prompted")).toHaveLength(1) + expect(results.filter((result) => result === "finished")).toHaveLength(1) + expect((await Bun.file(recorder).text()).trim().split("\n")).toHaveLength(1) + children[results.indexOf("prompted")]!.stdin.end() + await within(Promise.all(children.map((child) => child.exited)), "scheduler child exit") + } finally { + await stopChildren(children) + } + }) + + test("recovers a lease whose PID start identity was reused", async () => { + const dir = await getStorageDir(ctx) + await Bun.write(`${dir}/rem-dead.456.lease`, `pid=${process.pid}\nmarker=dead\nstart=not-this-process\ntoken=dead\n`) + + const lease = await acquireExecutionLease("rem-dead", 456, ctx) + expect(lease.status).toBe("acquired") + if (lease.status === "acquired") await lease.lease.release() + }) + + test("does not steal a lease held by this live process", async () => { + const first = await acquireExecutionLease("rem-live", 789, ctx) + const second = await acquireExecutionLease("rem-live", 789, ctx) + expect(first.status).toBe("acquired") + expect(second.status).toBe("contended") + if (first.status === "acquired") await first.lease.release() + }) + + test("does not steal an unreadable partial lease", async () => { + const dir = await getStorageDir(ctx) + await Bun.write(`${dir}/rem-partial.321.lease`, "pid=") + + expect((await acquireExecutionLease("rem-partial", 321, ctx)).status).toBe("contended") + }) + + test("ignores an abandoned unpublished candidate", async () => { + const dir = await getStorageDir(ctx) + await Bun.write(`${dir}/rem-candidate.654.lease.abandoned.candidate`, "partial") + + const lease = await acquireExecutionLease("rem-candidate", 654, ctx) + expect(lease.status).toBe("acquired") + if (lease.status === "acquired") await lease.lease.release() + }) + + test("does not reclaim a stale recovery guard", async () => { + const dir = await getStorageDir(ctx) + const owner = `pid=${process.pid}\nmarker=dead\nstart=not-this-process\ntoken=dead\n` + await Bun.write(`${dir}/rem-guard.987.lease`, owner) + await Bun.write(`${dir}/rem-guard.987.lease.recover`, owner) + + const lease = await acquireExecutionLease("rem-guard", 987, ctx) + expect(lease.status).toBe("contended") + }) +}) diff --git a/test/scheduler.test.ts b/test/scheduler.test.ts index 0d8533a..9cc694a 100644 --- a/test/scheduler.test.ts +++ b/test/scheduler.test.ts @@ -1,8 +1,10 @@ import { test, expect, describe, beforeEach, afterEach } from "bun:test" -import { scheduleTimer, executeReminder, cancelReminder } from "../scheduler" +import { scheduleTimer, executeReminder, cancelReminder, reconcileReminder, cleanupReminderSnapshot } from "../scheduler" import type { Reminder, State, PluginConfig } from "../types" import type { PluginInput } from "@opencode-ai/plugin" import { $ } from "bun" +import { saveReminder, loadReminder, deleteReminder, getStorageDir } from "../storage" +import { acquireExecutionLease, acquireMutationLock } from "../leases" async function createMockContext(tmpDir: string): Promise { return { @@ -174,6 +176,7 @@ describe("Scheduler", () => { } state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) await executeReminder(reminder, ctx, state, config) await new Promise((resolve) => setTimeout(resolve, 50)) @@ -199,6 +202,7 @@ describe("Scheduler", () => { } state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) const before = Date.now() await executeReminder(reminder, ctx, state, config) const after = Date.now() @@ -224,6 +228,7 @@ describe("Scheduler", () => { } state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) await executeReminder(reminder, ctx, state, config) await new Promise((resolve) => setTimeout(resolve, 50)) @@ -249,6 +254,7 @@ describe("Scheduler", () => { const oldNextExecution = reminder.time.nextExecution state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) await executeReminder(reminder, ctx, state, config) await new Promise((resolve) => setTimeout(resolve, 50)) @@ -259,7 +265,177 @@ describe("Scheduler", () => { expect(state.timers.has(reminder.id)).toBe(true) }) - test("scheduleTimer maintains cadence for missed recurring reminders", async () => { + test("losing execution contender does not prompt or mutate persistence", async () => { + const reminder: Reminder = { + id: "rem-cross-process-owner", + sessionID: "ses-test", + projectID: "test-project-123", + type: "one-time", + interval: 100, + originalPrompt: "only once", + userDescription: "Lease ownership test", + time: { created: Date.now(), nextExecution: Date.now() + 100 }, + status: "active", + } + const otherState: State = { reminders: new Map([[reminder.id, { ...reminder, time: { ...reminder.time } }]]), timers: new Map(), projectID: state.projectID } + let firstPrompted!: () => void + const promptStarted = new Promise((resolve) => { firstPrompted = resolve }) + let finishPrompt!: () => void + const promptFinished = new Promise((resolve) => { finishPrompt = resolve }) + let prompts = 0 + ;(ctx.client.session.prompt as any) = async () => { + prompts++ + firstPrompted() + await promptFinished + return { data: {} as any, error: undefined, response: {} as any } + } + + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + const winner = executeReminder(reminder, ctx, state, config) + await promptStarted + await executeReminder(otherState.reminders.get(reminder.id)!, ctx, otherState, config) + + expect(prompts).toBe(1) + expect(await loadReminder(reminder.id, ctx)).toEqual(reminder) + expect(otherState.reminders.has(reminder.id)).toBe(true) + finishPrompt() + await winner + }) + + test("recurring loser reconciles to the winner's future occurrence", async () => { + const reminder: Reminder = { + id: "rem-recurring-contended", + sessionID: "ses-test", + projectID: "test-project-123", + type: "recurring", + interval: 60000, + originalPrompt: "only once now", + userDescription: "Recurring lease reconciliation", + time: { created: Date.now(), nextExecution: Date.now() + 100 }, + status: "active", + } + const loserState: State = { + reminders: new Map([[reminder.id, { ...reminder, time: { ...reminder.time } }]]), + timers: new Map(), + projectID: state.projectID, + } + let started!: () => void + const promptStarted = new Promise((resolve) => { started = resolve }) + let finish!: () => void + const promptFinished = new Promise((resolve) => { finish = resolve }) + ;(ctx.client.session.prompt as any) = async () => { + started() + await promptFinished + return { data: {} as any, error: undefined, response: {} as any } + } + + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + const winner = executeReminder(reminder, ctx, state, config) + await promptStarted + await executeReminder(loserState.reminders.get(reminder.id)!, ctx, loserState, config) + expect(loserState.timers.has(reminder.id)).toBe(true) + + finish() + await winner + const persisted = await loadReminder(reminder.id, ctx) + expect(persisted).not.toBeNull() + // The loser keeps its unref'd retry; wait only after the winner's completion barrier. + await new Promise((resolve) => setTimeout(resolve, 500)) + expect(loserState.reminders.get(reminder.id)?.time.nextExecution).toBe(persisted!.time.nextExecution) + expect(loserState.timers.has(reminder.id)).toBe(true) + }) + + test("scheduler recovers a crashed owner's recurring occurrence", async () => { + const reminder: Reminder = { + id: "rem-dead-owner", + sessionID: "ses-test", + projectID: "test-project-123", + type: "recurring", + interval: 60000, + originalPrompt: "recover", + userDescription: "Dead owner recovery", + time: { created: Date.now(), nextExecution: Date.now() + 100 }, + status: "active", + } + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + const dir = await getStorageDir(ctx) + await Bun.write(`${dir}/${reminder.id}.${reminder.time.nextExecution}.lease`, `pid=${process.pid}\nmarker=dead\nstart=not-this-process\ntoken=dead\n`) + let prompts = 0 + ;(ctx.client.session.prompt as any) = async () => { + prompts++ + return { data: {} as any, error: undefined, response: {} as any } + } + + await executeReminder(reminder, ctx, state, config) + expect(prompts).toBe(1) + expect((await loadReminder(reminder.id, ctx))!.time.nextExecution).toBeGreaterThan(reminder.time.nextExecution) + }) + + test("cancellation before executor mutation cannot be overwritten", async () => { + const reminder: Reminder = { + id: "rem-cancel-during-prompt", + sessionID: "ses-test", + projectID: "test-project-123", + type: "recurring", + interval: 60000, + originalPrompt: "cancelled", + userDescription: "Cancellation fence", + time: { created: Date.now(), nextExecution: Date.now() + 100 }, + status: "active", + } + let started!: () => void + const promptStarted = new Promise((resolve) => { started = resolve }) + let finish!: () => void + const promptFinished = new Promise((resolve) => { finish = resolve }) + ;(ctx.client.session.prompt as any) = async () => { + started() + await promptFinished + return { data: {} as any, error: undefined, response: {} as any } + } + + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + const execution = executeReminder(reminder, ctx, state, config) + await promptStarted + const lock = await acquireMutationLock(reminder.id, ctx) + const cancellation = cancelReminder(reminder.id, ctx, state) + await lock.release() + await cancellation + finish() + await execution + + expect(await loadReminder(reminder.id, ctx)).toBeNull() + expect(state.reminders.has(reminder.id)).toBe(false) + expect(state.timers.has(reminder.id)).toBe(false) + }) + + test("cancellation after executor mutation deletes the saved recurrence", async () => { + const reminder: Reminder = { + id: "rem-cancel-after-mutation", + sessionID: "ses-test", + projectID: "test-project-123", + type: "recurring", + interval: 60000, + originalPrompt: "completed", + userDescription: "Cancellation after mutation", + time: { created: Date.now(), nextExecution: Date.now() + 100 }, + status: "active", + } + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + await executeReminder(reminder, ctx, state, config) + expect(await loadReminder(reminder.id, ctx)).not.toBeNull() + + await cancelReminder(reminder.id, ctx, state) + expect(await loadReminder(reminder.id, ctx)).toBeNull() + expect(state.reminders.has(reminder.id)).toBe(false) + expect(state.timers.has(reminder.id)).toBe(false) + }) + + test("execution owner skips missed recurring reminders with upstream cadence", async () => { const interval = 3600000 // 1 hour const originalScheduledTime = Date.now() - 5400000 // 1.5 hours ago @@ -279,21 +455,173 @@ describe("Scheduler", () => { } state.reminders.set(reminder.id, reminder) - await scheduleTimer(reminder, ctx, state, config) + await saveReminder(reminder, ctx) + let prompts = 0 + ;(ctx.client.session.prompt as any) = async () => { + prompts++ + return { data: {} as any, error: undefined, response: {} as any } + } + await executeReminder(reminder, ctx, state, config, true) - // Should schedule for next interval from the original time, not from now - // Original: now - 1.5h, interval: 1h, missed: 2 intervals - // Next should be: (now - 1.5h) + (2 * 1h) = now + 0.5h const expectedNext = originalScheduledTime + (2 * interval) - const actualNext = reminder.time.nextExecution - - // Allow 100ms tolerance for test execution time - expect(Math.abs(actualNext - expectedNext)).toBeLessThan(100) - - // Verify the delay is approximately 30 minutes (half an hour from now) - const now = Date.now() - const delay = actualNext - now - expect(delay).toBeGreaterThan(1700000) // ~28 minutes - expect(delay).toBeLessThan(1900000) // ~32 minutes + expect(Math.abs(state.reminders.get(reminder.id)!.time.nextExecution - expectedNext)).toBeLessThan(100) + expect(state.timers.has(reminder.id)).toBe(true) + expect(prompts).toBe(0) + }) + + test("normal late recurring execution prompts instead of skipping", async () => { + const reminder: Reminder = { + id: "rem-normal-late", + sessionID: "ses-test", + projectID: "test-project-123", + type: "recurring", + interval: 60000, + originalPrompt: "late but normal", + userDescription: "Normal late execution", + time: { created: Date.now() - 1000, nextExecution: Date.now() - 1 }, + status: "active", + } + let prompts = 0 + ;(ctx.client.session.prompt as any) = async () => { + prompts++ + return { data: {} as any, error: undefined, response: {} as any } + } + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + + await executeReminder(reminder, ctx, state, config) + expect(prompts).toBe(1) + expect((await loadReminder(reminder.id, ctx))!.time.nextExecution).toBeGreaterThan(Date.now()) + }) + + test("failover of a contended old occurrence prompts instead of skipping", async () => { + const reminder: Reminder = { + id: "rem-failover-old-occurrence", + sessionID: "ses-test", + projectID: "test-project-123", + type: "one-time", + interval: 1000, + originalPrompt: "recover old occurrence", + userDescription: "Old occurrence failover", + time: { created: Date.now() - 1000, nextExecution: Date.now() - 1 }, + status: "active", + } + let prompts = 0 + ;(ctx.client.session.prompt as any) = async () => { + prompts++ + return { data: {} as any, error: undefined, response: {} as any } + } + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + const holder = await acquireExecutionLease(reminder.id, reminder.time.nextExecution, ctx) + expect(holder.status).toBe("acquired") + + await executeReminder(reminder, ctx, state, config) + if (holder.status === "acquired") await holder.lease.release() + await new Promise((resolve) => setTimeout(resolve, 500)) + + expect(prompts).toBe(1) + expect(await loadReminder(reminder.id, ctx)).toBeNull() + }) + + test("pre-prompt parse failure preserves state and reconciles after repair", async () => { + const reminder: Reminder = { + id: "rem-read-retry", + sessionID: "ses-test", + projectID: "test-project-123", + type: "one-time", + interval: 1000, + originalPrompt: "must not prompt while corrupt", + userDescription: "Read retry", + time: { created: Date.now(), nextExecution: Date.now() + 1000 }, + status: "active", + } + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + const dir = await getStorageDir(ctx) + await Bun.write(`${dir}/${reminder.id}.json`, "{") + let prompts = 0 + ;(ctx.client.session.prompt as any) = async () => { prompts++; return { data: {} as any, error: undefined, response: {} as any } } + + await executeReminder(reminder, ctx, state, config) + expect(prompts).toBe(0) + expect(state.reminders.has(reminder.id)).toBe(true) + await saveReminder(reminder, ctx) + await new Promise((resolve) => setTimeout(resolve, 500)) + expect(state.timers.has(reminder.id)).toBe(true) + }) + + test("conditional startup cleanup cannot delete a newer occurrence", async () => { + const snapshot: Reminder = { + id: "rem-cleanup-snapshot", + sessionID: "ses-test", + projectID: "test-project-123", + type: "recurring", + interval: 1000, + originalPrompt: "newer wins", + userDescription: "Conditional cleanup", + time: { created: Date.now(), nextExecution: Date.now() - 1000 }, + status: "active", + } + const newer = { ...snapshot, time: { ...snapshot.time, nextExecution: Date.now() + 60000 } } + await saveReminder(newer, ctx) + expect(await cleanupReminderSnapshot(snapshot, ctx, state, config)).toBe(false) + expect((await loadReminder(snapshot.id, ctx))!.time.nextExecution).toBe(newer.time.nextExecution) + expect(state.timers.has(snapshot.id)).toBe(true) + }) + + test("toast failure after a successful transition does not cancel recurrence", async () => { + const reminder: Reminder = { + id: "rem-toast-failure", + sessionID: "ses-test", + projectID: "test-project-123", + type: "recurring", + interval: 60000, + originalPrompt: "toast failure", + userDescription: "Toast failure", + time: { created: Date.now(), nextExecution: Date.now() + 100 }, + status: "active", + } + ;(ctx.client.tui.showToast as any) = async () => { throw new Error("toast unavailable") } + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + await executeReminder(reminder, ctx, state, config) + expect((await loadReminder(reminder.id, ctx))!.time.nextExecution).toBeGreaterThan(Date.now()) + expect(state.timers.has(reminder.id)).toBe(true) + }) + + test("save failure after prompt acceptance reconciles without cancelling recurrence", async () => { + const reminder: Reminder = { + id: "rem-save-failure", + sessionID: "ses-test", + projectID: "test-project-123", + type: "recurring", + interval: 60000, + originalPrompt: "accepted before save fails", + userDescription: "Save failure", + time: { created: Date.now(), nextExecution: Date.now() + 100 }, + status: "active", + } + let started!: () => void + const promptStarted = new Promise((resolve) => { started = resolve }) + let finish!: () => void + const promptFinished = new Promise((resolve) => { finish = resolve }) + ;(ctx.client.session.prompt as any) = async () => { + started() + await promptFinished + return { data: {} as any, error: undefined, response: {} as any } + } + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + const execution = executeReminder(reminder, ctx, state, config) + await promptStarted + const dir = await getStorageDir(ctx) + await ctx.$`rm ${dir}/${reminder.id}.json`.quiet() + await ctx.$`mkdir ${dir}/${reminder.id}.json`.quiet() + finish() + await execution + + expect(state.reminders.has(reminder.id)).toBe(true) + expect(state.timers.has(reminder.id)).toBe(true) }) }) diff --git a/test/storage.test.ts b/test/storage.test.ts index 60e1218..149335d 100644 --- a/test/storage.test.ts +++ b/test/storage.test.ts @@ -67,6 +67,28 @@ describe("Storage", () => { expect(content).toEqual(reminder) }) + test("saveReminder atomically replaces JSON and removes its temporary file", async () => { + const reminder: Reminder = { + id: "rem-atomic", + sessionID: "ses-atomic", + projectID: "test-project-123", + type: "recurring", + interval: 5000, + originalPrompt: "atomic", + userDescription: "Atomic reminder", + time: { created: Date.now(), nextExecution: Date.now() + 5000 }, + status: "active", + } + await saveReminder(reminder, ctx) + const updated = { ...reminder, time: { ...reminder.time, nextExecution: reminder.time.nextExecution + 1 } } + await saveReminder(updated, ctx) + + const dir = await getStorageDir(ctx) + expect(await Bun.file(`${dir}/rem-atomic.json`).json()).toEqual(updated) + const temporaryFiles = await Array.fromAsync(new Bun.Glob("*.tmp").scan({ cwd: dir })) + expect(temporaryFiles).toHaveLength(0) + }) + test("loadReminder reads existing file", async () => { const reminder: Reminder = { id: "rem-load-test", @@ -94,6 +116,12 @@ describe("Storage", () => { expect(loaded).toBeNull() }) + test("loadReminder propagates malformed JSON instead of treating it as absent", async () => { + const dir = await getStorageDir(ctx) + await Bun.write(`${dir}/rem-malformed.json`, "{") + await expect(loadReminder("rem-malformed", ctx)).rejects.toThrow() + }) + test("deleteReminder removes file", async () => { const reminder: Reminder = { id: "rem-delete-test",