fix: coordinate reminders across processes

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/<pid>/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:

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

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<void> }
+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<string | null> {
+  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<boolean> {
+  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<boolean> {
+  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<Owner | null> {
+  try {
+    return parseOwner(await readFile(file, "utf8"))
+  } catch {
+    return null
+  }
+}
+
+async function recoverDeadOwnerFile(file: string): Promise<boolean> {
+  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<boolean> {
+  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<LeaseAcquisition> {
+  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<Owner> {
+  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<LeaseAcquisition> {
+  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<ExecutionLease> {
+  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<void>((resolve) => setTimeout(resolve, Math.min(MUTATION_LOCK_RETRY_MS, deadline - Date.now())))
+  }
+  throw new Error(`Timed out acquiring mutation lock for reminder ${reminderID}`)
+}

package.json

@@ -23,6 +23,7 @@
     "types.ts",
     "storage.ts",
     "scheduler.ts",
+    "leases.ts",
     "logger.ts",
     "tools/",
     "README.md"