From 38d538407343d3bd8ffaf1ef67a63334721871df Mon Sep 17 00:00:00 2001 From: hqhq1025 <1506751656@qq.com> Date: Sun, 5 Jul 2026 16:07:04 +0800 Subject: [PATCH 1/2] feat(runtime): add session-internal CronJob scheduling (Issue #15 Primitive 4) Implements CronCreate/CronDelete/CronList tools aligned with Claude Code's pattern. The agent can schedule prompts to fire at future times within the same session, enabling polling, monitoring, and periodic check workflows. New files: - wakeup-scheduler.ts: core scheduler with timer management, cron parser, auto-expire (7d for recurring), jitter, idle-gate, and race-safe fire() - wakeup-tools.ts: CronCreate/CronDelete/CronList tool wrappers - wakeup-scheduler.test.ts: 28 unit tests covering scheduling, cron parsing, recurring reschedule, auto-expire, jitter bounds, idle-gate backoff, and cancel-during-canFire race condition Integration: - runtime/index.ts: export WakeupScheduler and cron tool builders - desktop/main.ts: instantiate scheduler, register tools, wire session stop/archive/delete cleanup, dispose on before-quit - ui/tool-activity.tsx: CronJob result preview cards with Lucide icons (Clock for one-shot, Repeat for recurring, Check for cancelled) Closes #15 --- apps/desktop/src/main/main.ts | 31 ++ .../src/__tests__/wakeup-scheduler.test.ts | 379 +++++++++++++++ packages/runtime/src/index.ts | 15 + packages/runtime/src/wakeup-scheduler.ts | 434 ++++++++++++++++++ packages/runtime/src/wakeup-tools.ts | 118 +++++ packages/ui/src/tool-activity.tsx | 82 +++- 6 files changed, 1058 insertions(+), 1 deletion(-) create mode 100644 packages/runtime/src/__tests__/wakeup-scheduler.test.ts create mode 100644 packages/runtime/src/wakeup-scheduler.ts create mode 100644 packages/runtime/src/wakeup-tools.ts diff --git a/apps/desktop/src/main/main.ts b/apps/desktop/src/main/main.ts index 3c9394f31c..0b69a8133b 100644 --- a/apps/desktop/src/main/main.ts +++ b/apps/desktop/src/main/main.ts @@ -80,6 +80,8 @@ import { getWechatBridgeQrCode, testBotChannel as testRuntimeBotChannel, setActiveProxy, + WakeupScheduler, + buildCronTools, } from '@maka/runtime'; import type { ToolAvailabilityConfig, @@ -376,6 +378,28 @@ const officeTools = [buildOfficeDocumentTool(), buildOfficeDocumentEditTool()]; const browserTools = buildBrowserTools(); const agentTools = [buildSubagentSpawnTool(), ...buildSubagentProjectionTools()]; const deferredTools = [...riveTools, ...officeTools, ...browserTools, ...agentTools]; +// WakeupScheduler — session-internal CronJob scheduling (Issue #15, Primitive 4). +// The scheduler is created before `runtime` (which is defined below) because its +// tools must be listed in `builtinTools`. The `injectTurn` / `canFire` callbacks +// capture `runtime` via closure and only execute asynchronously (when timers fire), +// so `runtime` is guaranteed to be initialized by then. +const wakeupScheduler = new WakeupScheduler({ + newId: randomUUID, + now: Date.now, + injectTurn: (sessionId, input) => { + const iterator = runtime.sendMessage(sessionId, input); + void streamEvents(sessionId, iterator, input.turnId); + }, + canFire: async (sessionId) => { + try { + const header = await store.readHeader(sessionId); + return header.status === 'active' && !header.archivedAt; + } catch { + return false; + } + }, +}); +const cronTools = buildCronTools(wakeupScheduler); const toolAvailability: ToolAvailabilityConfig = { economy: economyEnabled, groups: [ @@ -406,6 +430,9 @@ const builtinTools = [ // Session task ledger: model manages a flat task list; the current list is // re-injected each turn tail. Pure local state, so no permission gate. ...taskLedgerWiring.tools, + // CronJob tools: CronCreate/CronDelete/CronList for session-internal scheduling. + // The agent can schedule prompts to fire at future times within the same session. + ...cronTools, // The `load_tools` connector is built by ToolAvailabilityRuntime; deferred // group tools just need to be present so they are dispatchable once loaded. ...deferredTools, @@ -1130,6 +1157,7 @@ function registerIpc(): void { }); ipcMain.handle('sessions:stop', async (_event, sessionId: string, input?: { source?: 'stop_button' }) => { await runtime.stopSession(sessionId, normalizeStopSessionInput(input)); + wakeupScheduler.cancelAllForSession(sessionId); emitSessionsChanged('status-change', sessionId); emitSessionsChanged('turn-status-change', sessionId); emitSessionsChanged('message-appended', sessionId); @@ -1175,6 +1203,7 @@ function registerIpc(): void { }); ipcMain.handle('sessions:archive', async (_event, sessionId: string) => { await runtime.archive(sessionId); + wakeupScheduler.cancelAllForSession(sessionId); // An archived conversation is no longer shown: drop its browser connection // and view so it does not keep a live Chromium page in the background. await releaseBrowserSession(sessionId); @@ -1228,6 +1257,7 @@ function registerIpc(): void { }); ipcMain.handle('sessions:remove', async (_event, sessionId: string) => { await runtime.remove(sessionId); + wakeupScheduler.cancelAllForSession(sessionId); // Drop the conversation's browser connection and destroy its view (no-op // if it never opened one). releaseBrowserSession disposes the view via the // host, covering both agent-driven and hand-opened views. @@ -1787,6 +1817,7 @@ app.on('window-all-closed', () => { }); app.on('before-quit', () => { + wakeupScheduler.dispose(); planReminders.stopTimers(); dailyReview.stopScheduler(); void botRegistry.stopAll(); diff --git a/packages/runtime/src/__tests__/wakeup-scheduler.test.ts b/packages/runtime/src/__tests__/wakeup-scheduler.test.ts new file mode 100644 index 0000000000..9b0352d5a2 --- /dev/null +++ b/packages/runtime/src/__tests__/wakeup-scheduler.test.ts @@ -0,0 +1,379 @@ +import assert from 'node:assert/strict'; +import { describe, test } from 'node:test'; +import { WakeupScheduler, computeNextCronRun, computeJitter } from '../wakeup-scheduler.js'; + +function createTestScheduler(overrides: Partial[0]> = {}) { + const fired: Array<{ sessionId: string; turnId: string; text: string }> = []; + const timers: Array<{ id: number; cb: () => void; ms: number; cleared: boolean }> = []; + let nextId = 1; + let mockNow = 1000000; + + const scheduler = new WakeupScheduler({ + newId: () => `id-${nextId++}`, + now: () => mockNow, + injectTurn: (sessionId, input) => { fired.push({ sessionId, ...input }); }, + canFire: async () => true, + setTimer: (cb, ms) => { + const timer = { id: timers.length + 1, cb, ms, cleared: false }; + timers.push(timer); + return timer.id as unknown as ReturnType; + }, + clearTimer: (id) => { + const timer = timers.find(t => t.id === (id as unknown as number)); + if (timer) timer.cleared = true; + }, + ...overrides, + }); + + return { + scheduler, + fired, + timers, + advanceTime: (ms: number) => { mockNow += ms; }, + fireNextTimer: async () => { + const pending = timers.find(t => !t.cleared); + if (pending) { + pending.cleared = true; + mockNow += pending.ms; + pending.cb(); + await new Promise(resolve => setTimeout(resolve, 0)); + } + }, + }; +} + +describe('WakeupScheduler', () => { + test('schedule creates a pending record and sets a timer', () => { + const { scheduler, timers } = createTestScheduler(); + const record = scheduler.schedule('session-1', { + delaySeconds: 60, + message: 'check status', + reason: 'polling deployment', + }); + assert.equal(record.status, 'pending'); + assert.equal(record.sessionId, 'session-1'); + assert.equal(record.message, 'check status'); + assert.equal(timers.length, 1); + assert.equal(timers[0].ms, 60000); + }); + + test('timer fire injects a turn with the message', async () => { + const { scheduler, fired, fireNextTimer } = createTestScheduler(); + scheduler.schedule('session-1', { delaySeconds: 10, message: 'hello wakeup', reason: 'test' }); + await fireNextTimer(); + assert.equal(fired.length, 1); + assert.equal(fired[0].sessionId, 'session-1'); + assert.ok(fired[0].text.includes('hello wakeup')); + }); + + test('fired record status is updated', async () => { + const { scheduler, fireNextTimer } = createTestScheduler(); + const record = scheduler.schedule('session-1', { delaySeconds: 5, message: 'check', reason: 'test' }); + await fireNextTimer(); + const records = scheduler.listForSession('session-1'); + assert.equal(records.find(r => r.id === record.id)?.status, 'fired'); + }); + + test('cancel prevents firing', () => { + const { scheduler, fired, timers } = createTestScheduler(); + const record = scheduler.schedule('session-1', { delaySeconds: 30, message: 'should not fire', reason: 'test' }); + const cancelled = scheduler.cancel(record.id); + assert.equal(cancelled, true); + assert.ok(timers[0].cleared); + assert.equal(fired.length, 0); + assert.equal(scheduler.listForSession('session-1')[0].status, 'cancelled'); + }); + + test('cancelAllForSession cancels all pending wakeups', () => { + const { scheduler } = createTestScheduler(); + scheduler.schedule('session-1', { delaySeconds: 10, message: 'a', reason: 'test' }); + scheduler.schedule('session-1', { delaySeconds: 20, message: 'b', reason: 'test' }); + scheduler.schedule('session-2', { delaySeconds: 30, message: 'c', reason: 'test' }); + scheduler.cancelAllForSession('session-1'); + const s1 = scheduler.listForSession('session-1'); + const s2 = scheduler.listForSession('session-2'); + assert.ok(s1.every(r => r.status === 'cancelled')); + assert.equal(s2[0].status, 'pending'); + }); + + test('rejects more than 5 pending wakeups per session', () => { + const { scheduler } = createTestScheduler(); + for (let i = 0; i < 5; i++) { + scheduler.schedule('session-1', { delaySeconds: 10, message: `m${i}`, reason: 'test' }); + } + assert.throws( + () => scheduler.schedule('session-1', { delaySeconds: 10, message: 'overflow', reason: 'test' }), + /Max 5 pending wakeups/, + ); + }); + + test('rejects delay out of range', () => { + const { scheduler } = createTestScheduler(); + assert.throws(() => scheduler.schedule('s', { delaySeconds: 0, message: 'm', reason: 'r' }), /delay_seconds/); + assert.throws(() => scheduler.schedule('s', { delaySeconds: 100000, message: 'm', reason: 'r' }), /delay_seconds/); + }); + + test('delaySeconds=1 (minimum boundary) succeeds', () => { + const { scheduler, timers } = createTestScheduler(); + const record = scheduler.schedule('session-1', { delaySeconds: 1, message: 'boundary', reason: 'test' }); + assert.equal(record.status, 'pending'); + assert.equal(timers.length, 1); + assert.equal(timers[0].ms, 1000); + }); + + test('rejects when both cronExpression and delaySeconds are provided', () => { + const { scheduler } = createTestScheduler(); + assert.throws( + () => scheduler.schedule('s', { cronExpression: '* * * * *', delaySeconds: 60, message: 'm', reason: 'r' }), + /Provide cronExpression or delaySeconds, not both/, + ); + }); + + test('backs off and retries when canFire returns false', async () => { + let canFireCount = 0; + const { scheduler, fired, fireNextTimer } = createTestScheduler({ + canFire: async () => { canFireCount++; return canFireCount >= 3; }, + }); + scheduler.schedule('session-1', { delaySeconds: 5, message: 'retry-test', reason: 'test' }); + await fireNextTimer(); + assert.equal(fired.length, 0); + await fireNextTimer(); + assert.equal(fired.length, 0); + await fireNextTimer(); + assert.equal(fired.length, 1); + assert.ok(fired[0].text.includes('retry-test')); + }); + + test('expires after max retries', async () => { + const { scheduler, fired, fireNextTimer } = createTestScheduler({ canFire: async () => false }); + const record = scheduler.schedule('session-1', { delaySeconds: 5, message: 'expire-test', reason: 'test' }); + await fireNextTimer(); + await fireNextTimer(); + await fireNextTimer(); + await fireNextTimer(); + assert.equal(fired.length, 0); + assert.equal(scheduler.listForSession('session-1').find(r => r.id === record.id)?.status, 'expired'); + }); + + test('dispose clears all timers', () => { + const { scheduler, timers } = createTestScheduler(); + scheduler.schedule('session-1', { delaySeconds: 10, message: 'a', reason: 'test' }); + scheduler.schedule('session-1', { delaySeconds: 20, message: 'b', reason: 'test' }); + scheduler.dispose(); + assert.ok(timers.every(t => t.cleared)); + }); + + test('multiple sessions work independently', async () => { + const { scheduler, fired, fireNextTimer } = createTestScheduler(); + scheduler.schedule('session-1', { delaySeconds: 5, message: 'msg-1', reason: 'test' }); + scheduler.schedule('session-2', { delaySeconds: 10, message: 'msg-2', reason: 'test' }); + await fireNextTimer(); + assert.equal(fired.length, 1); + assert.equal(fired[0].sessionId, 'session-1'); + await fireNextTimer(); + assert.equal(fired.length, 2); + assert.equal(fired[1].sessionId, 'session-2'); + }); + + // ─── Cron expression tests ────────────────────────────────────────────────── + + test('computeNextCronRun returns a future timestamp for valid expression', () => { + // "every minute" should return 1 minute after the reference time + const refMs = Date.now(); + const next = computeNextCronRun('* * * * *', refMs); + assert.notEqual(next, null); + assert.ok(next! > refMs, 'next run should be in the future'); + // Should be within 60 seconds of the reference + assert.ok(next! - refMs <= 60_000, 'next run for * * * * * should be within 60s'); + }); + + test('computeNextCronRun returns null for invalid expression', () => { + assert.equal(computeNextCronRun('invalid', 1000000), null); + assert.equal(computeNextCronRun('60 * * * *', 1000000), null); // minute out of range + assert.equal(computeNextCronRun('* * * *', 1000000), null); // only 4 fields + }); + + test('schedule with cronExpression creates a pending record', () => { + const { scheduler, timers } = createTestScheduler(); + const record = scheduler.schedule('session-1', { + cronExpression: '* * * * *', + message: 'cron check', + reason: 'every minute', + }); + assert.equal(record.status, 'pending'); + assert.equal(record.cronExpression, '* * * * *'); + assert.ok(record.firesAt > 1000000, 'firesAt should be in the future'); + assert.equal(timers.length, 1); + assert.ok(timers[0].ms > 0, 'timer delay should be positive'); + }); + + test('schedule rejects when neither delaySeconds nor cronExpression provided', () => { + const { scheduler } = createTestScheduler(); + assert.throws( + () => scheduler.schedule('s', { message: 'm', reason: 'r' }), + /Either cronExpression or delaySeconds/, + ); + }); + + test('schedule rejects invalid cron expression', () => { + const { scheduler } = createTestScheduler(); + assert.throws( + () => scheduler.schedule('s', { cronExpression: 'bad bad bad', message: 'm', reason: 'r' }), + /Invalid cron expression/, + ); + }); + + test('cron-based recurring job reschedules after firing', async () => { + const { scheduler, fired, fireNextTimer } = createTestScheduler(); + const record = scheduler.schedule('session-1', { + cronExpression: '* * * * *', + message: 'cron recurring', + reason: 'every minute', + recurring: true, + }); + await fireNextTimer(); + assert.equal(fired.length, 1); + assert.ok(fired[0].text.includes('cron recurring')); + // Should have created a new pending record for the next occurrence + const pending = scheduler.listForSession('session-1').filter(r => r.status === 'pending'); + assert.equal(pending.length, 1, 'should have one new pending record'); + assert.equal(pending[0].cronExpression, '* * * * *'); + }); + + // ─── Auto-expire tests ───────────────────────────────────────────────────── + + test('recurring job expires after 7 days', async () => { + const sevenDaysMs = 7 * 24 * 60 * 60 * 1000; + const { scheduler, fired, timers, advanceTime, fireNextTimer } = createTestScheduler(); + const record = scheduler.schedule('session-1', { + delaySeconds: 60, + message: 'long running', + reason: 'test', + recurring: true, + }); + // Verify expiresAt is set + assert.notEqual(record.expiresAt, null); + assert.equal(record.expiresAt, record.scheduledAt + sevenDaysMs); + + // Advance time past the 7-day expiry before firing + advanceTime(sevenDaysMs + 1000); + await fireNextTimer(); + + // The job should have expired, not fired + assert.equal(fired.length, 0); + const records = scheduler.listForSession('session-1'); + assert.equal(records.find(r => r.id === record.id)?.status, 'expired'); + }); + + test('one-shot jobs have null expiresAt', () => { + const { scheduler } = createTestScheduler(); + const record = scheduler.schedule('session-1', { + delaySeconds: 60, + message: 'one-shot', + reason: 'test', + recurring: false, + }); + assert.equal(record.expiresAt, null); + }); + + // ─── Jitter tests ────────────────────────────────────────────────────────── + + test('computeJitter for recurring returns a value within bounds', () => { + for (let i = 0; i < 50; i++) { + const delayMs = 600_000; // 10 minutes + const jitter = computeJitter(delayMs, true); + // 10% of 600000 = 60000, which is < MAX_JITTER_MS (15 min = 900000) + assert.ok(jitter >= 0, 'recurring jitter should be non-negative'); + assert.ok(jitter <= 60_000, 'recurring jitter should be <= 10% of delay'); + } + }); + + test('computeJitter for one-shot non-30min-aligned returns 0', () => { + const jitter = computeJitter(60_000, false); // 1 minute, not 30min multiple + assert.equal(jitter, 0); + }); + + // ─── Idle-gate observability tests ───────────────────────────────────────── + + test('fireAttempts increments and deferredFires logs timestamps on idle rejection', async () => { + let callCount = 0; + const { scheduler, fireNextTimer } = createTestScheduler({ + canFire: async () => { callCount++; return callCount >= 2; }, + }); + const record = scheduler.schedule('session-1', { + delaySeconds: 5, + message: 'idle-gate test', + reason: 'test', + }); + // First fire attempt: rejected + await fireNextTimer(); + const afterFirst = scheduler.listForSession('session-1').find(r => r.id === record.id)!; + assert.equal(afterFirst.fireAttempts, 1); + assert.equal(afterFirst.deferredFires.length, 1); + // Second fire attempt: succeeds + await fireNextTimer(); + const afterSecond = scheduler.listForSession('session-1').find(r => r.id === record.id)!; + assert.equal(afterSecond.fireAttempts, 2); + assert.equal(afterSecond.status, 'fired'); + // deferredFires should still have 1 entry (only logged on rejection) + assert.equal(afterSecond.deferredFires.length, 1); + }); + + // ─── Cancel edge-case tests ──────────────────────────────────────────────── + + test('cancel returns false when the wakeup has already fired', async () => { + const { scheduler, fireNextTimer } = createTestScheduler(); + const record = scheduler.schedule('session-1', { delaySeconds: 5, message: 'fire-first', reason: 'test' }); + await fireNextTimer(); + assert.equal(record.status, 'fired'); + const result = scheduler.cancel(record.id); + assert.equal(result, false); + assert.equal(record.status, 'fired'); + }); + + test('cancel returns false on second call (already cancelled)', () => { + const { scheduler } = createTestScheduler(); + const record = scheduler.schedule('session-1', { delaySeconds: 30, message: 'cancel-twice', reason: 'test' }); + const first = scheduler.cancel(record.id); + assert.equal(first, true); + assert.equal(record.status, 'cancelled'); + const second = scheduler.cancel(record.id); + assert.equal(second, false); + assert.equal(record.status, 'cancelled'); + }); + + // ─── Race condition: cancel during canFire await ─────────────────────────── + + test('cancel during canFire await prevents firing', async () => { + let canFireResolve: (value: boolean) => void; + const { scheduler, fired, fireNextTimer } = createTestScheduler({ + canFire: () => new Promise((resolve) => { canFireResolve = resolve; }), + }); + const record = scheduler.schedule('session-1', { delaySeconds: 5, message: 'race', reason: 'test' }); + + // Fire the timer -- this starts the async canFire call + const firePromise = fireNextTimer(); + + // While canFire is in-flight, cancel the wakeup + scheduler.cancel(record.id); + assert.equal(record.status, 'cancelled'); + + // Now resolve canFire -- fire() should see the cancelled status and bail + canFireResolve!(true); + await firePromise; + + assert.equal(fired.length, 0, 'cancelled wakeup should not have fired'); + assert.equal(record.status, 'cancelled'); + }); + + // ─── One-shot jitter on 30-minute-aligned delays ────────────────────────── + + test('computeJitter for one-shot 30min-aligned returns negative value in bounds', () => { + for (let i = 0; i < 50; i++) { + const delayMs = 30 * 60 * 1000; // 30 minutes, aligned + const jitter = computeJitter(delayMs, false); + assert.ok(jitter <= 0, `one-shot jitter should be <= 0, got ${jitter}`); + assert.ok(jitter >= -90_000, `one-shot jitter should be >= -90000, got ${jitter}`); + } + }); +}); diff --git a/packages/runtime/src/index.ts b/packages/runtime/src/index.ts index 5d0198c5f6..4a1701b481 100644 --- a/packages/runtime/src/index.ts +++ b/packages/runtime/src/index.ts @@ -490,6 +490,21 @@ export type { RuntimeEventLike, } from './tool-availability.js'; +// ─────────────────────────────────────────────────────────────────────────── +// WakeupScheduler — session-internal CronJob scheduling (Issue #15, Primitive 4). +// ─────────────────────────────────────────────────────────────────────────── +export { WakeupScheduler, computeNextCronRun, computeJitter } from './wakeup-scheduler.js'; +export type { WakeupRecord, WakeupSchedulerDeps } from './wakeup-scheduler.js'; +export { + CRON_CREATE_TOOL_NAME, + CRON_DELETE_TOOL_NAME, + CRON_LIST_TOOL_NAME, + buildCronCreateTool, + buildCronDeleteTool, + buildCronListTool, + buildCronTools, +} from './wakeup-tools.js'; + // ─────────────────────────────────────────────────────────────────────────── // System-prompt fragments (shared by the desktop app and the CLI/TUI). // Read-only, stateless builders for project instructions, personalization, git diff --git a/packages/runtime/src/wakeup-scheduler.ts b/packages/runtime/src/wakeup-scheduler.ts new file mode 100644 index 0000000000..f5dc4eeac3 --- /dev/null +++ b/packages/runtime/src/wakeup-scheduler.ts @@ -0,0 +1,434 @@ +/** + * WakeupScheduler — session-internal timer that re-injects a turn after a delay. + * + * The agent calls `schedule_wakeup` to arrange a future synthetic turn in the + * same session, enabling polling/monitoring loops without user interaction. + * Analogous to Claude Code's ScheduleWakeup tool. + */ + +// ─── Minimal cron parser (no import from core to avoid circular deps) ──────── + +interface ParsedCronField { + wildcard: boolean; + values: Set; +} + +interface ParsedCronExpression { + minute: ParsedCronField; + hour: ParsedCronField; + dayOfMonth: ParsedCronField; + month: ParsedCronField; + dayOfWeek: ParsedCronField; +} + +function parseCronInteger(input: string, min: number, max: number): number | null { + if (!/^\d+$/.test(input)) return null; + const value = Number(input); + if (!Number.isSafeInteger(value) || value < min || value > max) return null; + return value; +} + +function parseCronField(input: string, min: number, max: number, normalizeSevenToZero: boolean): ParsedCronField | null { + if (!/^[\d*,/\-]+$/.test(input)) return null; + const values = new Set(); + let wildcard = false; + + for (const rawPart of input.split(',')) { + if (!rawPart) return null; + const stepSplit = rawPart.split('/'); + if (stepSplit.length > 2) return null; + + const base = stepSplit[0] ?? ''; + const stepVal = stepSplit[1] === undefined ? 1 : parseCronInteger(stepSplit[1], 1, max - min + 1); + if (stepVal === null) return null; + + let start: number; + let end: number; + + if (base === '*') { + if (stepSplit.length === 1) wildcard = true; + start = min; + end = max; + } else if (base.includes('-')) { + const range = base.split('-'); + if (range.length !== 2) return null; + const parsedStart = parseCronInteger(range[0] ?? '', min, max); + const parsedEnd = parseCronInteger(range[1] ?? '', min, max); + if (parsedStart === null || parsedEnd === null || parsedStart > parsedEnd) return null; + start = parsedStart; + end = parsedEnd; + } else { + const parsed = parseCronInteger(base, min, max); + if (parsed === null) return null; + start = parsed; + end = parsed; + } + + for (let v = start; v <= end; v += stepVal) { + values.add(normalizeSevenToZero && v === 7 ? 0 : v); + } + } + + if (values.size === 0) return null; + return { wildcard, values }; +} + +function parseCronExpression(expression: string): ParsedCronExpression | null { + const parts = expression.trim().split(/\s+/); + if (parts.length !== 5) return null; + + const minute = parseCronField(parts[0]!, 0, 59, false); + if (!minute) return null; + const hour = parseCronField(parts[1]!, 0, 23, false); + if (!hour) return null; + const dayOfMonth = parseCronField(parts[2]!, 1, 31, false); + if (!dayOfMonth) return null; + const month = parseCronField(parts[3]!, 1, 12, false); + if (!month) return null; + const dayOfWeek = parseCronField(parts[4]!, 0, 7, true); + if (!dayOfWeek) return null; + + return { minute, hour, dayOfMonth, month, dayOfWeek }; +} + +function cronExpressionMatches(expr: ParsedCronExpression, date: Date): boolean { + if (!expr.minute.values.has(date.getMinutes())) return false; + if (!expr.hour.values.has(date.getHours())) return false; + if (!expr.month.values.has(date.getMonth() + 1)) return false; + + const dayOfMonthMatches = expr.dayOfMonth.values.has(date.getDate()); + const dayOfWeekMatches = expr.dayOfWeek.values.has(date.getDay()); + + // Standard cron: if both day-of-month and day-of-week are restricted (non-wildcard), + // match if EITHER matches. If only one is restricted, both must match. + if (!expr.dayOfMonth.wildcard && !expr.dayOfWeek.wildcard) { + return dayOfMonthMatches || dayOfWeekMatches; + } + return dayOfMonthMatches && dayOfWeekMatches; +} + +/** + * Compute the next cron run time after the given epoch-ms timestamp. + * Scans minute-by-minute up to ~366 days into the future. + * Returns epoch-ms of next matching minute, or null if expression is invalid + * or no match found within the scan window. + */ +export function computeNextCronRun(expression: string, afterMs: number): number | null { + const parsed = parseCronExpression(expression); + if (!parsed) return null; + + // Start from the next full minute after `afterMs` + const start = new Date(afterMs); + start.setSeconds(0, 0); + start.setMinutes(start.getMinutes() + 1); + + // Scan up to 366 days * 24 hours * 60 minutes = 527,040 iterations max + const maxIterations = 366 * 24 * 60; + const candidate = new Date(start.getTime()); + + for (let i = 0; i < maxIterations; i++) { + if (cronExpressionMatches(parsed, candidate)) { + return candidate.getTime(); + } + candidate.setMinutes(candidate.getMinutes() + 1); + } + + return null; +} + +// ─── WakeupScheduler ───────────────────────────────────────────────────────── + +export interface WakeupRecord { + id: string; + sessionId: string; + message: string; + reason: string; + scheduledAt: number; + firesAt: number; + delaySeconds: number; + recurring: boolean; + /** Standard 5-field cron expression (minute hour dom month dow). */ + cronExpression?: string; + status: 'pending' | 'fired' | 'cancelled' | 'expired'; + /** For recurring jobs: absolute timestamp when this job chain expires. */ + expiresAt: number | null; + /** Number of times fire() was attempted (including idle-deferred attempts). */ + fireAttempts: number; + /** Timestamps when fire was deferred due to non-idle session. */ + deferredFires: number[]; +} + +export interface WakeupSchedulerDeps { + newId: () => string; + now: () => number; + injectTurn: (sessionId: string, input: { turnId: string; text: string }) => void; + canFire: (sessionId: string) => Promise; + setTimer?: (cb: () => void, ms: number) => ReturnType; + clearTimer?: (timer: ReturnType) => void; + random?: () => number; +} + +const MAX_WAKEUPS_PER_SESSION = 5; +const MAX_DELAY_SECONDS = 86_400; +const BACKOFF_MS = 5_000; +const MAX_FIRE_RETRIES = 3; + +/** Recurring jobs auto-expire after 7 days to prevent infinite loops. */ +const MAX_RECURRING_AGE_MS = 7 * 24 * 60 * 60 * 1000; + +/** Maximum jitter cap for recurring re-schedules: 15 minutes. */ +const MAX_JITTER_MS = 15 * 60 * 1000; + +/** Maximum early jitter for one-shot jobs firing on round minutes: 90 seconds. */ +const ONE_SHOT_JITTER_MS = 90 * 1000; + +/** + * Compute jitter to add to a scheduled delay. + * + * - Recurring: up to 10% of the delay, capped at 15 minutes. + * - One-shot firing on :00 or :30: up to 90s early jitter (returned as negative). + * Otherwise 0 for one-shot. + */ +export function computeJitter(delayMs: number, recurring: boolean, random: () => number = Math.random): number { + if (recurring) { + const maxJitter = Math.min(delayMs * 0.1, MAX_JITTER_MS); + return Math.floor(random() * maxJitter); + } + // One-shot: check if firesAt lands on :00 or :30 + // Caller passes the actual firesAt timestamp for this check; + // here we compute based on delay alignment to 30-minute boundaries. + // We apply early jitter (negative) if delay is a multiple of 30 minutes. + if (delayMs > 0 && delayMs % (30 * 60 * 1000) === 0) { + return -(Math.floor(random() * ONE_SHOT_JITTER_MS)); + } + return 0; +} + +export class WakeupScheduler { + private readonly records = new Map(); + private readonly timers = new Map>(); + private readonly retries = new Map(); + private readonly deps: Required; + private disposed = false; + + constructor(deps: WakeupSchedulerDeps) { + this.deps = { + setTimer: deps.setTimer ?? setTimeout, + clearTimer: deps.clearTimer ?? clearTimeout, + random: deps.random ?? Math.random, + ...deps, + } as Required; + } + + schedule( + sessionId: string, + input: { + delaySeconds?: number; + cronExpression?: string; + message: string; + reason: string; + recurring?: boolean; + }, + ): WakeupRecord { + const pendingCount = this.listForSession(sessionId).filter(r => r.status === 'pending').length; + if (pendingCount >= MAX_WAKEUPS_PER_SESSION) { + throw new Error(`Max ${MAX_WAKEUPS_PER_SESSION} pending wakeups per session.`); + } + + const now = this.deps.now(); + const recurring = input.recurring ?? false; + let firesAt: number; + let delaySeconds: number; + + if (input.cronExpression && input.delaySeconds !== undefined) { + throw new Error('Provide cronExpression or delaySeconds, not both.'); + } + + if (input.cronExpression) { + // Cron-based scheduling: compute next matching time + const nextRun = computeNextCronRun(input.cronExpression, now); + if (nextRun === null) { + throw new Error('Invalid cron expression or no matching time found within scan window.'); + } + firesAt = nextRun; + delaySeconds = Math.round((nextRun - now) / 1000); + } else if (input.delaySeconds !== undefined) { + if (input.delaySeconds < 1 || input.delaySeconds > MAX_DELAY_SECONDS) { + throw new Error(`delay_seconds must be between 1 and ${MAX_DELAY_SECONDS}.`); + } + delaySeconds = input.delaySeconds; + const delayMs = delaySeconds * 1000; + // Apply jitter to the initial fire time + const jitter = computeJitter(delayMs, recurring, this.deps.random); + const adjustedDelay = Math.max(0, delayMs + jitter); + firesAt = now + adjustedDelay; + } else { + throw new Error('Either cronExpression or delaySeconds must be provided.'); + } + + const record: WakeupRecord = { + id: this.deps.newId(), + sessionId, + message: input.message, + reason: input.reason, + scheduledAt: now, + firesAt, + delaySeconds, + recurring, + cronExpression: input.cronExpression, + status: 'pending', + expiresAt: recurring ? now + MAX_RECURRING_AGE_MS : null, + fireAttempts: 0, + deferredFires: [], + }; + + this.records.set(record.id, record); + this.scheduleTimer(record); + return record; + } + + cancel(wakeupId: string): boolean { + const record = this.records.get(wakeupId); + if (!record || record.status !== 'pending') return false; + record.status = 'cancelled'; + const timer = this.timers.get(wakeupId); + if (timer !== undefined) { + this.deps.clearTimer(timer); + this.timers.delete(wakeupId); + } + this.retries.delete(wakeupId); + return true; + } + + cancelAllForSession(sessionId: string): void { + for (const record of this.records.values()) { + if (record.sessionId === sessionId && record.status === 'pending') { + this.cancel(record.id); + } + } + } + + listForSession(sessionId: string): WakeupRecord[] { + return [...this.records.values()].filter(r => r.sessionId === sessionId); + } + + dispose(): void { + this.disposed = true; + for (const timer of this.timers.values()) { + this.deps.clearTimer(timer); + } + for (const record of this.records.values()) { + if (record.status === 'pending') record.status = 'cancelled'; + } + this.timers.clear(); + this.records.clear(); + this.retries.clear(); + } + + private scheduleTimer(record: WakeupRecord): void { + const delay = Math.max(0, record.firesAt - this.deps.now()); + const timer = this.deps.setTimer(() => { + this.timers.delete(record.id); + void this.fire(record); + }, delay); + this.timers.set(record.id, timer); + } + + private async fire(record: WakeupRecord): Promise { + if (this.disposed) return; + if (record.status !== 'pending') return; + + // Track fire attempts for idle-gate observability + record.fireAttempts += 1; + + // Auto-expire check for recurring jobs + if (record.recurring && record.expiresAt !== null) { + const now = this.deps.now(); + if (now >= record.expiresAt) { + record.status = 'expired'; + this.retries.delete(record.id); + return; + } + } + + const canFire = await this.deps.canFire(record.sessionId).catch(() => false); + // Re-check after async gap: cancel() or dispose() may have changed status + if (record.status !== 'pending') return; + if (!canFire) { + // Idle-gate: log the deferred fire attempt + record.deferredFires.push(this.deps.now()); + + const retryCount = (this.retries.get(record.id) ?? 0) + 1; + this.retries.set(record.id, retryCount); + if (retryCount >= MAX_FIRE_RETRIES) { + record.status = 'expired'; + this.retries.delete(record.id); + return; + } + const backoffTimer = this.deps.setTimer(() => { + this.timers.delete(record.id); + void this.fire(record); + }, BACKOFF_MS); + this.timers.set(record.id, backoffTimer); + return; + } + + record.status = 'fired'; + this.retries.delete(record.id); + const turnId = this.deps.newId(); + this.deps.injectTurn(record.sessionId, { + turnId, + text: `[Scheduled wakeup: ${record.reason}]\n\n${record.message}`, + }); + + if (record.recurring) { + const now = this.deps.now(); + + // Check auto-expire before scheduling next occurrence + if (record.expiresAt !== null && now >= record.expiresAt) { + // Chain has expired; do not re-schedule + return; + } + + let nextFiresAt: number; + let nextDelaySeconds: number; + + if (record.cronExpression) { + // For cron-based recurring: compute the NEXT cron match after now + const nextRun = computeNextCronRun(record.cronExpression, now); + if (nextRun === null) { + // No future match found — stop recurring + return; + } + nextFiresAt = nextRun; + nextDelaySeconds = Math.round((nextRun - now) / 1000); + } else { + // Fixed-delay recurring: same delay as before, with jitter + const delayMs = record.delaySeconds * 1000; + const jitter = computeJitter(delayMs, true, this.deps.random); + const adjustedDelay = Math.max(0, delayMs + jitter); + nextFiresAt = now + adjustedDelay; + nextDelaySeconds = record.delaySeconds; + } + + const next: WakeupRecord = { + id: this.deps.newId(), + sessionId: record.sessionId, + message: record.message, + reason: record.reason, + scheduledAt: now, + firesAt: nextFiresAt, + delaySeconds: nextDelaySeconds, + recurring: true, + cronExpression: record.cronExpression, + status: 'pending', + // Inherit the original expiresAt from the chain (not reset on each occurrence) + expiresAt: record.expiresAt, + fireAttempts: 0, + deferredFires: [], + }; + this.records.set(next.id, next); + this.scheduleTimer(next); + } + } +} diff --git a/packages/runtime/src/wakeup-tools.ts b/packages/runtime/src/wakeup-tools.ts new file mode 100644 index 0000000000..44da06e851 --- /dev/null +++ b/packages/runtime/src/wakeup-tools.ts @@ -0,0 +1,118 @@ +/** + * Agent-facing cron tools aligned with Claude Code's CronCreate/CronDelete/CronList. + * + * CronCreate schedules a prompt to fire at a future time — either recurring + * (cron expression) or one-shot. Session-only by default (v1: no durable mode). + */ +import { z } from 'zod'; +import type { MakaTool } from './tool-runtime.js'; +import type { WakeupScheduler } from './wakeup-scheduler.js'; + +export const CRON_CREATE_TOOL_NAME = 'CronCreate'; +export const CRON_DELETE_TOOL_NAME = 'CronDelete'; +export const CRON_LIST_TOOL_NAME = 'CronList'; + +export function buildCronCreateTool(scheduler: WakeupScheduler): MakaTool< + { delay_seconds?: number; cron?: string; prompt: string; reason: string; recurring?: boolean }, + unknown +> { + return { + name: CRON_CREATE_TOOL_NAME, + displayName: '定时任务', + description: + 'Schedule a prompt to run at a future time within this session. ' + + 'When the timer fires, the prompt is injected as a new turn and the agent continues working. ' + + 'Use for polling, monitoring, periodic checks, or delayed actions. ' + + 'Provide either delay_seconds (one-shot/fixed-interval) or cron (5-field cron expression). ' + + 'Uses standard 5-field cron in the user\'s local timezone. ' + + 'Jobs are session-only (in-memory, gone when the session ends). ' + + 'Max 5 pending jobs per session, delay 1-86400 seconds.', + parameters: z.object({ + delay_seconds: z.number().int().min(1).max(86400).optional() + .describe('Seconds from now to fire. Min 1, max 86400 (24h). Either this or cron must be provided.'), + cron: z.string().max(100).optional() + .describe('Standard 5-field cron expression (minute hour day-of-month month day-of-week). Uses the user\'s local timezone. Either this or delay_seconds must be provided.'), + prompt: z.string().min(1).max(10000) + .describe('The prompt/instruction to execute when the timer fires.'), + reason: z.string().min(1).max(200) + .describe('One short sentence explaining the chosen schedule. Shown to the user.'), + recurring: z.boolean().optional() + .describe('If true, re-schedule the same job after each fire. Default false (one-shot).'), + }).refine( + data => data.delay_seconds !== undefined || (data.cron !== undefined && data.cron.length > 0), + { message: 'Either delay_seconds or cron must be provided.' }, + ), + categoryHint: 'custom_tool', + impl: async (input, ctx) => { + try { + const record = scheduler.schedule(ctx.sessionId, { + delaySeconds: input.delay_seconds, + cronExpression: input.cron, + message: input.prompt, + reason: input.reason, + recurring: input.recurring, + }); + return { + ok: true, + job_id: record.id, + fires_at: record.firesAt, + fires_in_seconds: record.delaySeconds, + recurring: input.recurring ?? false, + cron: input.cron ?? null, + }; + } catch (error) { + return { ok: false, error: error instanceof Error ? error.message : String(error) }; + } + }, + }; +} + +export function buildCronDeleteTool(scheduler: WakeupScheduler): MakaTool<{ job_id: string }, unknown> { + return { + name: CRON_DELETE_TOOL_NAME, + displayName: '取消定时任务', + description: 'Cancel a scheduled cron job by its ID. Use CronList to find job IDs.', + parameters: z.object({ + job_id: z.string().min(1).describe('The job_id returned by CronCreate.'), + }), + categoryHint: 'custom_tool', + impl: async ({ job_id }) => { + const cancelled = scheduler.cancel(job_id); + return { ok: true, cancelled }; + }, + }; +} + +export function buildCronListTool(scheduler: WakeupScheduler): MakaTool, unknown> { + return { + name: CRON_LIST_TOOL_NAME, + displayName: '列出定时任务', + description: 'List all scheduled cron jobs in this session.', + parameters: z.object({}), + categoryHint: 'custom_tool', + impl: async (_input, ctx) => { + const records = scheduler.listForSession(ctx.sessionId); + return { + ok: true, + count: records.length, + jobs: records.map(r => ({ + job_id: r.id, + status: r.status, + prompt: r.message.slice(0, 100), + reason: r.reason, + fires_at: r.firesAt, + recurring: r.recurring ?? false, + cron: r.cronExpression ?? null, + })), + }; + }, + }; +} + +export function buildCronTools(scheduler: WakeupScheduler): MakaTool[] { + return [ + buildCronCreateTool(scheduler), + buildCronDeleteTool(scheduler), + buildCronListTool(scheduler), + ] as MakaTool[]; +} diff --git a/packages/ui/src/tool-activity.tsx b/packages/ui/src/tool-activity.tsx index 7a1d063790..b529bda642 100644 --- a/packages/ui/src/tool-activity.tsx +++ b/packages/ui/src/tool-activity.tsx @@ -1,6 +1,6 @@ import { useEffect, useRef } from 'react'; import { type ToolResultContent } from '@maka/core'; -import { AlertOctagon, Check, Copy, X } from './icons.js'; +import { AlertOctagon, Check, Clock, Copy, Repeat, X } from './icons.js'; import { useClipboardCopyFeedback } from './clipboard-feedback.js'; import { detectUiLocale } from './locale-helpers.js'; import { type ToolActivityItem, type ToolOutputChunk } from './materialize.js'; @@ -48,6 +48,84 @@ function LoadToolResultPreview(props: { args: unknown; value: unknown }) { ); } +// ── CronJob result preview ────────────────────────────────────────────────── + +const CRON_TOOL_NAMES: ReadonlySet = new Set(['CronCreate', 'CronDelete', 'CronList']); + +function isCronTool(name: string): boolean { + return CRON_TOOL_NAMES.has(name); +} + +/** Compact preview card for CronCreate / CronDelete / CronList results. */ +function CronJobResultPreview(props: { toolName: string; value: unknown }) { + const data = props.value && typeof props.value === 'object' ? (props.value as Record) : {}; + const ok = data.ok === true; + + if (props.toolName === 'CronCreate' && ok) { + const recurring = data.recurring === true; + const firesIn = typeof data.fires_in_seconds === 'number' ? data.fires_in_seconds : null; + const cron = typeof data.cron === 'string' ? data.cron : null; + const jobId = typeof data.job_id === 'string' ? data.job_id : null; + const Icon = recurring ? Repeat : Clock; + return ( +
+

+

+ {firesIn !== null && ( +

+ {firesIn < 60 ? `${firesIn}s 后触发` : `${Math.round(firesIn / 60)}min 后触发`} +

+ )} + {cron &&

cron: {cron}

} + {jobId &&

job_id: {jobId.slice(0, 8)}

} +
+ ); + } + + if (props.toolName === 'CronDelete' && ok) { + const cancelled = data.cancelled === true; + return ( +
+

+

+
+ ); + } + + if (props.toolName === 'CronList' && ok) { + const count = typeof data.count === 'number' ? data.count : 0; + const jobs = Array.isArray(data.jobs) ? data.jobs : []; + return ( +
+

+

+ {jobs.slice(0, 5).map((job, i) => { + const j = job && typeof job === 'object' ? (job as Record) : {}; + const recurring = j.recurring === true; + const status = typeof j.status === 'string' ? j.status : 'unknown'; + const reason = typeof j.reason === 'string' ? j.reason : ''; + const JobIcon = recurring ? Repeat : Clock; + return ( +

+

+ ); + })} +
+ ); + } + + // Fallback for error or unexpected shape + return ; +} + const STATUS_LABEL: Record = { pending: '排队中', waiting_permission: '等待权限', @@ -141,6 +219,8 @@ export function ToolActivity(props: { items: ToolActivityItem[] }) { {item.result && !permissionDenied && ( isConnectorTool(item.toolName) && item.result.kind === 'json' ? ( + ) : isCronTool(item.toolName) && item.result.kind === 'json' ? ( + ) : ( ) From cd37517bc7977cfb2dea9fce637a901554168413 Mon Sep 17 00:00:00 2001 From: hqhq1025 <1506751656@qq.com> Date: Sun, 5 Jul 2026 17:38:59 +0800 Subject: [PATCH 2/2] fix(runtime): prevent unbounded CronJob record accumulation Recurring cron jobs were accumulating ALL historical fired records indefinitely. When CronList was called, it returned all records (800+ observed), triggering context budget archiving and creating hundreds of tool-result-active-* files. Three fixes applied: 1. Non-recurring jobs: remove from records after successful fire (no reason to keep them; they'll never fire again) 2. Recurring jobs: reuse the same record in-place instead of creating a new record on each fire. The record stays in the map with its original id, just gets updated firesAt/status. 3. Safety cap: MAX_RECORDS_PER_SESSION = 50. pruneSession() drops oldest terminal (fired/expired/cancelled) records when exceeded. Additionally, CronList now only returns active (pending) jobs by default via listForSession({ activeOnly: true }), preventing the full history from being serialized into tool results. --- .../src/__tests__/wakeup-scheduler.test.ts | 96 ++++++++++++++++--- packages/runtime/src/wakeup-scheduler.ts | 69 ++++++++----- packages/runtime/src/wakeup-tools.ts | 5 +- 3 files changed, 133 insertions(+), 37 deletions(-) diff --git a/packages/runtime/src/__tests__/wakeup-scheduler.test.ts b/packages/runtime/src/__tests__/wakeup-scheduler.test.ts index 9b0352d5a2..8772ffc65a 100644 --- a/packages/runtime/src/__tests__/wakeup-scheduler.test.ts +++ b/packages/runtime/src/__tests__/wakeup-scheduler.test.ts @@ -1,6 +1,6 @@ import assert from 'node:assert/strict'; import { describe, test } from 'node:test'; -import { WakeupScheduler, computeNextCronRun, computeJitter } from '../wakeup-scheduler.js'; +import { WakeupScheduler, computeNextCronRun, computeJitter, MAX_RECORDS_PER_SESSION } from '../wakeup-scheduler.js'; function createTestScheduler(overrides: Partial[0]> = {}) { const fired: Array<{ sessionId: string; turnId: string; text: string }> = []; @@ -66,12 +66,15 @@ describe('WakeupScheduler', () => { assert.ok(fired[0].text.includes('hello wakeup')); }); - test('fired record status is updated', async () => { + test('fired non-recurring record is removed from scheduler', async () => { const { scheduler, fireNextTimer } = createTestScheduler(); const record = scheduler.schedule('session-1', { delaySeconds: 5, message: 'check', reason: 'test' }); await fireNextTimer(); + // The record object still reflects the fired status + assert.equal(record.status, 'fired'); + // But it is no longer in the scheduler's records (cleaned up after fire) const records = scheduler.listForSession('session-1'); - assert.equal(records.find(r => r.id === record.id)?.status, 'fired'); + assert.equal(records.length, 0, 'fired non-recurring record should be removed'); }); test('cancel prevents firing', () => { @@ -223,7 +226,7 @@ describe('WakeupScheduler', () => { ); }); - test('cron-based recurring job reschedules after firing', async () => { + test('cron-based recurring job reschedules in-place after firing', async () => { const { scheduler, fired, fireNextTimer } = createTestScheduler(); const record = scheduler.schedule('session-1', { cronExpression: '* * * * *', @@ -231,12 +234,14 @@ describe('WakeupScheduler', () => { reason: 'every minute', recurring: true, }); + const originalId = record.id; await fireNextTimer(); assert.equal(fired.length, 1); assert.ok(fired[0].text.includes('cron recurring')); - // Should have created a new pending record for the next occurrence + // The same record should be reused in-place (no new record created) const pending = scheduler.listForSession('session-1').filter(r => r.status === 'pending'); - assert.equal(pending.length, 1, 'should have one new pending record'); + assert.equal(pending.length, 1, 'should have one pending record'); + assert.equal(pending[0].id, originalId, 'should reuse the same record id'); assert.equal(pending[0].cronExpression, '* * * * *'); }); @@ -307,16 +312,17 @@ describe('WakeupScheduler', () => { }); // First fire attempt: rejected await fireNextTimer(); - const afterFirst = scheduler.listForSession('session-1').find(r => r.id === record.id)!; - assert.equal(afterFirst.fireAttempts, 1); - assert.equal(afterFirst.deferredFires.length, 1); + // Record is still pending (not yet fired), so it remains in the scheduler + assert.equal(record.fireAttempts, 1); + assert.equal(record.deferredFires.length, 1); // Second fire attempt: succeeds await fireNextTimer(); - const afterSecond = scheduler.listForSession('session-1').find(r => r.id === record.id)!; - assert.equal(afterSecond.fireAttempts, 2); - assert.equal(afterSecond.status, 'fired'); + // After successful fire, the non-recurring record is removed from the map + // but the object reference still has the final state + assert.equal(record.fireAttempts, 2); + assert.equal(record.status, 'fired'); // deferredFires should still have 1 entry (only logged on rejection) - assert.equal(afterSecond.deferredFires.length, 1); + assert.equal(record.deferredFires.length, 1); }); // ─── Cancel edge-case tests ──────────────────────────────────────────────── @@ -376,4 +382,68 @@ describe('WakeupScheduler', () => { assert.ok(jitter >= -90_000, `one-shot jitter should be >= -90000, got ${jitter}`); } }); + + // ─── Record accumulation prevention tests ───────────────────────────────── + + test('fired non-recurring job is removed from records', async () => { + const { scheduler, fired, fireNextTimer } = createTestScheduler(); + scheduler.schedule('session-1', { delaySeconds: 5, message: 'one-shot', reason: 'test' }); + await fireNextTimer(); + assert.equal(fired.length, 1); + // The record should be removed after firing + const records = scheduler.listForSession('session-1'); + assert.equal(records.length, 0, 'fired non-recurring job should be removed'); + }); + + test('recurring job does not duplicate records on fire', async () => { + const { scheduler, fired, fireNextTimer } = createTestScheduler(); + scheduler.schedule('session-1', { + delaySeconds: 60, + message: 'recurring-check', + reason: 'test', + recurring: true, + }); + // Fire multiple times + await fireNextTimer(); + await fireNextTimer(); + await fireNextTimer(); + assert.equal(fired.length, 3, 'should have fired 3 times'); + // Only ONE record should exist (reused in-place) + const records = scheduler.listForSession('session-1'); + assert.equal(records.length, 1, 'should have exactly one record for recurring job'); + assert.equal(records[0].status, 'pending', 'record should be pending for next fire'); + }); + + test('max records cap is enforced', () => { + const { scheduler } = createTestScheduler(); + // Create many records that exceed the cap by scheduling and cancelling + for (let i = 0; i < MAX_RECORDS_PER_SESSION + 10; i++) { + const rec = scheduler.schedule('session-1', { + delaySeconds: 60, + message: `job-${i}`, + reason: 'test', + }); + scheduler.cancel(rec.id); + } + // All records are cancelled (terminal), schedule() should prune + // Schedule one more to trigger pruning + scheduler.schedule('session-1', { delaySeconds: 60, message: 'final', reason: 'test' }); + const records = scheduler.listForSession('session-1'); + assert.ok(records.length <= MAX_RECORDS_PER_SESSION, `records (${records.length}) should not exceed cap (${MAX_RECORDS_PER_SESSION})`); + }); + + test('listForSession with activeOnly only returns pending records', async () => { + const { scheduler, fireNextTimer } = createTestScheduler(); + scheduler.schedule('session-1', { delaySeconds: 5, message: 'will-fire', reason: 'test', recurring: true }); + const cancelMe = scheduler.schedule('session-1', { delaySeconds: 30, message: 'will-cancel', reason: 'test' }); + scheduler.cancel(cancelMe.id); + // Fire the recurring job once (it stays as pending for next) + await fireNextTimer(); + // Now we have: 1 pending (recurring, re-scheduled) + 1 cancelled + const all = scheduler.listForSession('session-1'); + const active = scheduler.listForSession('session-1', { activeOnly: true }); + assert.equal(all.length, 2, 'all records includes cancelled'); + assert.equal(active.length, 1, 'activeOnly returns only pending'); + assert.equal(active[0].status, 'pending'); + }); }); diff --git a/packages/runtime/src/wakeup-scheduler.ts b/packages/runtime/src/wakeup-scheduler.ts index f5dc4eeac3..601257b36a 100644 --- a/packages/runtime/src/wakeup-scheduler.ts +++ b/packages/runtime/src/wakeup-scheduler.ts @@ -182,6 +182,9 @@ const MAX_JITTER_MS = 15 * 60 * 1000; /** Maximum early jitter for one-shot jobs firing on round minutes: 90 seconds. */ const ONE_SHOT_JITTER_MS = 90 * 1000; +/** Maximum total records per session before oldest fired/expired records are pruned. */ +export const MAX_RECORDS_PER_SESSION = 50; + /** * Compute jitter to add to a scheduled delay. * @@ -283,6 +286,7 @@ export class WakeupScheduler { }; this.records.set(record.id, record); + this.pruneSession(sessionId); this.scheduleTimer(record); return record; } @@ -308,8 +312,31 @@ export class WakeupScheduler { } } - listForSession(sessionId: string): WakeupRecord[] { - return [...this.records.values()].filter(r => r.sessionId === sessionId); + listForSession(sessionId: string, opts?: { activeOnly?: boolean }): WakeupRecord[] { + const all = [...this.records.values()].filter(r => r.sessionId === sessionId); + if (opts?.activeOnly) { + return all.filter(r => r.status === 'pending'); + } + return all; + } + + /** + * Remove fired/expired/cancelled records for a session that exceed the cap. + * Keeps all pending records; drops oldest terminal records first. + */ + private pruneSession(sessionId: string): void { + const sessionRecords = [...this.records.values()].filter(r => r.sessionId === sessionId); + if (sessionRecords.length <= MAX_RECORDS_PER_SESSION) return; + + // Sort terminal records by scheduledAt ascending (oldest first) + const terminal = sessionRecords + .filter(r => r.status === 'fired' || r.status === 'expired' || r.status === 'cancelled') + .sort((a, b) => a.scheduledAt - b.scheduledAt); + + const excess = sessionRecords.length - MAX_RECORDS_PER_SESSION; + for (let i = 0; i < excess && i < terminal.length; i++) { + this.records.delete(terminal[i].id); + } } dispose(): void { @@ -386,7 +413,9 @@ export class WakeupScheduler { // Check auto-expire before scheduling next occurrence if (record.expiresAt !== null && now >= record.expiresAt) { - // Chain has expired; do not re-schedule + // Chain has expired; do not re-schedule. Remove the fired record. + this.records.delete(record.id); + this.pruneSession(record.sessionId); return; } @@ -397,7 +426,9 @@ export class WakeupScheduler { // For cron-based recurring: compute the NEXT cron match after now const nextRun = computeNextCronRun(record.cronExpression, now); if (nextRun === null) { - // No future match found — stop recurring + // No future match found — stop recurring. Remove the fired record. + this.records.delete(record.id); + this.pruneSession(record.sessionId); return; } nextFiresAt = nextRun; @@ -411,24 +442,18 @@ export class WakeupScheduler { nextDelaySeconds = record.delaySeconds; } - const next: WakeupRecord = { - id: this.deps.newId(), - sessionId: record.sessionId, - message: record.message, - reason: record.reason, - scheduledAt: now, - firesAt: nextFiresAt, - delaySeconds: nextDelaySeconds, - recurring: true, - cronExpression: record.cronExpression, - status: 'pending', - // Inherit the original expiresAt from the chain (not reset on each occurrence) - expiresAt: record.expiresAt, - fireAttempts: 0, - deferredFires: [], - }; - this.records.set(next.id, next); - this.scheduleTimer(next); + // Reuse the same record in-place: update fields for next occurrence + record.status = 'pending'; + record.scheduledAt = now; + record.firesAt = nextFiresAt; + record.delaySeconds = nextDelaySeconds; + record.fireAttempts = 0; + record.deferredFires = []; + this.scheduleTimer(record); + } else { + // Non-recurring job: remove from records after successful fire + this.records.delete(record.id); + this.pruneSession(record.sessionId); } } } diff --git a/packages/runtime/src/wakeup-tools.ts b/packages/runtime/src/wakeup-tools.ts index 44da06e851..d97cc3572e 100644 --- a/packages/runtime/src/wakeup-tools.ts +++ b/packages/runtime/src/wakeup-tools.ts @@ -87,11 +87,11 @@ export function buildCronListTool(scheduler: WakeupScheduler): MakaTool { - const records = scheduler.listForSession(ctx.sessionId); + const records = scheduler.listForSession(ctx.sessionId, { activeOnly: true }); return { ok: true, count: records.length, @@ -103,6 +103,7 @@ export function buildCronListTool(scheduler: WakeupScheduler): MakaTool