diff --git a/server/src/conversation/frontend-reminders.mjs b/server/src/conversation/frontend-reminders.mjs new file mode 100644 index 00000000..a030a9ac --- /dev/null +++ b/server/src/conversation/frontend-reminders.mjs @@ -0,0 +1,694 @@ +import { createHash, randomUUID } from 'node:crypto' +import { + chmodSync, + mkdirSync, + readFileSync, + renameSync, + statSync, + writeFileSync, +} from 'node:fs' +import { dirname } from 'node:path' +import { + replaceFileSync, + withFileTransaction, +} from '../../../shared/file-transaction-lock.mjs' + +export const REMINDER_STORE_VERSION = 1 +export const MAX_REMINDERS_PER_OWNER = 100 +export const MAX_REMINDER_TEXT_CHARS = 500 +export const MAX_REMINDER_OWNER_ID_CHARS = 200 +export const MAX_REMINDER_TIMEZONE_CHARS = 100 + +export const ReminderKind = Object.freeze({ + REMINDER: 'reminder', + TASK: 'task', +}) + +export const ReminderRecurrence = Object.freeze({ + ONCE: 'once', + DAILY: 'daily', + WEEKLY: 'weekly', + WEEKDAYS: 'weekdays', +}) + +export const ReminderStatus = Object.freeze({ + ACTIVE: 'active', + FIRING: 'firing', + COMPLETED: 'completed', + CANCELLED: 'cancelled', + FAILED: 'failed', +}) + +const KINDS = new Set(Object.values(ReminderKind)) +const RECURRENCES = new Set(Object.values(ReminderRecurrence)) +const STATUSES = new Set(Object.values(ReminderStatus)) +const TERMINAL_STATUSES = new Set([ + ReminderStatus.COMPLETED, + ReminderStatus.CANCELLED, + ReminderStatus.FAILED, +]) + +function clean(value, maxChars) { + return [...String(value ?? '').replace(/\s+/g, ' ').trim()] + .slice(0, maxChars) + .join('') +} + +function plainObject(value) { + return Boolean(value) && typeof value === 'object' && !Array.isArray(value) +} + +function timestamp(value, code = 'invalid_execute_at') { + const number = Number(value) + if (!Number.isSafeInteger(number) || number < 0) { + throw new ReminderStoreError(code, 'reminder time must be a non-negative integer') + } + return number +} + +function ownerId(value) { + const normalized = clean(value, MAX_REMINDER_OWNER_ID_CHARS) + if (!normalized) { + throw new ReminderStoreError('invalid_owner', 'reminder owner is required') + } + if (String(value ?? '').trim().length > MAX_REMINDER_OWNER_ID_CHARS) { + throw new ReminderStoreError('invalid_owner', 'reminder owner is too long') + } + return normalized +} + +function reminderId(value) { + const source = String(value ?? '').trim() + const normalized = clean(source, 120) + if (!normalized || [...source].length > 120) { + throw new ReminderStoreError('invalid_id', 'reminder id is invalid') + } + return normalized +} + +function reminderText(value) { + const source = String(value ?? '').trim() + const normalized = clean(source, MAX_REMINDER_TEXT_CHARS) + if (!normalized) { + throw new ReminderStoreError('invalid_text', 'reminder text is required') + } + if ([...source].length > MAX_REMINDER_TEXT_CHARS) { + throw new ReminderStoreError('invalid_text', 'reminder text is too long') + } + return normalized +} + +function reminderKind(value) { + const normalized = String(value || ReminderKind.REMINDER).trim().toLowerCase() + if (!KINDS.has(normalized)) { + throw new ReminderStoreError('invalid_kind', `unsupported reminder kind: ${normalized}`) + } + return normalized +} + +function recurrence(value) { + const normalized = String(value || ReminderRecurrence.ONCE).trim().toLowerCase() + if (!RECURRENCES.has(normalized)) { + throw new ReminderStoreError( + 'invalid_recurrence', + `unsupported reminder recurrence: ${normalized}`, + ) + } + return normalized +} + +function timezone(value) { + const source = String(value || 'UTC').trim() + if (!source || source.length > MAX_REMINDER_TIMEZONE_CHARS) { + throw new ReminderStoreError('invalid_timezone', 'reminder timezone is invalid') + } + try { + new Intl.DateTimeFormat('en-US', { timeZone: source }).format(0) + } catch { + throw new ReminderStoreError('invalid_timezone', `unknown reminder timezone: ${source}`) + } + return source +} + +function cloneReminder(reminder) { + return reminder ? { ...reminder } : null +} + +export class ReminderStoreError extends Error { + constructor(code, message) { + super(message) + this.name = 'ReminderStoreError' + this.code = code + } +} + +/** + * Durable, owner-scoped storage for frontend reminders. + * + * This store deliberately does not know about the Realtime provider, the + * TaskManager, or any backend Agent. A later runtime can use claimDue() and + * complete()/fail() to connect the store to announcements or WorkSubmissionPort + * without putting scheduling policy into persistence. + */ +export class FrontendReminderStore { + constructor({ + filePath = null, + maxOwners = 1000, + maxRemindersPerOwner = MAX_REMINDERS_PER_OWNER, + ownerTtlMs = 0, + now = () => Date.now(), + idFactory = () => `rem_${randomUUID()}`, + onWarning = warning => console.warn(warning.message), + } = {}) { + this.filePath = filePath + this.maxOwners = Math.max(1, Number(maxOwners) || 1000) + this.maxRemindersPerOwner = Math.max( + 1, + Number(maxRemindersPerOwner) || MAX_REMINDERS_PER_OWNER, + ) + this.ownerTtlMs = Math.max(0, Number(ownerTtlMs) || 0) + this.now = now + this.idFactory = idFactory + this.onWarning = onWarning + this.reminders = new Map() + this.ownerAccess = new Map() + this.warning = null + this.persistenceDisabled = false + this.loadedMtimeMs = 0 + this.loadedContentHash = '' + if (filePath) this.load() + } + + fileMtimeMs() { + try { + return statSync(this.filePath).mtimeMs + } catch (error) { + if (error.code === 'ENOENT') return 0 + throw error + } + } + + fileContentHash() { + try { + return createHash('sha1').update(readFileSync(this.filePath)).digest('hex') + } catch { + return '' + } + } + + refreshIfChanged() { + if (!this.filePath || this.persistenceDisabled) return + const mtimeMs = this.fileMtimeMs() + if (mtimeMs === this.loadedMtimeMs + && this.fileContentHash() === this.loadedContentHash) return + this.reminders = new Map() + this.ownerAccess = new Map() + this.load() + } + + writeTransaction(action) { + return withFileTransaction(this.filePath, () => { + // Reload after acquiring the lock. This prevents a second Gateway from + // having its write silently overwritten by this instance's stale cache. + if (this.filePath && !this.persistenceDisabled) { + this.reminders = new Map() + this.ownerAccess = new Map() + this.load() + } + return action() + }) + } + + setWarning(message, quarantinePath = null) { + this.warning = { message, quarantinePath, at: this.now() } + try { + this.onWarning?.(this.warning) + } catch { + // Diagnostics must never prevent the voice service from starting. + } + } + + disablePersistence(message) { + this.persistenceDisabled = true + this.setWarning(`${message}；已禁用提醒持久化，服务将继续运行。`) + } + + quarantine(reason) { + const quarantinePath = `${this.filePath}.corrupt-${this.now()}` + try { + renameSync(this.filePath, quarantinePath) + this.setWarning( + `${reason}；原文件已隔离为 ${quarantinePath}，服务将使用空提醒继续运行。`, + quarantinePath, + ) + } catch (error) { + this.persistenceDisabled = true + this.setWarning( + `${reason}；隔离失败（${error.message}），已禁用提醒持久化以保护原文件。`, + ) + } + } + + normalizePersisted(raw, storedOwnerId) { + if (!plainObject(raw)) return null + let id + let text + let kind + let repeat + let zone + let nextFireAt + try { + id = reminderId(raw.id) + text = reminderText(raw.text) + kind = reminderKind(raw.kind) + repeat = recurrence(raw.recurrence) + zone = timezone(raw.timezone) + nextFireAt = timestamp(raw.nextFireAt) + } catch { + return null + } + const status = STATUSES.has(raw.status) + ? raw.status + : ReminderStatus.ACTIVE + const createdAt = Number.isSafeInteger(raw.createdAt) && raw.createdAt >= 0 + ? raw.createdAt + : this.now() + const updatedAt = Number.isSafeInteger(raw.updatedAt) && raw.updatedAt >= 0 + ? raw.updatedAt + : createdAt + const lastFiredAt = Number.isSafeInteger(raw.lastFiredAt) && raw.lastFiredAt >= 0 + ? raw.lastFiredAt + : null + const fireCount = Number.isSafeInteger(raw.fireCount) && raw.fireCount >= 0 + ? raw.fireCount + : 0 + const lastError = raw.lastError + ? clean(raw.lastError, 500) + : null + let safeOwnerId + try { + safeOwnerId = ownerId(storedOwnerId) + } catch { + return null + } + return { + id, + ownerId: safeOwnerId, + text, + kind, + timezone: zone, + recurrence: repeat, + nextFireAt, + // A process can die after claiming but before delivering. Re-opening a + // firing reminder as active makes it retryable instead of losing it. + status: status === ReminderStatus.FIRING + ? ReminderStatus.ACTIVE + : status, + createdAt, + updatedAt, + lastFiredAt, + fireCount, + lastError, + } + } + + load() { + if (!this.filePath || this.persistenceDisabled) return + let raw + try { + raw = readFileSync(this.filePath, 'utf8') + } catch (error) { + if (error.code === 'ENOENT') { + this.loadedMtimeMs = 0 + this.loadedContentHash = '' + return + } + this.disablePersistence(`无法读取提醒文件：${error.message}`) + return + } + this.loadedMtimeMs = this.fileMtimeMs() + this.loadedContentHash = createHash('sha1').update(raw).digest('hex') + let parsed + try { + parsed = JSON.parse(raw) + } catch (error) { + this.quarantine(`提醒文件不是有效的 JSON：${error.message}`) + return + } + if ( + !plainObject(parsed) + || parsed.version !== REMINDER_STORE_VERSION + || !plainObject(parsed.owners) + || !plainObject(parsed.ownerAccess) + ) { + this.quarantine('提醒文件格式或版本无效') + return + } + Object.entries(parsed.owners).slice(0, this.maxOwners).forEach(([storedOwnerId, entries]) => { + if (!plainObject(entries)) return + const ownerReminders = new Map() + Object.entries(entries) + .slice(0, this.maxRemindersPerOwner) + .forEach(([storedId, rawReminder]) => { + const reminder = this.normalizePersisted( + { ...rawReminder, id: rawReminder?.id || storedId }, + storedOwnerId, + ) + if (!reminder || ownerReminders.has(reminder.id)) return + ownerReminders.set(reminder.id, reminder) + }) + if (!ownerReminders.size) return + let safeOwnerId + try { + safeOwnerId = ownerId(storedOwnerId) + } catch { + return + } + this.reminders.set(safeOwnerId, ownerReminders) + const access = Number(parsed.ownerAccess[storedOwnerId]) + this.ownerAccess.set( + safeOwnerId, + Number.isSafeInteger(access) && access >= 0 ? access : this.now(), + ) + }) + this.pruneOwners({ persist: false }) + } + + persist() { + if (!this.filePath) return true + if (this.persistenceDisabled) return false + try { + const owners = Object.create(null) + this.reminders.forEach((entries, safeOwnerId) => { + owners[safeOwnerId] = Object.create(null) + entries.forEach((reminder, id) => { + owners[safeOwnerId][id] = reminder + }) + }) + mkdirSync(dirname(this.filePath), { recursive: true, mode: 0o700 }) + const temporary = `${this.filePath}.${process.pid}.tmp` + const body = `${JSON.stringify({ + version: REMINDER_STORE_VERSION, + owners, + ownerAccess: Object.fromEntries(this.ownerAccess), + }, null, 2)}\n` + writeFileSync(temporary, body, { encoding: 'utf8', mode: 0o600 }) + replaceFileSync(temporary, this.filePath) + chmodSync(this.filePath, 0o600) + this.loadedMtimeMs = this.fileMtimeMs() + this.loadedContentHash = createHash('sha1').update(body).digest('hex') + return true + } catch (error) { + this.disablePersistence(`无法保存提醒文件：${error.message}`) + return false + } + } + + pruneOwners({ persist = true } = {}) { + const now = this.now() + let changed = false + this.ownerAccess.forEach((lastAccessedAt, safeOwnerId) => { + if (!(this.ownerTtlMs > 0) || now - lastAccessedAt < this.ownerTtlMs) return + this.ownerAccess.delete(safeOwnerId) + changed = this.reminders.delete(safeOwnerId) || changed + }) + while (this.reminders.size > this.maxOwners) { + const oldest = [...this.reminders.keys()] + .sort((left, right) => ( + Number(this.ownerAccess.get(left) || 0) + - Number(this.ownerAccess.get(right) || 0) + ))[0] + if (!oldest) break + this.reminders.delete(oldest) + this.ownerAccess.delete(oldest) + changed = true + } + if (changed && persist) this.persist() + return changed + } + + touch(safeOwnerId) { + if (this.reminders.has(safeOwnerId)) this.ownerAccess.set(safeOwnerId, this.now()) + } + + health() { + let reminderCount = 0 + this.reminders.forEach(entries => { reminderCount += entries.size }) + return { + ok: !this.warning, + persistenceEnabled: Boolean(this.filePath) && !this.persistenceDisabled, + warning: this.warning, + owners: this.reminders.size, + reminders: reminderCount, + } + } + + create(safeOwnerId, { + text, + executeAt, + execute_at: legacyExecuteAt, + timezone: zone = 'UTC', + recurrence: repeat = ReminderRecurrence.ONCE, + kind = ReminderKind.REMINDER, + } = {}) { + return this.writeTransaction(() => { + const owner = ownerId(safeOwnerId) + const reminderTextValue = reminderText(text) + const nextFireAt = timestamp(executeAt ?? legacyExecuteAt) + const reminderKindValue = reminderKind(kind) + const recurrenceValue = recurrence(repeat) + const timezoneValue = timezone(zone) + this.pruneOwners({ persist: false }) + const entries = this.reminders.get(owner) || new Map() + if (entries.size >= this.maxRemindersPerOwner) { + throw new ReminderStoreError( + 'owner_limit', + `owner has reached the ${this.maxRemindersPerOwner} reminder limit`, + ) + } + const now = this.now() + let id + for (let attempts = 0; attempts < 10; attempts += 1) { + id = reminderId(this.idFactory()) + if (!entries.has(id)) break + } + if (!id || entries.has(id)) { + throw new ReminderStoreError('id_exhausted', 'unable to allocate a reminder id') + } + const reminder = { + id, + ownerId: owner, + text: reminderTextValue, + kind: reminderKindValue, + timezone: timezoneValue, + recurrence: recurrenceValue, + nextFireAt, + status: ReminderStatus.ACTIVE, + createdAt: now, + updatedAt: now, + lastFiredAt: null, + fireCount: 0, + lastError: null, + } + const previousEntries = this.reminders.get(owner) + const previousAccess = this.ownerAccess.get(owner) + entries.set(id, reminder) + this.reminders.set(owner, entries) + this.ownerAccess.set(owner, now) + this.pruneOwners({ persist: false }) + if (!this.persist()) { + if (previousEntries) this.reminders.set(owner, previousEntries) + else this.reminders.delete(owner) + if (previousAccess === undefined) this.ownerAccess.delete(owner) + else this.ownerAccess.set(owner, previousAccess) + throw new ReminderStoreError( + 'persistence_unavailable', + 'reminder persistence is unavailable', + ) + } + return cloneReminder(reminder) + }) + } + + get(safeOwnerId, id) { + this.refreshIfChanged() + const owner = ownerId(safeOwnerId) + this.pruneOwners({ persist: false }) + this.touch(owner) + return cloneReminder(this.reminders.get(owner)?.get(String(id))) + } + + list(safeOwnerId, { statuses = null } = {}) { + this.refreshIfChanged() + const owner = ownerId(safeOwnerId) + this.pruneOwners({ persist: false }) + this.touch(owner) + const allowed = statuses == null + ? null + : new Set(statuses.map(value => String(value))) + return [...(this.reminders.get(owner)?.values() || [])] + .filter(reminder => !allowed || allowed.has(reminder.status)) + .sort((left, right) => ( + left.nextFireAt - right.nextFireAt + || left.createdAt - right.createdAt + || left.id.localeCompare(right.id) + )) + .map(cloneReminder) + } + + due({ at = this.now(), ownerId: requestedOwner = null, limit = Infinity } = {}) { + const timestampNow = timestamp(at, 'invalid_now') + const owner = requestedOwner == null ? null : ownerId(requestedOwner) + const max = Number.isFinite(Number(limit)) + ? Math.max(0, Math.floor(Number(limit))) + : Infinity + this.refreshIfChanged() + this.pruneOwners({ persist: false }) + const result = [] + const owners = owner + ? [[owner, this.reminders.get(owner)]] + : [...this.reminders.entries()] + for (const [, entries] of owners) { + for (const reminder of entries?.values() || []) { + if ( + reminder.status === ReminderStatus.ACTIVE + && reminder.nextFireAt <= timestampNow + ) { + result.push(cloneReminder(reminder)) + } + } + } + return result + .sort((left, right) => left.nextFireAt - right.nextFireAt || left.id.localeCompare(right.id)) + .slice(0, max) + } + + claimDue({ at = this.now(), ownerId: requestedOwner = null, limit = Infinity } = {}) { + const timestampNow = timestamp(at, 'invalid_now') + const owner = requestedOwner == null ? null : ownerId(requestedOwner) + const max = Number.isFinite(Number(limit)) + ? Math.max(0, Math.floor(Number(limit))) + : Infinity + return this.writeTransaction(() => { + this.pruneOwners({ persist: false }) + const owners = owner + ? [[owner, this.reminders.get(owner)]] + : [...this.reminders.entries()] + const due = [] + for (const [, entries] of owners) { + for (const reminder of entries?.values() || []) { + if ( + reminder.status === ReminderStatus.ACTIVE + && reminder.nextFireAt <= timestampNow + ) due.push(reminder) + } + } + due.sort((left, right) => left.nextFireAt - right.nextFireAt || left.id.localeCompare(right.id)) + const claimed = due.slice(0, max) + const previous = claimed.map(reminder => ({ ...reminder })) + claimed.forEach(reminder => { + reminder.status = ReminderStatus.FIRING + reminder.lastFiredAt = timestampNow + reminder.fireCount += 1 + reminder.updatedAt = timestampNow + }) + if (claimed.length && !this.persist()) { + claimed.forEach((reminder, index) => Object.assign(reminder, previous[index])) + throw new ReminderStoreError( + 'persistence_unavailable', + 'reminder persistence is unavailable', + ) + } + return claimed.map(cloneReminder) + }) + } + + cancel(safeOwnerId, id) { + return this.writeTransaction(() => { + const owner = ownerId(safeOwnerId) + const entries = this.reminders.get(owner) + const reminder = entries?.get(String(id)) + if (!reminder) return null + if (TERMINAL_STATUSES.has(reminder.status)) return cloneReminder(reminder) + const previous = { ...reminder } + reminder.status = ReminderStatus.CANCELLED + reminder.updatedAt = this.now() + reminder.lastError = null + if (!this.persist()) { + Object.assign(reminder, previous) + throw new ReminderStoreError( + 'persistence_unavailable', + 'reminder persistence is unavailable', + ) + } + return cloneReminder(reminder) + }) + } + + complete(safeOwnerId, id, { nextFireAt = null, at = this.now() } = {}) { + return this.writeTransaction(() => { + const owner = ownerId(safeOwnerId) + const reminder = this.reminders.get(owner)?.get(String(id)) + if (!reminder) return null + if (reminder.status !== ReminderStatus.FIRING) { + throw new ReminderStoreError( + 'invalid_transition', + `cannot complete reminder in ${reminder.status} state`, + ) + } + const timestampNow = timestamp(at, 'invalid_now') + const next = nextFireAt == null ? null : timestamp(nextFireAt) + if (reminder.recurrence !== ReminderRecurrence.ONCE && next == null) { + throw new ReminderStoreError( + 'next_fire_required', + 'recurring reminders require the next fire time', + ) + } + if (reminder.recurrence === ReminderRecurrence.ONCE && next != null) { + throw new ReminderStoreError( + 'unexpected_next_fire', + 'one-time reminders cannot be rescheduled', + ) + } + const previous = { ...reminder } + reminder.status = next == null ? ReminderStatus.COMPLETED : ReminderStatus.ACTIVE + reminder.nextFireAt = next == null ? reminder.nextFireAt : next + reminder.updatedAt = timestampNow + reminder.lastError = null + if (!this.persist()) { + Object.assign(reminder, previous) + throw new ReminderStoreError( + 'persistence_unavailable', + 'reminder persistence is unavailable', + ) + } + return cloneReminder(reminder) + }) + } + + fail(safeOwnerId, id, error, { at = this.now() } = {}) { + return this.writeTransaction(() => { + const owner = ownerId(safeOwnerId) + const reminder = this.reminders.get(owner)?.get(String(id)) + if (!reminder) return null + if (reminder.status !== ReminderStatus.FIRING) { + throw new ReminderStoreError( + 'invalid_transition', + `cannot fail reminder in ${reminder.status} state`, + ) + } + const previous = { ...reminder } + reminder.status = ReminderStatus.FAILED + reminder.updatedAt = timestamp(at, 'invalid_now') + reminder.lastError = clean(error, 500) || 'reminder failed' + if (!this.persist()) { + Object.assign(reminder, previous) + throw new ReminderStoreError( + 'persistence_unavailable', + 'reminder persistence is unavailable', + ) + } + return cloneReminder(reminder) + }) + } +} diff --git a/server/test/frontend-reminders.test.mjs b/server/test/frontend-reminders.test.mjs new file mode 100644 index 00000000..de36709b --- /dev/null +++ b/server/test/frontend-reminders.test.mjs @@ -0,0 +1,261 @@ +import assert from 'node:assert/strict' +import { spawn } from 'node:child_process' +import { + existsSync, + mkdtempSync, + readFileSync, + rmSync, + statSync, + writeFileSync, +} from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import test from 'node:test' +import { + FrontendReminderStore, + MAX_REMINDER_TEXT_CHARS, + ReminderKind, + ReminderStatus, + ReminderStoreError, +} from '../src/conversation/frontend-reminders.mjs' + +function tempFile(prefix = 'qwen-audio-agent-reminders-') { + const directory = mkdtempSync(join(tmpdir(), prefix)) + return { + directory, + filePath: join(directory, 'frontend-reminders.json'), + } +} + +test('creates reminders and isolates owners', () => { + let now = 1_000 + const store = new FrontendReminderStore({ now: () => now }) + const reminder = store.create('owner-a', { + text: ' 明天 开会 ', + executeAt: 2_000, + timezone: 'Asia/Shanghai', + recurrence: 'daily', + }) + + assert.equal(reminder.ownerId, 'owner-a') + assert.equal(reminder.text, '明天 开会') + assert.equal(reminder.kind, ReminderKind.REMINDER) + assert.equal(reminder.status, ReminderStatus.ACTIVE) + assert.equal(reminder.nextFireAt, 2_000) + assert.equal(store.get('owner-a', reminder.id).id, reminder.id) + assert.equal(store.get('owner-b', reminder.id), null) + assert.equal(store.list('owner-b').length, 0) + + now = 1_100 + assert.equal(store.list('owner-a')[0].updatedAt, 1_000) +}) + +test('validates reminder fields and enforces the owner limit', () => { + const store = new FrontendReminderStore({ maxRemindersPerOwner: 1 }) + assert.throws( + () => store.create('', { text: 'x', executeAt: 1 }), + error => error instanceof ReminderStoreError && error.code === 'invalid_owner', + ) + assert.throws( + () => store.create('owner', { text: ' ', executeAt: 1 }), + error => error.code === 'invalid_text', + ) + assert.throws( + () => store.create('owner', { + text: 'x'.repeat(MAX_REMINDER_TEXT_CHARS + 1), + executeAt: 1, + }), + error => error.code === 'invalid_text', + ) + assert.throws( + () => store.create('owner', { text: 'x', executeAt: -1 }), + error => error.code === 'invalid_execute_at', + ) + assert.throws( + () => store.create('owner', { text: 'x', executeAt: 1, kind: 'other' }), + error => error.code === 'invalid_kind', + ) + assert.throws( + () => store.create('owner', { text: 'x', executeAt: 1, recurrence: 'hourly' }), + error => error.code === 'invalid_recurrence', + ) + assert.throws( + () => store.create('owner', { text: 'x', executeAt: 1, timezone: 'Mars/Olympus' }), + error => error.code === 'invalid_timezone', + ) + + store.create('owner', { text: 'first', executeAt: 1 }) + assert.throws( + () => store.create('owner', { text: 'second', executeAt: 2 }), + error => error.code === 'owner_limit', + ) +}) + +test('persists reminders atomically and restores them with private permissions', t => { + const { directory, filePath } = tempFile() + t.after(() => rmSync(directory, { recursive: true, force: true })) + const first = new FrontendReminderStore({ filePath, now: () => 100 }) + const created = first.create('owner-a', { + text: '提交报告', + executeAt: 200, + kind: ReminderKind.TASK, + recurrence: 'weekly', + timezone: 'Asia/Shanghai', + }) + const restored = new FrontendReminderStore({ filePath, now: () => 300 }) + + assert.deepEqual(restored.get('owner-a', created.id), created) + assert.equal(JSON.parse(readFileSync(filePath, 'utf8')).version, 1) + if (process.platform !== 'win32') { + assert.equal(statSync(filePath).mode & 0o777, 0o600) + } +}) + +test('quarantines corrupt data and remains usable after the warning', t => { + const { directory, filePath } = tempFile('qwen-audio-agent-reminders-corrupt-') + t.after(() => rmSync(directory, { recursive: true, force: true })) + writeFileSync(filePath, '{not-json') + const warnings = [] + const store = new FrontendReminderStore({ + filePath, + now: () => 12345, + onWarning: warning => warnings.push(warning), + }) + + assert.deepEqual(store.list('owner-a'), []) + assert.equal(warnings.length, 1) + assert.equal(existsSync(`${filePath}.corrupt-12345`), true) + const reminder = store.create('owner-a', { text: '恢复后可写', executeAt: 2 }) + assert.equal(store.get('owner-a', reminder.id).text, '恢复后可写') + assert.equal(store.health().ok, false) + assert.equal(store.health().persistenceEnabled, true) +}) + +test('rolls back an in-memory create when persistence is unavailable', t => { + const { directory } = tempFile('qwen-audio-agent-reminders-write-failure-') + t.after(() => rmSync(directory, { recursive: true, force: true })) + const warnings = [] + const store = new FrontendReminderStore({ + filePath: directory, + onWarning: warning => warnings.push(warning), + }) + + assert.throws( + () => store.create('owner-a', { text: '不能写入', executeAt: 2 }), + error => error.code === 'persistence_unavailable', + ) + assert.deepEqual(store.list('owner-a'), []) + assert.equal(store.health().persistenceEnabled, false) + assert.equal(warnings.length > 0, true) +}) + +test('claims due reminders once and completes one-time or recurring lifecycles', () => { + const ids = ['rem_one', 'rem_recurring', 'rem_future'] + const store = new FrontendReminderStore({ + now: () => 1_000, + idFactory: () => ids.shift(), + }) + const oneTime = store.create('owner-a', { text: '一次', executeAt: 900 }) + const recurring = store.create('owner-a', { + text: '每天', + executeAt: 900, + recurrence: 'daily', + }) + const future = store.create('owner-a', { text: '未来', executeAt: 2_000 }) + + assert.deepEqual(store.due({ at: 1_000 }).map(item => item.id), [oneTime.id, recurring.id]) + const claimed = store.claimDue({ at: 1_000 }) + assert.deepEqual(claimed.map(item => item.id), [oneTime.id, recurring.id]) + assert.equal(store.claimDue({ at: 1_000 }).length, 0) + assert.equal(store.get('owner-a', oneTime.id).fireCount, 1) + assert.equal(store.get('owner-a', future.id).status, ReminderStatus.ACTIVE) + + assert.equal(store.complete('owner-a', oneTime.id, { at: 1_100 }).status, ReminderStatus.COMPLETED) + const next = store.complete('owner-a', recurring.id, { + at: 1_100, + nextFireAt: 2_000, + }) + assert.equal(next.status, ReminderStatus.ACTIVE) + assert.equal(next.nextFireAt, 2_000) +}) + +test('requires explicit next fire times for recurring reminders', () => { + const store = new FrontendReminderStore({ now: () => 1_000 }) + const reminder = store.create('owner-a', { + text: '重复', + executeAt: 900, + recurrence: 'daily', + }) + store.claimDue({ at: 1_000 }) + + assert.throws( + () => store.complete('owner-a', reminder.id, { at: 1_100 }), + error => error.code === 'next_fire_required', + ) + assert.equal(store.get('owner-a', reminder.id).status, ReminderStatus.FIRING) +}) + +test('cancels only the owner-owned reminder and is idempotent', () => { + const store = new FrontendReminderStore({ now: () => 1_000 }) + const reminder = store.create('owner-a', { text: '取消我', executeAt: 900 }) + + assert.equal(store.cancel('owner-b', reminder.id), null) + assert.equal(store.cancel('owner-a', reminder.id).status, ReminderStatus.CANCELLED) + assert.equal(store.cancel('owner-a', reminder.id).status, ReminderStatus.CANCELLED) + assert.equal(store.claimDue({ at: 2_000 }).length, 0) +}) + +test('reopens an interrupted firing reminder after restart', t => { + const { directory, filePath } = tempFile('qwen-audio-agent-reminders-recovery-') + t.after(() => rmSync(directory, { recursive: true, force: true })) + const first = new FrontendReminderStore({ filePath, now: () => 1_000 }) + const reminder = first.create('owner-a', { text: '恢复', executeAt: 900 }) + first.claimDue({ at: 1_000 }) + + const restored = new FrontendReminderStore({ filePath, now: () => 2_000 }) + assert.equal(restored.get('owner-a', reminder.id).status, ReminderStatus.ACTIVE) + assert.equal(restored.due({ at: 2_000 }).length, 1) +}) + +test('merges writes from two Gateway instances sharing one file', t => { + const { directory, filePath } = tempFile('qwen-audio-agent-reminders-shared-') + t.after(() => rmSync(directory, { recursive: true, force: true })) + const first = new FrontendReminderStore({ filePath, now: () => 1_000 }) + const second = new FrontendReminderStore({ filePath, now: () => 1_100 }) + first.create('owner-a', { text: '来自 CLI', executeAt: 2_000 }) + second.create('owner-a', { text: '来自桌面', executeAt: 3_000 }) + + assert.deepEqual(first.list('owner-a').map(item => item.text), ['来自 CLI', '来自桌面']) +}) + +test('serializes concurrent reminder creation from independent processes', async t => { + const { directory, filePath } = tempFile('qwen-audio-agent-reminders-processes-') + t.after(() => rmSync(directory, { recursive: true, force: true })) + const moduleUrl = new URL('../src/conversation/frontend-reminders.mjs', import.meta.url).href + const texts = Array.from({ length: 8 }, (_, index) => `process-${index}`) + const script = ` + import { FrontendReminderStore } from ${JSON.stringify(moduleUrl)} + const store = new FrontendReminderStore({ filePath: process.argv[1] }) + store.create('owner-a', { text: process.argv[2], executeAt: 1000 }) + ` + await Promise.all(texts.map(text => new Promise((resolvePromise, rejectPromise) => { + const child = spawn(process.execPath, [ + '--input-type=module', + '-e', + script, + filePath, + text, + ], { stdio: ['ignore', 'pipe', 'pipe'] }) + let stderr = '' + child.stderr.setEncoding('utf8') + child.stderr.on('data', chunk => { stderr += chunk }) + child.once('error', rejectPromise) + child.once('exit', code => { + if (code === 0) resolvePromise() + else rejectPromise(new Error(`child exited ${code}: ${stderr}`)) + }) + }))) + + const store = new FrontendReminderStore({ filePath }) + assert.deepEqual(store.list('owner-a').map(item => item.text).sort(), texts.sort()) +})