From 99e8b13a534e04ee698be45b77f207a2e009f488 Mon Sep 17 00:00:00 2001 From: Marc Warne Date: Mon, 20 Jul 2026 14:38:38 +0100 Subject: [PATCH] fix cancellation races for active reminders --- index.ts | 68 ++++++-- scheduler.ts | 325 +++++++++++++++++++++++++++++++++++++-- test/integration.test.ts | 137 +++++++++++++++++ test/scheduler.test.ts | 200 +++++++++++++++++++++++- tools/reminderadd.ts | 6 +- tools/reminderremove.ts | 14 +- types.ts | 1 + 7 files changed, 716 insertions(+), 35 deletions(-) diff --git a/index.ts b/index.ts index 56a1c2e..ff90166 100644 --- a/index.ts +++ b/index.ts @@ -2,14 +2,23 @@ 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 { + beginSchedulerGeneration, + finishSchedulerGeneration, + isSchedulerGenerationCurrent, + waitForSchedulerPersistence, + scheduleTimer, + cancelReminder, + deleteStoredReminder, +} 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 +28,7 @@ const RemindersPlugin: Plugin = async (ctx) => { reminders: new Map(), timers: new Map(), projectID: project.id, + generation, } // Configuration with defaults @@ -38,46 +48,75 @@ const RemindersPlugin: Plugin = async (ctx) => { let invalidCount = 0 let healthyCount = 0 + if (!(await waitForSchedulerPersistence(ctx, generation))) { + return {} + } const storedReminders = await listReminders(ctx) + if (!isSchedulerGenerationCurrent(ctx, generation)) { + return {} + } for (const reminder of storedReminders) { + if (!isSchedulerGenerationCurrent(ctx, generation)) { + return {} + } try { ReminderSchema.parse(reminder) + } catch (error) { + logger.error(`[RemindersPlugin] Failed to restore reminder:`, error) + if (reminder.id) { + await deleteStoredReminder(reminder.id, ctx, state) + } + invalidCount++ + continue + } + try { // 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) + await deleteStoredReminder(reminder.id, ctx, state) expiredCount++ continue } state.reminders.set(reminder.id, reminder) - await scheduleTimer(reminder, ctx, state, config) + const scheduled = await scheduleTimer(reminder, ctx, state, config, { + persist: false, + cleanupOnPersistenceFailure: false, + }) + 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++ logger.info(`[RemindersPlugin] Restored and validated reminder ${reminder.id}`) } else { - await deleteReminder(reminder.id, ctx) + await deleteStoredReminder(reminder.id, ctx, state) 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) - } + // Keep a previously valid stored reminder after a transient scheduling or + // persistence failure so a later initialization can retry it. + logger.error(`[RemindersPlugin] Failed to schedule stored reminder ${reminder.id}:`, error) + state.reminders.delete(reminder.id) invalidCount++ } } + await finishSchedulerGeneration(ctx, generation) + if (!isSchedulerGenerationCurrent(ctx, generation)) { + return {} + } + logger.info( `[RemindersPlugin] Timer persistence validation completed: ${storedReminders.length} total, ${restoredCount} restored, ${expiredCount} expired, ${invalidCount} invalid, ${healthyCount} healthy`, ) @@ -85,7 +124,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 plugin generation clears timers and aborts active executions from the old generation // - Reminder state persists to storage and is restored on next startup // - Max reminders per project (50) bounds memory usage @@ -105,11 +144,14 @@ 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/scheduler.ts b/scheduler.ts index 94076b8..a45c072 100644 --- a/scheduler.ts +++ b/scheduler.ts @@ -3,19 +3,208 @@ import type { Reminder, State, PluginConfig } from "./types" import { saveReminder, deleteReminder } from "./storage" import { logger } from "./logger" +type ScheduledExecution = { + generation: string + token: string + controller?: AbortController + timer?: NodeJS.Timeout + state: State +} + +type ProjectRuntime = { + generation: string + initializing: boolean + executions: Map + persistence: Map> + cancelled: Map +} + +type ScheduleOptions = { + persist?: boolean + cleanupOnPersistenceFailure?: 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(), + initializing: false, + executions: new Map(), + persistence: new Map(), + cancelled: new Map(), + } + runtimes.set(key, runtime) + } + return runtime +} + +export function beginSchedulerGeneration(ctx: PluginInput): string { + const runtime = getRuntime(ctx) + for (const [id, execution] of runtime.executions) { + if (execution.timer) { + clearTimeout(execution.timer) + } + execution.controller?.abort() + execution.state.timers.delete(id) + } + runtime.executions.clear() + runtime.generation = crypto.randomUUID() + runtime.initializing = true + return runtime.generation +} + +export function isSchedulerGenerationCurrent(ctx: PluginInput, generation: string): boolean { + return getRuntime(ctx).generation === generation +} + +export async function waitForSchedulerPersistence(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 +} + +export async function finishSchedulerGeneration(ctx: PluginInput, generation: string): Promise { + const runtime = getRuntime(ctx) + await Promise.all(Array.from(runtime.persistence.values(), (operation) => operation.catch(() => {}))) + if (runtime.generation === generation) { + runtime.initializing = false + for (const [id, tombstone] of runtime.cancelled) { + if (tombstone.deleted) { + runtime.cancelled.delete(id) + } + } + } +} + +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 + ) +} + +async function queuePersistence( + id: string, + ctx: PluginInput, + operation: () => Promise, +): Promise { + const runtime = getRuntime(ctx) + const previous = runtime.persistence.get(id) ?? Promise.resolve() + const current = previous.catch(() => {}).then(operation) + runtime.persistence.set(id, current) + try { + await current + } finally { + if (runtime.persistence.get(id) === current) { + runtime.persistence.delete(id) + } + } +} + +async function persistCurrentReminder( + reminder: Reminder, + ctx: PluginInput, + state: State, + token: string, +): Promise { + const snapshot = structuredClone(reminder) + await queuePersistence(reminder.id, ctx, async () => { + if (isCurrentExecution(reminder.id, ctx, state, token)) { + await saveReminder(snapshot, ctx) + } + }) + return isCurrentExecution(reminder.id, ctx, state, token) +} + +async function deleteCurrentStoredReminder(id: string, ctx: PluginInput): Promise { + const runtime = getRuntime(ctx) + const tombstone = runtime.cancelled.get(id) ?? { deleted: false } + runtime.cancelled.set(id, tombstone) + await queuePersistence(id, ctx, async () => { + if (runtime.cancelled.get(id) === tombstone) { + await deleteReminder(id, ctx) + tombstone.deleted = true + } + }) + if (!runtime.initializing && tombstone.deleted && runtime.cancelled.get(id) === tombstone) { + runtime.cancelled.delete(id) + } +} + +export async function deleteStoredReminder(id: string, ctx: PluginInput, state: State): Promise { + if (!isCurrentGeneration(state, ctx)) { + return + } + await deleteCurrentStoredReminder(id, ctx) +} + export async function scheduleTimer( reminder: Reminder, ctx: PluginInput, state: State, config: PluginConfig, -): Promise { + options: ScheduleOptions = {}, +): Promise { + const runtime = getRuntime(ctx) + const generation = stateGeneration(state, ctx) + if (generation !== runtime.generation) { + state.reminders.delete(reminder.id) + return false + } + if (runtime.cancelled.has(reminder.id)) { + state.reminders.delete(reminder.id) + await deleteStoredReminder(reminder.id, ctx, state) + return false + } + const existingTimer = state.timers.get(reminder.id) if (existingTimer) { clearTimeout(existingTimer) } + const existingExecution = runtime.executions.get(reminder.id) + if (existingExecution?.timer) { + clearTimeout(existingExecution.timer) + existingExecution.state.timers.delete(reminder.id) + } + existingExecution?.controller?.abort() + const token = crypto.randomUUID() + const execution: ScheduledExecution = { generation, token, state } + runtime.executions.set(reminder.id, execution) + const now = Date.now() let delay = reminder.time.nextExecution - now + let shouldPersist = options.persist ?? true // Handle missed execution windows for recurring reminders if (delay < 0 && reminder.type === "recurring") { @@ -24,7 +213,7 @@ export async function scheduleTimer( // 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) + shouldPersist = true logger.info( `[RemindersPlugin] Skipped ${missedIntervals} missed execution(s) for recurring reminder ${reminder.id}`, ) @@ -33,17 +222,48 @@ export async function scheduleTimer( delay = Math.max(0, delay) } + try { + if (shouldPersist && !(await persistCurrentReminder(reminder, ctx, state, token))) { + return false + } + } catch (error) { + if (isCurrentExecution(reminder.id, ctx, state, token)) { + runtime.executions.delete(reminder.id) + state.reminders.delete(reminder.id) + if (options.cleanupOnPersistenceFailure !== false) { + try { + await deleteStoredReminder(reminder.id, ctx, state) + } catch (cleanupError) { + logger.error(`Failed to clean up reminder ${reminder.id} after persistence failure:`, cleanupError) + } + } + } + throw error + } + const timer = setTimeout(async () => { - state.timers.delete(reminder.id) - await executeReminder(reminder, ctx, state, config) + 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 + await executeReminder(reminder, ctx, state, config, token, controller.signal) }, delay) // CRITICAL: Allow process to exit even with active timers timer.unref() + execution.timer = timer state.timers.set(reminder.id, timer) logger.info(`Scheduled reminder ${reminder.id} to execute in ${Math.round(delay / 1000)}s`) + return true } export async function executeReminder( @@ -51,12 +271,37 @@ export async function executeReminder( ctx: PluginInput, state: State, config: PluginConfig, + token?: string, + signal?: AbortSignal, ): Promise { + if (!token) { + const runtime = getRuntime(ctx) + if (!isCurrentGeneration(state, ctx) || runtime.cancelled.has(reminder.id)) { + return + } + const existingExecution = runtime.executions.get(reminder.id) + if (existingExecution?.timer) { + clearTimeout(existingExecution.timer) + existingExecution.state.timers.delete(reminder.id) + } + existingExecution?.controller?.abort() + token = crypto.randomUUID() + const controller = new AbortController() + signal = controller.signal + runtime.executions.set(reminder.id, { + generation: stateGeneration(state, ctx), + token, + controller, + state, + }) + } + logger.info(`Executing reminder ${reminder.id}: ${reminder.userDescription}`) try { await ctx.client.session.prompt({ path: { id: reminder.sessionID }, + signal, body: { parts: [ { @@ -67,13 +312,23 @@ export async function executeReminder( }, }) + if (!isCurrentExecution(reminder.id, ctx, state, token)) { + return + } + reminder.time.lastExecution = Date.now() 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) + try { + if (!(await scheduleTimer(reminder, ctx, state, config))) { + return + } + } catch (error) { + logger.error(`Failed to reschedule recurring reminder ${reminder.id}:`, error) + throw error + } logger.info(`Recurring reminder ${reminder.id} rescheduled`) if (config.notifications.enabled) { @@ -85,7 +340,12 @@ export async function executeReminder( }) } } else { - await cancelReminder(reminder.id, ctx, state) + try { + await cancelReminder(reminder.id, ctx, state, token) + } catch (error) { + logger.error(`Failed to remove completed reminder ${reminder.id}:`, error) + throw error + } logger.info(`One-time reminder ${reminder.id} completed and removed`) if (config.notifications.enabled) { @@ -98,6 +358,10 @@ export async function executeReminder( } } } catch (error: any) { + if (!isCurrentExecution(reminder.id, ctx, state, token)) { + return + } + logger.error(`Reminder ${reminder.id} execution failed:`, error) if (config.notifications.enabled) { @@ -114,25 +378,53 @@ export async function executeReminder( } } + if (!isCurrentExecution(reminder.id, ctx, state, token)) { + 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) + try { + if (!(await scheduleTimer(reminder, ctx, state, config))) { + return + } + } catch (rescheduleError) { + logger.error(`Failed to reschedule aborted reminder ${reminder.id}:`, rescheduleError) + throw rescheduleError + } logger.info(`Recurring reminder ${reminder.id} rescheduled after abort`) } else { - await cancelReminder(reminder.id, ctx, state) + await cancelReminder(reminder.id, ctx, state, token) logger.info(`One-time reminder ${reminder.id} cancelled after abort`) } } else { - await cancelReminder(reminder.id, ctx, state) + await cancelReminder(reminder.id, ctx, state, token) logger.info(`Reminder ${reminder.id} cancelled due to error`) } } } -export async function cancelReminder(id: string, ctx: PluginInput, state: State): Promise { +export async function cancelReminder( + id: string, + ctx: PluginInput, + state: State, + token?: string, +): Promise { + const runtime = getRuntime(ctx) + const execution = runtime.executions.get(id) + if (token && execution?.token !== token) { + return false + } + + if (execution?.timer) { + clearTimeout(execution.timer) + execution.state.timers.delete(id) + } + execution?.controller?.abort() + runtime.executions.delete(id) + const timer = state.timers.get(id) if (timer) { clearTimeout(timer) @@ -141,7 +433,14 @@ export async function cancelReminder(id: string, ctx: PluginInput, state: State) state.reminders.delete(id) - await deleteReminder(id, ctx) + if (execution?.state !== state) { + execution?.state.reminders.delete(id) + } + + // Tool calls and events from a replaced plugin instance still express a current + // user/session cancellation. Token-bound callbacks remain fenced above. + await deleteCurrentStoredReminder(id, ctx) logger.info(`Reminder ${id} cancelled`) + return true } diff --git a/test/integration.test.ts b/test/integration.test.ts index 58a4324..3dfb9f8 100644 --- a/test/integration.test.ts +++ b/test/integration.test.ts @@ -3,6 +3,16 @@ import RemindersPlugin from "../index" import type { PluginInput } from "@opencode-ai/plugin" import { $ } from "bun" +function deferred() { + let resolve!: (value: T | PromiseLike) => void + let reject!: (reason?: unknown) => void + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise + reject = rejectPromise + }) + return { promise, resolve, reject } +} + async function createMockContext(tmpDir: string): Promise { const sessions = new Map() @@ -80,6 +90,133 @@ describe("Integration Tests", () => { expect(result).toContain("File change check") }) + test("replacement initialization drains an in-flight reminder save before restoring", async () => { + const plugin = await RemindersPlugin(ctx) + const writeStarted = deferred() + const allowWrite = deferred() + const originalWrite = Bun.write + let intercepted = false + + ;(Bun as any).write = async (destination: unknown, input: unknown, options?: unknown) => { + if (!intercepted && typeof destination === "string" && destination.endsWith(".json")) { + intercepted = true + writeStarted.resolve() + await allowWrite.promise + } + return (originalWrite as any)(destination, input, options) + } + + try { + const addPromise = plugin.tool!.reminderadd.execute( + { + interval_seconds: 60, + type: "one-time" as const, + action_prompt: "test generation handoff", + description: "Generation handoff reminder", + }, + { sessionID: "ses-generation-handoff" } as any, + ) + await writeStarted.promise + + let replacementResolved = false + const replacementPromise = RemindersPlugin(ctx).then((replacement) => { + replacementResolved = true + return replacement + }) + await Bun.sleep(10) + expect(replacementResolved).toBe(false) + + allowWrite.resolve() + const [addResult, replacement] = await Promise.all([addPromise, replacementPromise]) + expect(addResult).toContain("scheduler reloaded") + + const listResult = await replacement.tool!.reminderlist.execute( + {}, + { sessionID: "ses-generation-handoff" } as any, + ) + expect(listResult).toContain("Generation handoff reminder") + + await replacement.tool!.reminderremove.execute( + { description_pattern: "Generation handoff" }, + { sessionID: "ses-generation-handoff" } as any, + ) + } finally { + allowWrite.resolve() + ;(Bun as any).write = originalWrite + } + }) + + test("a stale reminderremove tool cancels the replacement generation reminder", async () => { + const stalePlugin = await RemindersPlugin(ctx) + await stalePlugin.tool!.reminderadd.execute( + { + interval_seconds: 60, + type: "recurring" as const, + action_prompt: "test stale cancellation", + description: "Stale generation cancellation", + }, + { sessionID: "ses-stale-cancellation" } as any, + ) + + const replacement = await RemindersPlugin(ctx) + const removeResult = await stalePlugin.tool!.reminderremove.execute( + { description_pattern: "Stale generation" }, + { sessionID: "ses-stale-cancellation" } as any, + ) + expect(removeResult).toContain("Reminder cancelled") + + const listResult = await replacement.tool!.reminderlist.execute( + {}, + { sessionID: "ses-stale-cancellation" } as any, + ) + expect(listResult).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: "test startup persistence failure", + description: "Startup persistence failure", + }, + { sessionID: "ses-startup-persistence" } as any, + ) + + const storageDir = `${tmpDir}/.opencode/reminders/test-project-integration` + const files: string[] = [] + for await (const file of new Bun.Glob("*.json").scan({ cwd: storageDir, absolute: true })) { + files.push(file) + } + expect(files).toHaveLength(1) + const stored = await Bun.file(files[0]).json() + stored.time.nextExecution = Date.now() - 1000 + await Bun.write(files[0], JSON.stringify(stored, null, 2)) + + const originalWrite = Bun.write + ;(Bun as any).write = async (destination: unknown, input: unknown, options?: unknown) => { + if (typeof destination === "string" && destination.endsWith(".json")) { + throw new Error("simulated reminder write failure") + } + return (originalWrite as any)(destination, input, options) + } + + try { + const replacement = await RemindersPlugin(ctx) + const listResult = await replacement.tool!.reminderlist.execute( + {}, + { sessionID: "ses-startup-persistence" } as any, + ) + expect(listResult).toContain("No active reminders") + } finally { + ;(Bun as any).write = originalWrite + } + + expect(await Bun.file(files[0]).exists()).toBe(true) + expect((await Bun.file(files[0]).json()).id).toBe(stored.id) + }) + 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 0d8533a..8a38cfb 100644 --- a/test/scheduler.test.ts +++ b/test/scheduler.test.ts @@ -1,8 +1,25 @@ import { test, expect, describe, beforeEach, afterEach } from "bun:test" -import { scheduleTimer, executeReminder, cancelReminder } from "../scheduler" +import { + beginSchedulerGeneration, + scheduleTimer, + executeReminder, + cancelReminder, + deleteStoredReminder, +} from "../scheduler" import type { Reminder, State, PluginConfig } from "../types" import type { PluginInput } from "@opencode-ai/plugin" import { $ } from "bun" +import { loadReminder, saveReminder } from "../storage" + +function deferred() { + let resolve!: (value: T | PromiseLike) => void + let reject!: (reason?: unknown) => void + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise + reject = rejectPromise + }) + return { promise, resolve, reject } +} async function createMockContext(tmpDir: string): Promise { return { @@ -148,6 +165,33 @@ describe("Scheduler", () => { expect(state.reminders.size).toBe(0) }) + test("successful cancellation retires its tombstone", async () => { + const reminder: Reminder = { + id: "rem-tombstone-retirement", + sessionID: "ses-test", + projectID: "test-project-123", + type: "one-time", + interval: 5000, + originalPrompt: "test", + userDescription: "Tombstone retirement test", + time: { + created: Date.now(), + nextExecution: Date.now() + 5000, + }, + status: "active", + } + + state.reminders.set(reminder.id, reminder) + await scheduleTimer(reminder, ctx, state, config) + await cancelReminder(reminder.id, ctx, state) + + reminder.time.nextExecution = Date.now() + 5000 + state.reminders.set(reminder.id, reminder) + await scheduleTimer(reminder, ctx, state, config) + + expect(state.timers.has(reminder.id)).toBe(true) + }) + test("executeReminder calls session prompt", async () => { let promptCalled = false let capturedPrompt = "" @@ -259,6 +303,160 @@ describe("Scheduler", () => { expect(state.timers.has(reminder.id)).toBe(true) }) + test("cancelReminder aborts an active execution without rescheduling it", async () => { + const promptStarted = deferred() + const finishPrompt = deferred() + const promptFinished = deferred() + ;(ctx.client.session.prompt as any) = async (opts: any) => { + promptStarted.resolve(opts.signal) + await finishPrompt.promise + promptFinished.resolve() + return { data: {} as any, error: undefined, response: {} as any } + } + + const reminder: Reminder = { + id: "rem-active-cancel", + sessionID: "ses-test", + projectID: "test-project-123", + type: "recurring", + interval: 1000, + originalPrompt: "test", + userDescription: "Active cancellation test", + time: { + created: Date.now(), + nextExecution: Date.now() + 5, + }, + status: "active", + } + + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + await scheduleTimer(reminder, ctx, state, config) + + const signal = await promptStarted.promise + await cancelReminder(reminder.id, ctx, state) + finishPrompt.resolve() + await promptFinished.promise + await Bun.sleep(0) + + expect(signal.aborted).toBe(true) + expect(state.reminders.has(reminder.id)).toBe(false) + expect(state.timers.has(reminder.id)).toBe(false) + expect(await loadReminder(reminder.id, ctx)).toBeNull() + }) + + test("a replacement plugin instance invalidates an active old execution", async () => { + const promptStarted = deferred() + const finishPrompt = deferred() + const promptFinished = deferred() + ;(ctx.client.session.prompt as any) = async (opts: any) => { + promptStarted.resolve(opts.signal) + await finishPrompt.promise + promptFinished.resolve() + return { data: {} as any, error: undefined, response: {} as any } + } + + const reminder: Reminder = { + id: "rem-hot-reload", + sessionID: "ses-test", + projectID: "test-project-123", + type: "recurring", + interval: 1000, + originalPrompt: "test", + userDescription: "Hot reload test", + time: { + created: Date.now(), + nextExecution: Date.now() + 5, + }, + status: "active", + } + + state.reminders.set(reminder.id, reminder) + await saveReminder(reminder, ctx) + await scheduleTimer(reminder, ctx, state, config) + const oldSignal = await promptStarted.promise + + const replacementState: State = { + reminders: new Map(), + timers: new Map(), + projectID: state.projectID, + generation: beginSchedulerGeneration(ctx), + } + const replacementReminder = structuredClone(reminder) + replacementReminder.time.nextExecution = Date.now() + 1000 + replacementState.reminders.set(replacementReminder.id, replacementReminder) + await scheduleTimer(replacementReminder, ctx, replacementState, config) + + finishPrompt.resolve() + await promptFinished.promise + await Bun.sleep(0) + + expect(oldSignal.aborted).toBe(true) + expect(state.timers.has(reminder.id)).toBe(false) + expect(replacementState.timers.has(reminder.id)).toBe(true) + expect((await loadReminder(reminder.id, ctx))?.time.nextExecution).toBe( + replacementReminder.time.nextExecution, + ) + + await cancelReminder(reminder.id, ctx, replacementState) + }) + + test("a stale plugin generation cannot delete replacement persistence", async () => { + const reminder: Reminder = { + id: "rem-stale-delete", + sessionID: "ses-test", + projectID: "test-project-123", + type: "one-time", + interval: 1000, + originalPrompt: "test", + userDescription: "Stale delete test", + time: { + created: Date.now(), + nextExecution: Date.now() + 1000, + }, + status: "active", + } + + state.generation = beginSchedulerGeneration(ctx) + await saveReminder(reminder, ctx) + + const replacementState: State = { + reminders: new Map([[reminder.id, reminder]]), + timers: new Map(), + projectID: state.projectID, + generation: beginSchedulerGeneration(ctx), + } + await deleteStoredReminder(reminder.id, ctx, state) + + expect(await loadReminder(reminder.id, ctx)).toEqual(reminder) + await cancelReminder(reminder.id, ctx, replacementState) + }) + + test("beginSchedulerGeneration clears old future timers", async () => { + const reminder: Reminder = { + id: "rem-old-future-timer", + sessionID: "ses-test", + projectID: "test-project-123", + type: "one-time", + interval: 60000, + originalPrompt: "test", + userDescription: "Old future timer test", + time: { + created: Date.now(), + nextExecution: Date.now() + 60000, + }, + status: "active", + } + + state.reminders.set(reminder.id, reminder) + await scheduleTimer(reminder, ctx, state, config) + expect(state.timers.has(reminder.id)).toBe(true) + + beginSchedulerGeneration(ctx) + + expect(state.timers.has(reminder.id)).toBe(false) + }) + test("scheduleTimer maintains cadence for missed recurring reminders", async () => { const interval = 3600000 // 1 hour const originalScheduledTime = Date.now() - 5400000 // 1.5 hours ago diff --git a/tools/reminderadd.ts b/tools/reminderadd.ts index 9e6a9d3..ee3d708 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" @@ -50,8 +49,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..104f65c 100644 --- a/types.ts +++ b/types.ts @@ -22,6 +22,7 @@ export type State = { reminders: Map timers: Map projectID: string + generation?: string } export type PluginConfig = {