From 5e8c75401d7e5b58ae2543dd3b32e30aa6232637 Mon Sep 17 00:00:00 2001 From: Marc Warne Date: Thu, 23 Jul 2026 13:07:54 +0100 Subject: [PATCH 1/5] 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", From b3ce4d53b5016d1f865026d6a7be8a961d02963d Mon Sep 17 00:00:00 2001 From: Marc Warne Date: Sat, 25 Jul 2026 16:41:08 +0100 Subject: [PATCH 2/5] fix: integrate reminder lifecycle coordination --- README.md | 8 + index.ts | 30 +++- package.json | 2 +- scheduler.ts | 346 ++++++++++++++++++++++++++++++++------- storage.ts | 2 +- test/README.md | 7 +- test/integration.test.ts | 74 +++++++++ test/scheduler.test.ts | 126 +++++++++++++- test/types.test.ts | 17 ++ tools/reminderadd.ts | 7 +- tools/reminderremove.ts | 14 +- types.ts | 2 + 12 files changed, 557 insertions(+), 78 deletions(-) diff --git a/README.md b/README.md index b52bbe8..d2988b1 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,11 @@ Post-prompt recurrence updates and every plugin cancellation/restore cleanup sha per-reminder mutation lock, so the durable read-and-transition cannot race a deletion. The lock is not held while the prompt API runs. +Within one server process, plugin generations also fence hot reloads. Reinitialization clears old +timers, aborts active requests, waits for mutations already in progress, and prevents stale timer or +reconciliation callbacks from changing replacement state. Cancellation aborts the active local +request immediately, then uses the same mutation lock as cross-process transitions. + 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 @@ -93,6 +98,9 @@ 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. +Reminders capture the creating tool context's optional agent name and pass it back to the prompt API +when they execute. Existing reminder files without `agent` remain valid and omit the prompt field. + ## Data Storage Reminders stored in: diff --git a/index.ts b/index.ts index f81c9f0..243cb94 100644 --- a/index.ts +++ b/index.ts @@ -3,13 +3,21 @@ import { logger } from "./logger" import type { State, PluginConfig } from "./types" import { ReminderSchema } from "./types" import { getStorageDir, listReminders } from "./storage" -import { scheduleTimer, cancelReminder, cleanupReminderSnapshot } from "./scheduler" +import { + beginSchedulerGeneration, + isSchedulerGenerationCurrent, + waitForSchedulerMutations, + scheduleTimer, + cancelReminder, + cleanupReminderSnapshot, +} from "./scheduler" import { createReminderAddTool } from "./tools/reminderadd" import { createReminderListTool } from "./tools/reminderlist" import { createReminderRemoveTool } from "./tools/reminderremove" const RemindersPlugin: Plugin = async (ctx) => { const { project } = ctx + const generation = beginSchedulerGeneration(ctx) logger.info(`[RemindersPlugin] Initializing for project ${project.id}`) @@ -19,6 +27,7 @@ const RemindersPlugin: Plugin = async (ctx) => { reminders: new Map(), timers: new Map(), projectID: project.id, + generation, } // Configuration with defaults @@ -38,9 +47,12 @@ const RemindersPlugin: Plugin = async (ctx) => { let invalidCount = 0 let healthyCount = 0 + if (!(await waitForSchedulerMutations(ctx, generation))) return {} const storedReminders = await listReminders(ctx) + if (!isSchedulerGenerationCurrent(ctx, generation)) return {} for (const snapshot of storedReminders) { + if (!isSchedulerGenerationCurrent(ctx, generation)) return {} 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. @@ -59,14 +71,19 @@ const RemindersPlugin: Plugin = async (ctx) => { logger.info(`[RemindersPlugin] Reminder ${reminder.id} expired, removing`) expiredCount++ } + if (!isSchedulerGenerationCurrent(ctx, generation)) return {} continue } state.reminders.set(reminder.id, reminder) - await scheduleTimer(reminder, ctx, state, config, { skipOverdue: true }) + const scheduled = await scheduleTimer(reminder, ctx, state, config, { + persist: false, + skipOverdue: true, + }) + if (!isSchedulerGenerationCurrent(ctx, generation)) return {} // Validate timer was actually created (timer health validation) - const isHealthy = state.timers.has(reminder.id) + const isHealthy = scheduled && state.timers.has(reminder.id) if (isHealthy) { restoredCount++ healthyCount++ @@ -91,7 +108,7 @@ const RemindersPlugin: Plugin = async (ctx) => { // Cleanup considerations: // - timer.unref() allows clean exit without blocking the process // - Plugin API currently has no cleanup hook for graceful shutdown - // - On hot-reload, old timers may fire once but won't be rescheduled (state is in new instance) + // - A replacement generation clears timers and aborts active executions from the old instance // - Reminder state persists to storage and is restored on next startup // - Max reminders per project (50) bounds memory usage @@ -111,11 +128,12 @@ const RemindersPlugin: Plugin = async (ctx) => { const remindersToCancel = Array.from(state.reminders.values()).filter((r) => r.sessionID === sessionID) + let cancelledCount = 0 for (const reminder of remindersToCancel) { - await cancelReminder(reminder.id, ctx, state) + if (await cancelReminder(reminder.id, ctx, state)) cancelledCount++ } - logger.info(`[RemindersPlugin] Cancelled ${remindersToCancel.length} reminders for session ${sessionID}`) + logger.info(`[RemindersPlugin] Cancelled ${cancelledCount} reminders for session ${sessionID}`) } }, diff --git a/package.json b/package.json index df8deb1..972b8a3 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "opencode-reminders", - "version": "1.0.1", + "version": "1.0.2", "description": "Reminder plugin for OpenCode - schedule actions to run at intervals", "keywords": ["opencode", "reactive", "plugin", "reminders", "scheduler"], "author": "zelenkovsky", diff --git a/scheduler.ts b/scheduler.ts index 26f2101..47943fc 100644 --- a/scheduler.ts +++ b/scheduler.ts @@ -4,86 +4,249 @@ import { saveReminder, deleteReminder, loadReminder } from "./storage" import { acquireExecutionLease, acquireMutationLock } from "./leases" import { logger } from "./logger" +type ScheduledExecution = { + generation: string + token: string + controller?: AbortController + timer?: NodeJS.Timeout + state: State +} + +type ProjectRuntime = { + generation: string + executions: Map + persistence: Map> +} + +type ScheduleOptions = { + persist?: boolean + skipOverdue?: boolean +} + const RECONCILIATION_DELAY_MS = 250 -type ScheduleOptions = { skipOverdue?: boolean } +const runtimesKey = Symbol.for("opencode-reminders.scheduler-runtimes") +const runtimes: Map = + (globalThis as any)[runtimesKey] ?? ((globalThis as any)[runtimesKey] = new Map()) + +function projectKey(ctx: PluginInput): string { + return `${ctx.directory}\0${ctx.project.id}` +} + +function getRuntime(ctx: PluginInput): ProjectRuntime { + const key = projectKey(ctx) + let runtime = runtimes.get(key) + if (!runtime) { + runtime = { + generation: crypto.randomUUID(), + executions: new Map(), + persistence: new Map(), + } + runtimes.set(key, runtime) + } + return runtime +} + +function stateGeneration(state: State, ctx: PluginInput): string { + if (!state.generation) state.generation = getRuntime(ctx).generation + return state.generation +} + +function isCurrentGeneration(state: State, ctx: PluginInput): boolean { + return stateGeneration(state, ctx) === getRuntime(ctx).generation +} + +function isCurrentExecution(id: string, ctx: PluginInput, state: State, token: string): boolean { + const runtime = getRuntime(ctx) + const execution = runtime.executions.get(id) + return ( + stateGeneration(state, ctx) === runtime.generation + && execution?.generation === runtime.generation + && execution.token === token + ) +} + +function stopLocalExecution(id: string, runtime: ProjectRuntime): ScheduledExecution | undefined { + const execution = runtime.executions.get(id) + if (execution?.timer) clearTimeout(execution.timer) + execution?.controller?.abort() + execution?.state.timers.delete(id) + runtime.executions.delete(id) + return execution +} + +export function beginSchedulerGeneration(ctx: PluginInput): string { + const runtime = getRuntime(ctx) + for (const id of Array.from(runtime.executions.keys())) stopLocalExecution(id, runtime) + runtime.generation = crypto.randomUUID() + return runtime.generation +} + +export function isSchedulerGenerationCurrent(ctx: PluginInput, generation: string): boolean { + return getRuntime(ctx).generation === generation +} + +export async function waitForSchedulerMutations(ctx: PluginInput, generation: string): Promise { + const runtime = getRuntime(ctx) + while (runtime.generation === generation) { + const pending = Array.from(runtime.persistence.values()) + if (pending.length === 0) return true + await Promise.all(pending.map((operation) => operation.catch(() => {}))) + } + return false +} + +async function withMutationLock(id: string, ctx: PluginInput, action: () => Promise): Promise { + const runtime = getRuntime(ctx) + const operationID = `${id}\0${crypto.randomUUID()}` + const operation = (async () => { + const lock = await acquireMutationLock(id, ctx) + try { + return await action() + } finally { + await lock.release() + } + })() + runtime.persistence.set(operationID, operation) + try { + return await operation + } finally { + if (runtime.persistence.get(operationID) === operation) runtime.persistence.delete(operationID) + } +} + +function clearStateReminder(id: string, state: State): void { + const timer = state.timers.get(id) + if (timer) clearTimeout(timer) + state.timers.delete(id) + state.reminders.delete(id) +} + +async function deleteReminderLocked(id: string, ctx: PluginInput, state: State): Promise { + const execution = stopLocalExecution(id, getRuntime(ctx)) + clearStateReminder(id, state) + if (execution && execution.state !== state) clearStateReminder(id, execution.state) + await deleteReminder(id, ctx) +} + +export async function deleteStoredReminder(id: string, ctx: PluginInput, state: State): Promise { + if (!isCurrentGeneration(state, ctx)) return + await withMutationLock(id, ctx, async () => { + if (isCurrentGeneration(state, ctx)) await deleteReminderLocked(id, ctx, state) + }) +} + +export async function scheduleTimer( + reminder: Reminder, + ctx: PluginInput, + state: State, + config: PluginConfig, + options: ScheduleOptions = {}, +): Promise { + const runtime = getRuntime(ctx) + const generation = stateGeneration(state, ctx) + if (generation !== runtime.generation) { + clearStateReminder(reminder.id, state) + return false + } + + stopLocalExecution(reminder.id, runtime) + const token = crypto.randomUUID() + const execution: ScheduledExecution = { generation, token, state } + runtime.executions.set(reminder.id, execution) + + if (options.persist !== false) { + const snapshot = structuredClone(reminder) + try { + await withMutationLock(reminder.id, ctx, async () => { + if (isCurrentExecution(reminder.id, ctx, state, token)) await saveReminder(snapshot, ctx) + }) + } catch (error) { + if (isCurrentExecution(reminder.id, ctx, state, token)) { + stopLocalExecution(reminder.id, runtime) + clearStateReminder(reminder.id, state) + } + throw error + } + if (!isCurrentExecution(reminder.id, ctx, state, token)) { + clearStateReminder(reminder.id, state) + return false + } + } -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() 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) + if (state.timers.get(reminder.id) === timer) state.timers.delete(reminder.id) + if (!isCurrentExecution(reminder.id, ctx, state, token)) return + + const controller = new AbortController() + execution.timer = undefined + execution.controller = controller try { - await executeReminder(reminder, ctx, state, config, skipOverdue) + await executeReminder(reminder, ctx, state, config, skipOverdue, token, controller.signal) } catch (error) { logger.error(`Unhandled reminder timer failure ${reminder.id}:`, error) - scheduleReconciliation(reminder.id, ctx, state, config) + if (isCurrentExecution(reminder.id, ctx, state, token)) { + scheduleReconciliation(reminder.id, ctx, state, config) + } } }, delay) timer.unref() + execution.timer = timer state.timers.set(reminder.id, timer) + return true } function scheduleReconciliation(id: string, ctx: PluginInput, state: State, config: PluginConfig): void { + if (!isCurrentGeneration(state, ctx)) return const existing = state.timers.get(id) if (existing) clearTimeout(existing) + const generation = stateGeneration(state, ctx) const timer = setTimeout(async () => { - state.timers.delete(id) + if (state.timers.get(id) === timer) state.timers.delete(id) + if (!isSchedulerGenerationCurrent(ctx, generation)) return await reconcileReminder(id, ctx, state, config) }, RECONCILIATION_DELAY_MS) timer.unref() state.timers.set(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() - } -} - -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 reconcileReminder(id: string, ctx: PluginInput, state: State, config: PluginConfig): Promise { + if (!isCurrentGeneration(state, ctx)) return try { const stored = await loadReminder(id, ctx) + if (!isCurrentGeneration(state, ctx)) return if (!stored || stored.status !== "active") { - const timer = state.timers.get(id) - if (timer) clearTimeout(timer) - state.timers.delete(id) - state.reminders.delete(id) + clearStateReminder(id, state) return } state.reminders.set(id, stored) - await scheduleTimer(stored, ctx, state, config) + await scheduleTimer(stored, ctx, state, config, { persist: false }) } 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 { +export async function cleanupReminderSnapshot( + snapshot: Reminder, + ctx: PluginInput, + state: State, + config: PluginConfig, +): Promise { + if (!isCurrentGeneration(state, ctx)) return false try { - return await withMutationLock(snapshot.id, ctx, async () => { + let cleaned = false + await withMutationLock(snapshot.id, ctx, async () => { + if (!isCurrentGeneration(state, ctx)) return const current = await loadReminder(snapshot.id, ctx) - if (!current || current.time.nextExecution !== snapshot.time.nextExecution) { - await reconcileReminder(snapshot.id, ctx, state, config) - return false - } + if (!current || current.time.nextExecution !== snapshot.time.nextExecution) return await deleteReminderLocked(snapshot.id, ctx, state) - return true + cleaned = true }) + if (!cleaned) await reconcileReminder(snapshot.id, ctx, state, config) + return cleaned } catch (error) { logger.error(`Failed conditional cleanup for reminder ${snapshot.id}:`, error) await reconcileReminder(snapshot.id, ctx, state, config) @@ -91,7 +254,14 @@ export async function cleanupReminderSnapshot(snapshot: Reminder, ctx: PluginInp } } -async function skipMissedRecurring(reminder: Reminder, occurrence: number, ctx: PluginInput, state: State, config: PluginConfig, skipOverdue: boolean): Promise { +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 () => { @@ -99,16 +269,25 @@ async function skipMissedRecurring(reminder: Reminder, occurrence: number, 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) + if (isCurrentGeneration(state, ctx)) { + state.reminders.set(current.id, current) + await scheduleTimer(current, ctx, state, config, { persist: false }) + } advanced = true }) if (!advanced) await reconcileReminder(reminder.id, ctx, state, config) return true } -async function handlePromptError(reminder: Reminder, occurrence: number, error: any, ctx: PluginInput, state: State, config: PluginConfig): Promise { +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 () => { @@ -117,9 +296,11 @@ async function handlePromptError(reminder: Reminder, occurrence: number, error: 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) + if (isCurrentGeneration(state, ctx)) { + state.reminders.set(current.id, current) + await scheduleTimer(current, ctx, state, config, { persist: false }) + } } else { await deleteReminderLocked(current.id, ctx, state) } @@ -131,19 +312,46 @@ async function handlePromptError(reminder: Reminder, occurrence: number, error: } } -export async function executeReminder(reminder: Reminder, ctx: PluginInput, state: State, config: PluginConfig, skipOverdue = false): Promise { +export async function executeReminder( + reminder: Reminder, + ctx: PluginInput, + state: State, + config: PluginConfig, + skipOverdue = false, + token?: string, + signal?: AbortSignal, +): Promise { + const runtime = getRuntime(ctx) + if (!token) { + if (!isCurrentGeneration(state, ctx)) return + stopLocalExecution(reminder.id, runtime) + token = crypto.randomUUID() + const controller = new AbortController() + signal = controller.signal + runtime.executions.set(reminder.id, { + generation: stateGeneration(state, ctx), + token, + controller, + state, + }) + } + 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) + if (acquisition.status === "error") { + logger.error(`Could not acquire reminder lease ${reminder.id}:`, acquisition.error) + } + if (isCurrentExecution(reminder.id, ctx, state, token)) { + 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 (!isCurrentExecution(reminder.id, ctx, state, token)) return if (!stored || stored.status !== "active" || stored.time.nextExecution !== occurrence) { await reconcileReminder(reminder.id, ctx, state, config) return @@ -156,34 +364,52 @@ export async function executeReminder(reminder: Reminder, ctx: PluginInput, stat } try { - await ctx.client.session.prompt({ path: { id: reminder.sessionID }, body: { parts: [{ type: "text", text: reminder.originalPrompt }] } }) + await ctx.client.session.prompt({ + path: { id: reminder.sessionID }, + signal, + body: { + parts: [{ type: "text", text: reminder.originalPrompt }], + ...(reminder.agent === undefined ? {} : { agent: reminder.agent }), + }, + }) } catch (error: any) { + if (!isCurrentExecution(reminder.id, ctx, state, token)) return logger.error(`Reminder ${reminder.id} prompt failed:`, error) await handlePromptError(reminder, occurrence, error, ctx, state, config) return } - // Prompt acceptance is separate from its durable transition. Never treat a failure here as a prompt error. + // A stale generation may finish after an API ignores abort. Its occurrence lease and + // conditional durable transition still prevent resurrection or a duplicate local prompt. try { - const completed = await withMutationLock(reminder.id, ctx, async (): Promise => { + const completed = await withMutationLock(reminder.id, ctx, async () => { 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) + if (isCurrentExecution(reminder.id, ctx, state, token)) { + state.reminders.set(current.id, current) + await scheduleTimer(current, ctx, state, config, { persist: false }) + } } else { await deleteReminderLocked(current.id, ctx, state) } return current }) if (!completed) await reconcileReminder(reminder.id, ctx, state, config) - else if (config.notifications.enabled) { + else if (config.notifications.enabled && isCurrentGeneration(state, ctx)) { try { - await ctx.client.tui.showToast({ body: { message: completed.type === "recurring" ? `Reminder executed: ${completed.userDescription}` : `Reminder completed: ${completed.userDescription}`, variant: "success" } }) + 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) } @@ -197,15 +423,19 @@ export async function executeReminder(reminder: Reminder, ctx: PluginInput, stat await acquisition.lease.release() } catch (error) { logger.error(`Failed to release reminder lease ${reminder.id}:`, error) - scheduleReconciliation(reminder.id, ctx, state, config) + if (isCurrentGeneration(state, ctx)) scheduleReconciliation(reminder.id, ctx, state, config) } } } -export async function cancelReminder(id: string, ctx: PluginInput, state: State): Promise { +export async function cancelReminder(id: string, ctx: PluginInput, state: State): Promise { + const execution = stopLocalExecution(id, getRuntime(ctx)) + clearStateReminder(id, state) + if (execution && execution.state !== state) clearStateReminder(id, execution.state) try { - await withMutationLock(id, ctx, async () => deleteReminderLocked(id, ctx, state)) + await withMutationLock(id, ctx, async () => deleteReminder(id, ctx)) logger.info(`Reminder ${id} cancelled`) + return true } catch (error) { logger.error(`Failed to cancel reminder ${id}:`, error) throw error diff --git a/storage.ts b/storage.ts index e0662d8..1de5542 100644 --- a/storage.ts +++ b/storage.ts @@ -44,7 +44,7 @@ export async function loadReminder(id: string, ctx: PluginInput): Promise { const dir = await getStorageDir(ctx) const filePath = path.join(dir, `${id}.json`) - await ctx.$`rm -f ${filePath}`.quiet() + await rm(filePath, { force: true }) } export async function listReminders(ctx: PluginInput): Promise { diff --git a/test/README.md b/test/README.md index d196f1e..0cfc222 100644 --- a/test/README.md +++ b/test/README.md @@ -3,9 +3,10 @@ Run the suite with `bun test`; it uses isolated temporary project directories and child processes only. `bun x tsc --noEmit` checks the test and plugin TypeScript sources. -Coverage includes persistence, scheduler execution and cancellation fencing, restoration cadence, -atomic lease ownership/recovery, and multiprocess prompt contention. Child-process tests use strict -stdin barriers and are responsible for closing their children during cleanup. +Coverage includes persistence, agent capture and forwarding, scheduler execution and generation +fencing, active cancellation, restoration cadence, atomic lease ownership/recovery, and multiprocess +prompt contention. Child-process tests use strict stdin barriers and are responsible for closing +their children during cleanup. 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/integration.test.ts b/test/integration.test.ts index 58a4324..41b3061 100644 --- a/test/integration.test.ts +++ b/test/integration.test.ts @@ -2,6 +2,7 @@ import { test, expect, describe, beforeEach, afterEach } from "bun:test" import RemindersPlugin from "../index" import type { PluginInput } from "@opencode-ai/plugin" import { $ } from "bun" +import { acquireMutationLock } from "../leases" async function createMockContext(tmpDir: string): Promise { const sessions = new Map() @@ -80,6 +81,79 @@ describe("Integration Tests", () => { expect(result).toContain("File change check") }) + test("reminderadd captures the creating agent", async () => { + const plugin = await RemindersPlugin(ctx) + await plugin.tool!.reminderadd.execute( + { + interval_seconds: 60, + type: "one-time" as const, + action_prompt: "agent action", + description: "Agent capture", + }, + { sessionID: "ses-agent", agent: "build" } as any, + ) + + const directory = `${tmpDir}/.opencode/reminders/test-project-integration` + const files = await Array.fromAsync(new Bun.Glob("*.json").scan({ cwd: directory, absolute: true })) + expect(files).toHaveLength(1) + expect((await Bun.file(files[0]).json()).agent).toBe("build") + }) + + test("replacement initialization drains and fences an in-progress add", async () => { + const plugin = await RemindersPlugin(ctx) + const lock = await acquireMutationLock("blocked", ctx) + const originalUUID = crypto.randomUUID + ;(crypto as any).randomUUID = () => "blocked" + try { + const add = plugin.tool!.reminderadd.execute( + { + interval_seconds: 60, + type: "one-time" as const, + action_prompt: "blocked add", + description: "Blocked add", + }, + { sessionID: "ses-blocked" } as any, + ) + await Bun.sleep(50) + const replacement = RemindersPlugin(ctx) + await Bun.sleep(50) + await lock.release() + + expect(await add).toContain("scheduler reloaded") + const replacementPlugin = await replacement + expect(await replacementPlugin.tool!.reminderlist.execute( + {}, + { sessionID: "ses-blocked" } as any, + )).toContain("No active reminders") + } finally { + ;(crypto as any).randomUUID = originalUUID + await lock.release() + } + }) + + test("a stale remove tool cancels the replacement generation reminder", async () => { + const stale = await RemindersPlugin(ctx) + await stale.tool!.reminderadd.execute( + { + interval_seconds: 60, + type: "recurring" as const, + action_prompt: "stale cancellation", + description: "Stale generation cancellation", + }, + { sessionID: "ses-stale" } as any, + ) + const replacement = await RemindersPlugin(ctx) + + expect(await stale.tool!.reminderremove.execute( + { description_pattern: "Stale generation" }, + { sessionID: "ses-stale" } as any, + )).toContain("Reminder cancelled") + expect(await replacement.tool!.reminderlist.execute( + {}, + { sessionID: "ses-stale" } as any, + )).toContain("No active reminders") + }) + test("reminderlist tool returns empty for new session", async () => { const plugin = await RemindersPlugin(ctx) diff --git a/test/scheduler.test.ts b/test/scheduler.test.ts index 9cc694a..b987144 100644 --- a/test/scheduler.test.ts +++ b/test/scheduler.test.ts @@ -1,11 +1,26 @@ import { test, expect, describe, beforeEach, afterEach } from "bun:test" -import { scheduleTimer, executeReminder, cancelReminder, reconcileReminder, cleanupReminderSnapshot } from "../scheduler" +import { + beginSchedulerGeneration, + 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" +function deferred() { + let resolve!: (value: T | PromiseLike) => void + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise + }) + return { promise, resolve } +} + async function createMockContext(tmpDir: string): Promise { return { client: { @@ -624,4 +639,113 @@ describe("Scheduler", () => { expect(state.reminders.has(reminder.id)).toBe(true) expect(state.timers.has(reminder.id)).toBe(true) }) + + test("cancellation aborts an active prompt and cannot reschedule it", async () => { + const promptStarted = deferred() + const finishPrompt = deferred() + ;(ctx.client.session.prompt as any) = async (options: any) => { + promptStarted.resolve(options.signal) + await finishPrompt.promise + return { data: {}, error: undefined, response: {} } + } + const reminder: Reminder = { + id: "rem-active-cancel", + sessionID: "ses-test", + projectID: state.projectID, + type: "recurring", + interval: 60000, + originalPrompt: "active cancellation", + userDescription: "Active cancellation", + time: { created: Date.now(), nextExecution: Date.now() + 5 }, + status: "active", + } + state.reminders.set(reminder.id, reminder) + await scheduleTimer(reminder, ctx, state, config) + + const signal = await promptStarted.promise + await cancelReminder(reminder.id, ctx, state) + finishPrompt.resolve() + await Bun.sleep(0) + + expect(signal.aborted).toBe(true) + expect(await loadReminder(reminder.id, ctx)).toBeNull() + expect(state.reminders.has(reminder.id)).toBe(false) + expect(state.timers.has(reminder.id)).toBe(false) + }) + + test("replacement generation fences an active old occurrence", async () => { + const promptStarted = deferred() + const finishPrompt = deferred() + ;(ctx.client.session.prompt as any) = async (options: any) => { + promptStarted.resolve(options.signal) + await finishPrompt.promise + return { data: {}, error: undefined, response: {} } + } + const reminder: Reminder = { + id: "rem-hot-reload", + sessionID: "ses-test", + projectID: state.projectID, + type: "recurring", + interval: 60000, + originalPrompt: "hot reload", + userDescription: "Hot reload", + time: { created: Date.now(), nextExecution: Date.now() + 5 }, + status: "active", + } + state.reminders.set(reminder.id, reminder) + await scheduleTimer(reminder, ctx, state, config) + const oldSignal = await promptStarted.promise + + const replacement: State = { + reminders: new Map(), + timers: new Map(), + projectID: state.projectID, + generation: beginSchedulerGeneration(ctx), + } + const replacementReminder = structuredClone(reminder) + replacementReminder.time.nextExecution = Date.now() + 60000 + replacement.reminders.set(reminder.id, replacementReminder) + await scheduleTimer(replacementReminder, ctx, replacement, config) + finishPrompt.resolve() + await Bun.sleep(0) + + expect(oldSignal.aborted).toBe(true) + expect(state.timers.has(reminder.id)).toBe(false) + expect(replacement.timers.has(reminder.id)).toBe(true) + expect((await loadReminder(reminder.id, ctx))?.time.nextExecution).toBe( + replacementReminder.time.nextExecution, + ) + await cancelReminder(reminder.id, ctx, replacement) + }) + + test("forwards a captured agent and omits it for legacy reminders", async () => { + const bodies: any[] = [] + ;(ctx.client.session.prompt as any) = async (options: any) => { + bodies.push(options.body) + return { data: {}, error: undefined, response: {} } + } + const base: Reminder = { + id: "rem-agent", + sessionID: "ses-test", + projectID: state.projectID, + type: "one-time", + interval: 1000, + originalPrompt: "agent forwarding", + userDescription: "Agent forwarding", + time: { created: Date.now(), nextExecution: Date.now() + 1000 }, + status: "active", + } + const withAgent = { ...base, agent: "build" } + state.reminders.set(withAgent.id, withAgent) + await saveReminder(withAgent, ctx) + await executeReminder(withAgent, ctx, state, config) + + const legacy = { ...base, id: "rem-legacy-agent" } + state.reminders.set(legacy.id, legacy) + await saveReminder(legacy, ctx) + await executeReminder(legacy, ctx, state, config) + + expect(bodies[0].agent).toBe("build") + expect(Object.hasOwn(bodies[1], "agent")).toBe(false) + }) }) diff --git a/test/types.test.ts b/test/types.test.ts index bd6e386..b309e72 100644 --- a/test/types.test.ts +++ b/test/types.test.ts @@ -21,6 +21,23 @@ describe("Reminder Types", () => { expect(() => ReminderSchema.parse(validReminder)).not.toThrow() }) + test("ReminderSchema preserves an optional agent and accepts legacy omission", () => { + const reminder = { + id: "test-agent", + sessionID: "ses-agent", + projectID: "prj-agent", + type: "one-time" as const, + interval: 5000, + originalPrompt: "test", + userDescription: "Agent reminder", + time: { created: Date.now(), nextExecution: Date.now() + 5000 }, + status: "active" as const, + } + + expect(ReminderSchema.parse(reminder).agent).toBeUndefined() + expect(ReminderSchema.parse({ ...reminder, agent: "build" }).agent).toBe("build") + }) + test("ReminderSchema rejects invalid type", () => { const invalidReminder = { id: "test-123", diff --git a/tools/reminderadd.ts b/tools/reminderadd.ts index 9e6a9d3..4bbf7bb 100644 --- a/tools/reminderadd.ts +++ b/tools/reminderadd.ts @@ -1,6 +1,5 @@ import { tool, type PluginInput } from "@opencode-ai/plugin" import type { Reminder, State, PluginConfig } from "../types" -import { saveReminder } from "../storage" import { scheduleTimer } from "../scheduler" import DESCRIPTION from "./reminderadd.txt" import { logger } from "../logger" @@ -42,6 +41,7 @@ export function createReminderAddTool( interval: args.interval_seconds * 1000, originalPrompt: args.action_prompt, userDescription: args.description, + agent: context.agent, time: { created: Date.now(), nextExecution: Date.now() + args.interval_seconds * 1000, @@ -50,8 +50,9 @@ export function createReminderAddTool( } state.reminders.set(reminder.id, reminder) - await saveReminder(reminder, ctx) - await scheduleTimer(reminder, ctx, state, config) + if (!(await scheduleTimer(reminder, ctx, state, config))) { + return "Reminder scheduler reloaded while setting the reminder. Check active reminders before retrying." + } logger.info(`[RemindersPlugin] Created ${args.type} reminder ${reminder.id}: ${args.description}`) diff --git a/tools/reminderremove.ts b/tools/reminderremove.ts index 3e6f071..7a2283b 100644 --- a/tools/reminderremove.ts +++ b/tools/reminderremove.ts @@ -33,16 +33,20 @@ export function createReminderRemoveTool(ctx: PluginInput, state: State) { // Cancel all matching reminders const cancelledDescriptions: string[] = [] for (const reminder of matches) { - await cancelReminder(reminder.id, ctx, state) - cancelledDescriptions.push(reminder.userDescription) - logger.info(`[RemindersPlugin] Cancelled reminder ${reminder.id} via user request`) + if (await cancelReminder(reminder.id, ctx, state)) { + cancelledDescriptions.push(reminder.userDescription) + logger.info(`[RemindersPlugin] Cancelled reminder ${reminder.id} via user request`) + } } - if (matches.length === 1) { + if (cancelledDescriptions.length === 0) { + return "Reminder scheduler reloaded while cancelling. Check active reminders before retrying." + } + if (cancelledDescriptions.length === 1) { return `Reminder cancelled: ${cancelledDescriptions[0]}` } else { const cancelledList = cancelledDescriptions.map((desc) => `- ${desc}`).join("\n") - return `${matches.length} reminders cancelled:\n${cancelledList}` + return `${cancelledDescriptions.length} reminders cancelled:\n${cancelledList}` } }, }) diff --git a/types.ts b/types.ts index 6807372..92b5108 100644 --- a/types.ts +++ b/types.ts @@ -8,6 +8,7 @@ export const ReminderSchema = z.object({ interval: z.number(), originalPrompt: z.string(), userDescription: z.string(), + agent: z.string().optional(), time: z.object({ created: z.number(), nextExecution: z.number(), @@ -22,6 +23,7 @@ export type State = { reminders: Map timers: Map projectID: string + generation?: string } export type PluginConfig = { From 3eff7d38bc23a7c1ca83c5bc3888a9dca04125c0 Mon Sep 17 00:00:00 2001 From: Marc Warne Date: Sat, 25 Jul 2026 16:59:57 +0100 Subject: [PATCH 3/5] fix: preserve overdue policy across retries --- README.md | 6 ++ scheduler.ts | 87 ++++++++++++++++++++++------- test/README.md | 5 +- test/integration.test.ts | 115 ++++++++++++++++++++++++++++++++++++++- test/scheduler.test.ts | 17 +++++- 5 files changed, 207 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index d2988b1..48614bc 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,12 @@ Post-prompt recurrence updates and every plugin cancellation/restore cleanup sha per-reminder mutation lock, so the durable read-and-transition cannot race a deletion. The lock is not held while the prompt API runs. +Reconciliation retries use capped exponential backoff. A recurring occurrence found overdue during +startup retains its skip-overdue policy across read or atomic-write failures, so unchanged old JSON +is never reinterpreted as a normal occurrence and prompted. Normal runtime failures retain the +occurrence for an at-least-once retry; the retry history resets after durable state advances into the +future. + Within one server process, plugin generations also fence hot reloads. Reinitialization clears old timers, aborts active requests, waits for mutations already in progress, and prevents stale timer or reconciliation callbacks from changing replacement state. Cancellation aborts the active local diff --git a/scheduler.ts b/scheduler.ts index 47943fc..2b30781 100644 --- a/scheduler.ts +++ b/scheduler.ts @@ -21,9 +21,14 @@ type ProjectRuntime = { type ScheduleOptions = { persist?: boolean skipOverdue?: boolean + retryAttempt?: number } const RECONCILIATION_DELAY_MS = 250 +const MAX_RECONCILIATION_DELAY_MS = 30_000 +const MAX_RECONCILIATION_ATTEMPT = 1 + Math.ceil( + Math.log2(MAX_RECONCILIATION_DELAY_MS / RECONCILIATION_DELAY_MS), +) const runtimesKey = Symbol.for("opencode-reminders.scheduler-runtimes") const runtimes: Map = (globalThis as any)[runtimesKey] ?? ((globalThis as any)[runtimesKey] = new Map()) @@ -176,6 +181,7 @@ export async function scheduleTimer( const now = Date.now() const delay = Math.max(0, reminder.time.nextExecution - now) const skipOverdue = options.skipOverdue === true && reminder.time.nextExecution < now + const retryAttempt = reminder.time.nextExecution < now ? options.retryAttempt ?? 0 : 0 const timer = setTimeout(async () => { if (state.timers.get(reminder.id) === timer) state.timers.delete(reminder.id) if (!isCurrentExecution(reminder.id, ctx, state, token)) return @@ -184,11 +190,11 @@ export async function scheduleTimer( execution.timer = undefined execution.controller = controller try { - await executeReminder(reminder, ctx, state, config, skipOverdue, token, controller.signal) + await executeReminder(reminder, ctx, state, config, skipOverdue, retryAttempt, token, controller.signal) } catch (error) { logger.error(`Unhandled reminder timer failure ${reminder.id}:`, error) if (isCurrentExecution(reminder.id, ctx, state, token)) { - scheduleReconciliation(reminder.id, ctx, state, config) + scheduleReconciliation(reminder.id, ctx, state, config, { skipOverdue, retryAttempt }) } } }, delay) @@ -198,21 +204,45 @@ export async function scheduleTimer( return true } -function scheduleReconciliation(id: string, ctx: PluginInput, state: State, config: PluginConfig): void { +function scheduleReconciliation( + id: string, + ctx: PluginInput, + state: State, + config: PluginConfig, + options: ScheduleOptions = {}, +): void { if (!isCurrentGeneration(state, ctx)) return const existing = state.timers.get(id) if (existing) clearTimeout(existing) const generation = stateGeneration(state, ctx) + const retryAttempt = Math.min((options.retryAttempt ?? 0) + 1, MAX_RECONCILIATION_ATTEMPT) + const delay = Math.min( + RECONCILIATION_DELAY_MS * (2 ** (retryAttempt - 1)), + MAX_RECONCILIATION_DELAY_MS, + ) const timer = setTimeout(async () => { if (state.timers.get(id) === timer) state.timers.delete(id) + const execution = getRuntime(ctx).executions.get(id) + if (execution?.timer === timer) execution.timer = undefined if (!isSchedulerGenerationCurrent(ctx, generation)) return - await reconcileReminder(id, ctx, state, config) - }, RECONCILIATION_DELAY_MS) + await reconcileReminder(id, ctx, state, config, { + skipOverdue: options.skipOverdue, + retryAttempt, + }) + }, delay) timer.unref() state.timers.set(id, timer) + const execution = getRuntime(ctx).executions.get(id) + if (execution?.generation === generation) execution.timer = timer } -export async function reconcileReminder(id: string, ctx: PluginInput, state: State, config: PluginConfig): Promise { +export async function reconcileReminder( + id: string, + ctx: PluginInput, + state: State, + config: PluginConfig, + options: ScheduleOptions = {}, +): Promise { if (!isCurrentGeneration(state, ctx)) return try { const stored = await loadReminder(id, ctx) @@ -222,10 +252,14 @@ export async function reconcileReminder(id: string, ctx: PluginInput, state: Sta return } state.reminders.set(id, stored) - await scheduleTimer(stored, ctx, state, config, { persist: false }) + await scheduleTimer(stored, ctx, state, config, { + persist: false, + skipOverdue: options.skipOverdue, + retryAttempt: options.retryAttempt, + }) } catch (error) { logger.error(`Failed to reconcile reminder ${id}:`, error) - scheduleReconciliation(id, ctx, state, config) + scheduleReconciliation(id, ctx, state, config, options) } } @@ -261,6 +295,7 @@ async function skipMissedRecurring( state: State, config: PluginConfig, skipOverdue: boolean, + retryAttempt: number, ): Promise { if (!skipOverdue || reminder.type !== "recurring") return false let advanced = false @@ -276,7 +311,9 @@ async function skipMissedRecurring( } advanced = true }) - if (!advanced) await reconcileReminder(reminder.id, ctx, state, config) + if (!advanced) { + await reconcileReminder(reminder.id, ctx, state, config, { skipOverdue, retryAttempt }) + } return true } @@ -287,6 +324,7 @@ async function handlePromptError( ctx: PluginInput, state: State, config: PluginConfig, + retryAttempt: number, ): Promise { try { let changed = false @@ -305,10 +343,10 @@ async function handlePromptError( await deleteReminderLocked(current.id, ctx, state) } }) - if (!changed) await reconcileReminder(reminder.id, ctx, state, config) + if (!changed) await reconcileReminder(reminder.id, ctx, state, config, { retryAttempt }) } catch (mutationError) { logger.error(`Failed to handle reminder prompt error ${reminder.id}:`, mutationError) - await reconcileReminder(reminder.id, ctx, state, config) + scheduleReconciliation(reminder.id, ctx, state, config, { retryAttempt }) } } @@ -318,6 +356,7 @@ export async function executeReminder( state: State, config: PluginConfig, skipOverdue = false, + retryAttempt = 0, token?: string, signal?: AbortSignal, ): Promise { @@ -343,7 +382,7 @@ export async function executeReminder( logger.error(`Could not acquire reminder lease ${reminder.id}:`, acquisition.error) } if (isCurrentExecution(reminder.id, ctx, state, token)) { - scheduleReconciliation(reminder.id, ctx, state, config) + scheduleReconciliation(reminder.id, ctx, state, config, { skipOverdue, retryAttempt }) } return } @@ -353,13 +392,21 @@ export async function executeReminder( const stored = await loadReminder(reminder.id, ctx) if (!isCurrentExecution(reminder.id, ctx, state, token)) return if (!stored || stored.status !== "active" || stored.time.nextExecution !== occurrence) { - await reconcileReminder(reminder.id, ctx, state, config) + await reconcileReminder(reminder.id, ctx, state, config, { skipOverdue, retryAttempt }) return } - if (await skipMissedRecurring(stored, occurrence, ctx, state, config, skipOverdue)) return + if (await skipMissedRecurring( + stored, + occurrence, + ctx, + state, + config, + skipOverdue, + retryAttempt, + )) return } catch (error) { logger.error(`Pre-prompt validation failed for reminder ${reminder.id}:`, error) - await reconcileReminder(reminder.id, ctx, state, config) + scheduleReconciliation(reminder.id, ctx, state, config, { skipOverdue, retryAttempt }) return } @@ -375,7 +422,7 @@ export async function executeReminder( } catch (error: any) { if (!isCurrentExecution(reminder.id, ctx, state, token)) return logger.error(`Reminder ${reminder.id} prompt failed:`, error) - await handlePromptError(reminder, occurrence, error, ctx, state, config) + await handlePromptError(reminder, occurrence, error, ctx, state, config, retryAttempt) return } @@ -399,7 +446,7 @@ export async function executeReminder( } return current }) - if (!completed) await reconcileReminder(reminder.id, ctx, state, config) + if (!completed) await reconcileReminder(reminder.id, ctx, state, config, { retryAttempt }) else if (config.notifications.enabled && isCurrentGeneration(state, ctx)) { try { await ctx.client.tui.showToast({ @@ -416,14 +463,16 @@ export async function executeReminder( } } catch (error) { logger.error(`Post-prompt transition failed for reminder ${reminder.id}:`, error) - await reconcileReminder(reminder.id, ctx, state, config) + scheduleReconciliation(reminder.id, ctx, state, config, { retryAttempt }) } } finally { try { await acquisition.lease.release() } catch (error) { logger.error(`Failed to release reminder lease ${reminder.id}:`, error) - if (isCurrentGeneration(state, ctx)) scheduleReconciliation(reminder.id, ctx, state, config) + if (isCurrentGeneration(state, ctx)) { + scheduleReconciliation(reminder.id, ctx, state, config, { skipOverdue, retryAttempt }) + } } } } diff --git a/test/README.md b/test/README.md index 0cfc222..0735b29 100644 --- a/test/README.md +++ b/test/README.md @@ -5,8 +5,9 @@ only. `bun x tsc --noEmit` checks the test and plugin TypeScript sources. Coverage includes persistence, agent capture and forwarding, scheduler execution and generation fencing, active cancellation, restoration cadence, atomic lease ownership/recovery, and multiprocess -prompt contention. Child-process tests use strict stdin barriers and are responsible for closing -their children during cleanup. +prompt contention. Persistence-failure tests verify startup skip-overdue retention and capped retry +rates while the prior JSON remains readable. Child-process tests use strict stdin barriers and are +responsible for closing their children during cleanup. 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/integration.test.ts b/test/integration.test.ts index 41b3061..3d693b7 100644 --- a/test/integration.test.ts +++ b/test/integration.test.ts @@ -1,8 +1,30 @@ -import { test, expect, describe, beforeEach, afterEach } from "bun:test" +import { test, expect, describe, beforeEach, afterEach, spyOn } from "bun:test" +import * as fsPromises from "node:fs/promises" import RemindersPlugin from "../index" import type { PluginInput } from "@opencode-ai/plugin" import { $ } from "bun" import { acquireMutationLock } from "../leases" +import { cancelReminder, executeReminder } from "../scheduler" +import { getStorageDir, loadReminder, saveReminder } from "../storage" +import type { PluginConfig, Reminder, State } from "../types" + +async function waitFor(predicate: () => boolean | Promise, timeout = 4000): Promise { + const deadline = Date.now() + timeout + while (!(await predicate())) { + if (Date.now() >= deadline) throw new Error("Timed out waiting for condition") + await Bun.sleep(25) + } +} + +function failReminderRenames() { + const rename = fsPromises.rename + return spyOn(fsPromises, "rename").mockImplementation((async (source: any, destination: any) => { + if (String(destination).endsWith(".json")) { + throw Object.assign(new Error("simulated atomic rename failure"), { code: "EIO" }) + } + return rename(source, destination) + }) as any) +} async function createMockContext(tmpDir: string): Promise { const sessions = new Map() @@ -154,6 +176,97 @@ describe("Integration Tests", () => { )).toContain("No active reminders") }) + test("startup preserves a stored reminder when advancing its missed schedule cannot be persisted", async () => { + const plugin = await RemindersPlugin(ctx) + await plugin.tool!.reminderadd.execute( + { + interval_seconds: 60, + type: "recurring" as const, + action_prompt: "must not execute a missed startup occurrence", + description: "Startup persistence failure", + }, + { sessionID: "ses-startup-persistence" } as any, + ) + const directory = await getStorageDir(ctx) + const files = await Array.fromAsync(new Bun.Glob("*.json").scan({ cwd: directory, absolute: true })) + const stored = await Bun.file(files[0]).json() + stored.time.nextExecution = Date.now() - 1000 + await Bun.write(files[0], JSON.stringify(stored, null, 2)) + + let prompts = 0 + ;(ctx.client.session.prompt as any) = async () => { + prompts++ + return { data: {}, error: undefined, response: {} } + } + const rename = failReminderRenames() + let replacement: Awaited> | undefined + try { + replacement = await RemindersPlugin(ctx) + await Bun.sleep(850) + + expect(prompts).toBe(0) + expect((await Bun.file(files[0]).json()).time.nextExecution).toBe(stored.time.nextExecution) + expect(rename.mock.calls.length).toBeGreaterThanOrEqual(2) + expect(rename.mock.calls.length).toBeLessThanOrEqual(4) + } finally { + rename.mockRestore() + } + + await waitFor(async () => (await Bun.file(files[0]).json()).time.nextExecution > Date.now()) + expect(prompts).toBe(0) + expect(await replacement!.tool!.reminderremove.execute( + { description_pattern: "Startup persistence failure" }, + { sessionID: "ses-startup-persistence" } as any, + )).toContain("Reminder cancelled") + }) + + test("runtime transition failures retry with backoff and retain the future occurrence", async () => { + const reminder: Reminder = { + id: "rem-runtime-backoff", + sessionID: "ses-runtime-backoff", + projectID: ctx.project.id, + type: "recurring", + interval: 60000, + originalPrompt: "runtime retry", + userDescription: "Runtime retry", + time: { created: Date.now(), nextExecution: Date.now() - 1 }, + status: "active", + } + const state: State = { + reminders: new Map([[reminder.id, reminder]]), + timers: new Map(), + projectID: ctx.project.id, + } + const config: PluginConfig = { + enabled: true, + max_reminders_per_project: 50, + min_interval_seconds: 30, + notifications: { enabled: false }, + } + await saveReminder(reminder, ctx) + let prompts = 0 + ;(ctx.client.session.prompt as any) = async () => { + prompts++ + return { data: {}, error: undefined, response: {} } + } + + const rename = failReminderRenames() + try { + await executeReminder(reminder, ctx, state, config) + await Bun.sleep(850) + expect(prompts).toBeGreaterThanOrEqual(2) + expect(prompts).toBeLessThanOrEqual(3) + expect((await loadReminder(reminder.id, ctx))!.time.nextExecution).toBe(reminder.time.nextExecution) + } finally { + rename.mockRestore() + } + + await waitFor(async () => (await loadReminder(reminder.id, ctx))!.time.nextExecution > Date.now()) + expect(prompts).toBeLessThanOrEqual(4) + expect(state.timers.has(reminder.id)).toBe(true) + await cancelReminder(reminder.id, ctx, state) + }) + test("reminderlist tool returns empty for new session", async () => { const plugin = await RemindersPlugin(ctx) diff --git a/test/scheduler.test.ts b/test/scheduler.test.ts index b987144..4d2078a 100644 --- a/test/scheduler.test.ts +++ b/test/scheduler.test.ts @@ -1,6 +1,7 @@ import { test, expect, describe, beforeEach, afterEach } from "bun:test" import { beginSchedulerGeneration, + waitForSchedulerMutations, scheduleTimer, executeReminder, cancelReminder, @@ -21,6 +22,14 @@ function deferred() { return { promise, resolve } } +async function waitFor(predicate: () => boolean | Promise, timeout = 2000): Promise { + const deadline = Date.now() + timeout + while (!(await predicate())) { + if (Date.now() >= deadline) throw new Error("Timed out waiting for condition") + await Bun.sleep(10) + } +} + async function createMockContext(tmpDir: string): Promise { return { client: { @@ -72,6 +81,8 @@ describe("Scheduler", () => { }) afterEach(async () => { + const generation = beginSchedulerGeneration(ctx) + await waitForSchedulerMutations(ctx, generation) for (const timer of state.timers.values()) { clearTimeout(timer) } @@ -665,7 +676,11 @@ describe("Scheduler", () => { const signal = await promptStarted.promise await cancelReminder(reminder.id, ctx, state) finishPrompt.resolve() - await Bun.sleep(0) + const directory = await getStorageDir(ctx) + await waitFor(async () => { + const leases = await Array.fromAsync(new Bun.Glob("*.lease").scan({ cwd: directory })) + return leases.length === 0 + }) expect(signal.aborted).toBe(true) expect(await loadReminder(reminder.id, ctx)).toBeNull() From fa9fc6f0c27ae344dfb7f6ec5fc4943dfca07ea6 Mon Sep 17 00:00:00 2001 From: Marc Warne Date: Sat, 25 Jul 2026 17:09:19 +0100 Subject: [PATCH 4/5] fix: fence reconciliation after cancellation --- README.md | 4 ++ scheduler.ts | 130 ++++++++++++++++++++++++++++++++++++----- test/README.md | 3 + test/scheduler.test.ts | 52 ++++++++++++++++- 4 files changed, 172 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 48614bc..7309fd8 100644 --- a/README.md +++ b/README.md @@ -86,6 +86,10 @@ timers, aborts active requests, waits for mutations already in progress, and pre reconciliation callbacks from changing replacement state. Cancellation aborts the active local request immediately, then uses the same mutation lock as cross-process transitions. +Reconciliation reads are fenced per reminder and revalidated under the mutation lock. Once local +cancellation returns, a read that completed before deletion cannot restore stale list state or a +timer, even if its continuation resumes later. + 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 diff --git a/scheduler.ts b/scheduler.ts index 2b30781..f2a1317 100644 --- a/scheduler.ts +++ b/scheduler.ts @@ -16,6 +16,13 @@ type ProjectRuntime = { generation: string executions: Map persistence: Map> + fences: Map +} + +type ReminderFence = { + version: number + reconciliations: number + tombstone: boolean } type ScheduleOptions = { @@ -45,9 +52,11 @@ function getRuntime(ctx: PluginInput): ProjectRuntime { generation: crypto.randomUUID(), executions: new Map(), persistence: new Map(), + fences: new Map(), } runtimes.set(key, runtime) } + runtime.fences ??= new Map() return runtime } @@ -82,6 +91,7 @@ function stopLocalExecution(id: string, runtime: ProjectRuntime): ScheduledExecu export function beginSchedulerGeneration(ctx: PluginInput): string { const runtime = getRuntime(ctx) for (const id of Array.from(runtime.executions.keys())) stopLocalExecution(id, runtime) + for (const [id, fence] of runtime.fences) retireFence(id, runtime, fence) runtime.generation = crypto.randomUUID() return runtime.generation } @@ -126,11 +136,65 @@ function clearStateReminder(id: string, state: State): void { state.reminders.delete(id) } +function beginReconciliation(id: string, runtime: ProjectRuntime): { fence: ReminderFence; version: number } { + let fence = runtime.fences.get(id) + if (!fence) { + fence = { version: 0, reconciliations: 0, tombstone: false } + runtime.fences.set(id, fence) + } + fence.reconciliations++ + return { fence, version: fence.version } +} + +function reconciliationIsCurrent( + id: string, + runtime: ProjectRuntime, + fence: ReminderFence, + version: number, +): boolean { + return runtime.fences.get(id) === fence && fence.version === version && !fence.tombstone +} + +function finishReconciliation(id: string, runtime: ProjectRuntime, fence: ReminderFence): void { + fence.reconciliations-- + retireFence(id, runtime, fence) +} + +function invalidateReminder(id: string, runtime: ProjectRuntime): ReminderFence { + let fence = runtime.fences.get(id) + if (!fence) { + fence = { version: 0, reconciliations: 0, tombstone: false } + runtime.fences.set(id, fence) + } + fence.version++ + fence.tombstone = true + return fence +} + +function retireFence(id: string, runtime: ProjectRuntime, fence: ReminderFence): void { + if ( + runtime.fences.get(id) === fence + && fence.reconciliations === 0 + && !runtime.executions.has(id) + ) { + runtime.fences.delete(id) + } +} + async function deleteReminderLocked(id: string, ctx: PluginInput, state: State): Promise { - const execution = stopLocalExecution(id, getRuntime(ctx)) + const runtime = getRuntime(ctx) + const fence = invalidateReminder(id, runtime) + const execution = stopLocalExecution(id, runtime) clearStateReminder(id, state) if (execution && execution.state !== state) clearStateReminder(id, execution.state) - await deleteReminder(id, ctx) + try { + await deleteReminder(id, ctx) + } catch (error) { + fence.tombstone = false + throw error + } finally { + retireFence(id, runtime, fence) + } } export async function deleteStoredReminder(id: string, ctx: PluginInput, state: State): Promise { @@ -244,22 +308,46 @@ export async function reconcileReminder( options: ScheduleOptions = {}, ): Promise { if (!isCurrentGeneration(state, ctx)) return + const runtime = getRuntime(ctx) + const { fence, version } = beginReconciliation(id, runtime) try { - const stored = await loadReminder(id, ctx) - if (!isCurrentGeneration(state, ctx)) return - if (!stored || stored.status !== "active") { - clearStateReminder(id, state) - return - } - state.reminders.set(id, stored) - await scheduleTimer(stored, ctx, state, config, { - persist: false, - skipOverdue: options.skipOverdue, - retryAttempt: options.retryAttempt, + await loadReminder(id, ctx) + if ( + !isCurrentGeneration(state, ctx) + || !reconciliationIsCurrent(id, runtime, fence, version) + ) return + + await withMutationLock(id, ctx, async () => { + if ( + !isCurrentGeneration(state, ctx) + || !reconciliationIsCurrent(id, runtime, fence, version) + ) return + + // Re-read under the mutation lock so a cancellation in another process cannot + // be followed by restoration from the optimistic snapshot above. + const stored = await loadReminder(id, ctx) + if ( + !isCurrentGeneration(state, ctx) + || !reconciliationIsCurrent(id, runtime, fence, version) + ) return + if (!stored || stored.status !== "active") { + clearStateReminder(id, state) + return + } + state.reminders.set(id, stored) + await scheduleTimer(stored, ctx, state, config, { + persist: false, + skipOverdue: options.skipOverdue, + retryAttempt: options.retryAttempt, + }) }) } catch (error) { logger.error(`Failed to reconcile reminder ${id}:`, error) - scheduleReconciliation(id, ctx, state, config, options) + if (reconciliationIsCurrent(id, runtime, fence, version)) { + scheduleReconciliation(id, ctx, state, config, options) + } + } finally { + finishReconciliation(id, runtime, fence) } } @@ -478,14 +566,24 @@ export async function executeReminder( } export async function cancelReminder(id: string, ctx: PluginInput, state: State): Promise { - const execution = stopLocalExecution(id, getRuntime(ctx)) + const runtime = getRuntime(ctx) + const fence = invalidateReminder(id, runtime) + const execution = stopLocalExecution(id, runtime) clearStateReminder(id, state) if (execution && execution.state !== state) clearStateReminder(id, execution.state) try { - await withMutationLock(id, ctx, async () => deleteReminder(id, ctx)) + await withMutationLock(id, ctx, async () => { + await deleteReminder(id, ctx) + const replacement = stopLocalExecution(id, runtime) + clearStateReminder(id, state) + if (replacement && replacement.state !== state) clearStateReminder(id, replacement.state) + }) + retireFence(id, runtime, fence) logger.info(`Reminder ${id} cancelled`) return true } catch (error) { + fence.tombstone = false + retireFence(id, runtime, fence) logger.error(`Failed to cancel reminder ${id}:`, error) throw error } diff --git a/test/README.md b/test/README.md index 0735b29..cc7fbaa 100644 --- a/test/README.md +++ b/test/README.md @@ -9,5 +9,8 @@ prompt contention. Persistence-failure tests verify startup skip-overdue retenti rates while the prior JSON remains readable. Child-process tests use strict stdin barriers and are responsible for closing their children during cleanup. +Cancellation tests also pause a completed reconciliation read, finish durable cancellation, and +then release the stale continuation to verify it cannot restore local state or a timer. + 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/scheduler.test.ts b/test/scheduler.test.ts index 4d2078a..30315ee 100644 --- a/test/scheduler.test.ts +++ b/test/scheduler.test.ts @@ -1,4 +1,5 @@ -import { test, expect, describe, beforeEach, afterEach } from "bun:test" +import { test, expect, describe, beforeEach, afterEach, spyOn } from "bun:test" +import * as fsPromises from "node:fs/promises" import { beginSchedulerGeneration, waitForSchedulerMutations, @@ -596,6 +597,55 @@ describe("Scheduler", () => { expect(state.timers.has(snapshot.id)).toBe(true) }) + test("completed reconciliation read cannot resurrect state after cancellation returns", async () => { + const reminder: Reminder = { + id: "rem-reconcile-cancel-fence", + sessionID: "ses-test", + projectID: state.projectID, + type: "recurring", + interval: 60000, + originalPrompt: "must remain cancelled", + userDescription: "Reconciliation cancellation fence", + time: { created: Date.now(), nextExecution: Date.now() + 60000 }, + status: "active", + } + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + await scheduleTimer(reminder, ctx, state, config, { persist: false }) + + const readCompleted = deferred() + const releaseRead = deferred() + const readFile = fsPromises.readFile + let intercepted = false + const read = spyOn(fsPromises, "readFile").mockImplementation((async (...args: any[]) => { + const result = await (readFile as any)(...args) + if (!intercepted && String(args[0]).endsWith(`${reminder.id}.json`)) { + intercepted = true + readCompleted.resolve() + await releaseRead.promise + } + return result + }) as any) + + try { + const reconciliation = reconcileReminder(reminder.id, ctx, state, config) + await readCompleted.promise + await cancelReminder(reminder.id, ctx, state) + + expect(state.reminders.has(reminder.id)).toBe(false) + expect(state.timers.has(reminder.id)).toBe(false) + releaseRead.resolve() + await reconciliation + } finally { + releaseRead.resolve() + read.mockRestore() + } + + expect(await loadReminder(reminder.id, ctx)).toBeNull() + expect(state.reminders.has(reminder.id)).toBe(false) + expect(state.timers.has(reminder.id)).toBe(false) + }) + test("toast failure after a successful transition does not cancel recurrence", async () => { const reminder: Reminder = { id: "rem-toast-failure", From 1936f495e396c9bcb93a201adee3cd3bd271d650 Mon Sep 17 00:00:00 2001 From: Marc Warne Date: Sat, 25 Jul 2026 17:20:11 +0100 Subject: [PATCH 5/5] fix: restore reminders after cancellation failure --- README.md | 4 +++ index.ts | 4 +-- scheduler.ts | 57 +++++++++++++++++++++++++++++++--------- test/README.md | 2 ++ test/integration.test.ts | 2 +- test/scheduler.test.ts | 57 +++++++++++++++++++++++++++++++++++----- tools/reminderremove.ts | 10 ++++--- 7 files changed, 110 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index 7309fd8..58812b4 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,10 @@ Reconciliation reads are fenced per reminder and revalidated under the mutation cancellation returns, a read that completed before deletion cannot restore stale list state or a timer, even if its continuation resumes later. +If durable cancellation fails, the plugin re-reads authoritative state through the same fence and +lock protocol, restores a delayed timer, and returns the original deletion error. The reminder stays +visible and can be cancelled again without restarting the server. + 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 diff --git a/index.ts b/index.ts index 243cb94..97c2a40 100644 --- a/index.ts +++ b/index.ts @@ -130,7 +130,7 @@ const RemindersPlugin: Plugin = async (ctx) => { let cancelledCount = 0 for (const reminder of remindersToCancel) { - if (await cancelReminder(reminder.id, ctx, state)) cancelledCount++ + if (await cancelReminder(reminder.id, ctx, state, config)) cancelledCount++ } logger.info(`[RemindersPlugin] Cancelled ${cancelledCount} reminders for session ${sessionID}`) @@ -140,7 +140,7 @@ const RemindersPlugin: Plugin = async (ctx) => { tool: { reminderadd: createReminderAddTool(ctx, state, () => config), reminderlist: createReminderListTool(state), - reminderremove: createReminderRemoveTool(ctx, state), + reminderremove: createReminderRemoveTool(ctx, state, () => config), }, } } diff --git a/scheduler.ts b/scheduler.ts index f2a1317..a7a9513 100644 --- a/scheduler.ts +++ b/scheduler.ts @@ -22,6 +22,7 @@ type ProjectRuntime = { type ReminderFence = { version: number reconciliations: number + deletions: number tombstone: boolean } @@ -29,6 +30,7 @@ type ScheduleOptions = { persist?: boolean skipOverdue?: boolean retryAttempt?: number + minimumDelayMs?: number } const RECONCILIATION_DELAY_MS = 250 @@ -57,6 +59,7 @@ function getRuntime(ctx: PluginInput): ProjectRuntime { runtimes.set(key, runtime) } runtime.fences ??= new Map() + for (const fence of runtime.fences.values()) fence.deletions ??= 0 return runtime } @@ -139,7 +142,7 @@ function clearStateReminder(id: string, state: State): void { function beginReconciliation(id: string, runtime: ProjectRuntime): { fence: ReminderFence; version: number } { let fence = runtime.fences.get(id) if (!fence) { - fence = { version: 0, reconciliations: 0, tombstone: false } + fence = { version: 0, reconciliations: 0, deletions: 0, tombstone: false } runtime.fences.set(id, fence) } fence.reconciliations++ @@ -160,21 +163,31 @@ function finishReconciliation(id: string, runtime: ProjectRuntime, fence: Remind retireFence(id, runtime, fence) } -function invalidateReminder(id: string, runtime: ProjectRuntime): ReminderFence { +function invalidateReminder( + id: string, + runtime: ProjectRuntime, +): { fence: ReminderFence; version: number } { let fence = runtime.fences.get(id) if (!fence) { - fence = { version: 0, reconciliations: 0, tombstone: false } + fence = { version: 0, reconciliations: 0, deletions: 0, tombstone: false } runtime.fences.set(id, fence) } fence.version++ + fence.deletions++ fence.tombstone = true - return fence + return { fence, version: fence.version } +} + +function finishInvalidation(id: string, runtime: ProjectRuntime, fence: ReminderFence): void { + fence.deletions-- + retireFence(id, runtime, fence) } function retireFence(id: string, runtime: ProjectRuntime, fence: ReminderFence): void { if ( runtime.fences.get(id) === fence && fence.reconciliations === 0 + && fence.deletions === 0 && !runtime.executions.has(id) ) { runtime.fences.delete(id) @@ -183,17 +196,17 @@ function retireFence(id: string, runtime: ProjectRuntime, fence: ReminderFence): async function deleteReminderLocked(id: string, ctx: PluginInput, state: State): Promise { const runtime = getRuntime(ctx) - const fence = invalidateReminder(id, runtime) + const { fence, version } = invalidateReminder(id, runtime) const execution = stopLocalExecution(id, runtime) clearStateReminder(id, state) if (execution && execution.state !== state) clearStateReminder(id, execution.state) try { await deleteReminder(id, ctx) } catch (error) { - fence.tombstone = false + if (runtime.fences.get(id) === fence && fence.version === version) fence.tombstone = false throw error } finally { - retireFence(id, runtime, fence) + finishInvalidation(id, runtime, fence) } } @@ -243,7 +256,7 @@ export async function scheduleTimer( } const now = Date.now() - const delay = Math.max(0, reminder.time.nextExecution - now) + const delay = Math.max(options.minimumDelayMs ?? 0, reminder.time.nextExecution - now, 0) const skipOverdue = options.skipOverdue === true && reminder.time.nextExecution < now const retryAttempt = reminder.time.nextExecution < now ? options.retryAttempt ?? 0 : 0 const timer = setTimeout(async () => { @@ -292,6 +305,7 @@ function scheduleReconciliation( await reconcileReminder(id, ctx, state, config, { skipOverdue: options.skipOverdue, retryAttempt, + minimumDelayMs: options.minimumDelayMs, }) }, delay) timer.unref() @@ -339,6 +353,7 @@ export async function reconcileReminder( persist: false, skipOverdue: options.skipOverdue, retryAttempt: options.retryAttempt, + minimumDelayMs: options.minimumDelayMs, }) }) } catch (error) { @@ -565,10 +580,16 @@ export async function executeReminder( } } -export async function cancelReminder(id: string, ctx: PluginInput, state: State): Promise { +export async function cancelReminder( + id: string, + ctx: PluginInput, + state: State, + config: PluginConfig, +): Promise { const runtime = getRuntime(ctx) - const fence = invalidateReminder(id, runtime) + const { fence, version } = invalidateReminder(id, runtime) const execution = stopLocalExecution(id, runtime) + const rollbackState = execution?.state ?? state clearStateReminder(id, state) if (execution && execution.state !== state) clearStateReminder(id, execution.state) try { @@ -578,13 +599,23 @@ export async function cancelReminder(id: string, ctx: PluginInput, state: State) clearStateReminder(id, state) if (replacement && replacement.state !== state) clearStateReminder(id, replacement.state) }) - retireFence(id, runtime, fence) + finishInvalidation(id, runtime, fence) logger.info(`Reminder ${id} cancelled`) return true } catch (error) { - fence.tombstone = false - retireFence(id, runtime, fence) + const ownsFence = runtime.fences.get(id) === fence && fence.version === version + if (ownsFence) fence.tombstone = false + finishInvalidation(id, runtime, fence) logger.error(`Failed to cancel reminder ${id}:`, error) + if (ownsFence) { + try { + await reconcileReminder(id, ctx, rollbackState, config, { + minimumDelayMs: RECONCILIATION_DELAY_MS, + }) + } catch (rollbackError) { + logger.error(`Failed to restore reminder ${id} after cancellation failure:`, rollbackError) + } + } throw error } } diff --git a/test/README.md b/test/README.md index cc7fbaa..063eeae 100644 --- a/test/README.md +++ b/test/README.md @@ -11,6 +11,8 @@ responsible for closing their children during cleanup. Cancellation tests also pause a completed reconciliation read, finish durable cancellation, and then release the stale continuation to verify it cannot restore local state or a timer. +Failed-deletion coverage verifies that authoritative state and a future timer are restored before +the original error is returned, allowing a subsequent cancellation attempt to succeed. 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/integration.test.ts b/test/integration.test.ts index 3d693b7..fc1110d 100644 --- a/test/integration.test.ts +++ b/test/integration.test.ts @@ -264,7 +264,7 @@ describe("Integration Tests", () => { await waitFor(async () => (await loadReminder(reminder.id, ctx))!.time.nextExecution > Date.now()) expect(prompts).toBeLessThanOrEqual(4) expect(state.timers.has(reminder.id)).toBe(true) - await cancelReminder(reminder.id, ctx, state) + await cancelReminder(reminder.id, ctx, state, config) }) test("reminderlist tool returns empty for new session", async () => { diff --git a/test/scheduler.test.ts b/test/scheduler.test.ts index 30315ee..253e7d1 100644 --- a/test/scheduler.test.ts +++ b/test/scheduler.test.ts @@ -164,19 +164,62 @@ describe("Scheduler", () => { expect(state.timers.has(reminder.id)).toBe(true) expect(state.reminders.has(reminder.id)).toBe(true) - await cancelReminder(reminder.id, ctx, state) + await cancelReminder(reminder.id, ctx, state, config) expect(state.timers.has(reminder.id)).toBe(false) expect(state.reminders.has(reminder.id)).toBe(false) }) test("cancelReminder handles non-existent reminder gracefully", async () => { - await cancelReminder("non-existent", ctx, state) + await cancelReminder("non-existent", ctx, state, config) expect(state.timers.size).toBe(0) expect(state.reminders.size).toBe(0) }) + test("failed durable cancellation restores authoritative state for retry", async () => { + const reminder: Reminder = { + id: "rem-cancel-delete-failure", + sessionID: "ses-test", + projectID: state.projectID, + type: "recurring", + interval: 60000, + originalPrompt: "remain retryable", + userDescription: "Cancellation delete failure", + time: { created: Date.now(), nextExecution: Date.now() + 60000 }, + status: "active", + } + let prompts = 0 + ;(ctx.client.session.prompt as any) = async () => { + prompts++ + return { data: {}, error: undefined, response: {} } + } + state.reminders.set(reminder.id, reminder) + await scheduleTimer(reminder, ctx, state, config) + + const deletionError = Object.assign(new Error("simulated reminder deletion failure"), { code: "EIO" }) + const rmFile = fsPromises.rm + const remove = spyOn(fsPromises, "rm").mockImplementation((async (file: any, options: any) => { + if (String(file).endsWith(`${reminder.id}.json`)) throw deletionError + return rmFile(file, options) + }) as any) + try { + await expect(cancelReminder(reminder.id, ctx, state, config)).rejects.toBe(deletionError) + } finally { + remove.mockRestore() + } + + expect(await loadReminder(reminder.id, ctx)).toEqual(reminder) + expect(state.reminders.get(reminder.id)).toEqual(reminder) + expect(state.timers.has(reminder.id)).toBe(true) + expect(prompts).toBe(0) + + await cancelReminder(reminder.id, ctx, state, config) + expect(await loadReminder(reminder.id, ctx)).toBeNull() + expect(state.reminders.has(reminder.id)).toBe(false) + expect(state.timers.has(reminder.id)).toBe(false) + }) + test("executeReminder calls session prompt", async () => { let promptCalled = false let capturedPrompt = "" @@ -428,7 +471,7 @@ describe("Scheduler", () => { const execution = executeReminder(reminder, ctx, state, config) await promptStarted const lock = await acquireMutationLock(reminder.id, ctx) - const cancellation = cancelReminder(reminder.id, ctx, state) + const cancellation = cancelReminder(reminder.id, ctx, state, config) await lock.release() await cancellation finish() @@ -456,7 +499,7 @@ describe("Scheduler", () => { await executeReminder(reminder, ctx, state, config) expect(await loadReminder(reminder.id, ctx)).not.toBeNull() - await cancelReminder(reminder.id, ctx, state) + await cancelReminder(reminder.id, ctx, state, config) expect(await loadReminder(reminder.id, ctx)).toBeNull() expect(state.reminders.has(reminder.id)).toBe(false) expect(state.timers.has(reminder.id)).toBe(false) @@ -630,7 +673,7 @@ describe("Scheduler", () => { try { const reconciliation = reconcileReminder(reminder.id, ctx, state, config) await readCompleted.promise - await cancelReminder(reminder.id, ctx, state) + await cancelReminder(reminder.id, ctx, state, config) expect(state.reminders.has(reminder.id)).toBe(false) expect(state.timers.has(reminder.id)).toBe(false) @@ -724,7 +767,7 @@ describe("Scheduler", () => { await scheduleTimer(reminder, ctx, state, config) const signal = await promptStarted.promise - await cancelReminder(reminder.id, ctx, state) + await cancelReminder(reminder.id, ctx, state, config) finishPrompt.resolve() const directory = await getStorageDir(ctx) await waitFor(async () => { @@ -780,7 +823,7 @@ describe("Scheduler", () => { expect((await loadReminder(reminder.id, ctx))?.time.nextExecution).toBe( replacementReminder.time.nextExecution, ) - await cancelReminder(reminder.id, ctx, replacement) + await cancelReminder(reminder.id, ctx, replacement, config) }) test("forwards a captured agent and omits it for legacy reminders", async () => { diff --git a/tools/reminderremove.ts b/tools/reminderremove.ts index 7a2283b..5035c42 100644 --- a/tools/reminderremove.ts +++ b/tools/reminderremove.ts @@ -1,10 +1,14 @@ import { tool, type PluginInput } from "@opencode-ai/plugin" -import type { State } from "../types" +import type { PluginConfig, State } from "../types" import { cancelReminder } from "../scheduler" import DESCRIPTION from "./reminderremove.txt" import { logger } from "../logger" -export function createReminderRemoveTool(ctx: PluginInput, state: State) { +export function createReminderRemoveTool( + ctx: PluginInput, + state: State, + getConfig: () => PluginConfig, +) { return tool({ description: DESCRIPTION, @@ -33,7 +37,7 @@ export function createReminderRemoveTool(ctx: PluginInput, state: State) { // Cancel all matching reminders const cancelledDescriptions: string[] = [] for (const reminder of matches) { - if (await cancelReminder(reminder.id, ctx, state)) { + if (await cancelReminder(reminder.id, ctx, state, getConfig())) { cancelledDescriptions.push(reminder.userDescription) logger.info(`[RemindersPlugin] Cancelled reminder ${reminder.id} via user request`) }