From e96f255bb3ea028780616a1ad4593ac2ccd94e6e Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 27 May 2026 19:04:32 +0800 Subject: [PATCH 01/26] feat(agent-core): add cron ClockSources abstraction ClockSources splits wall-clock and monotonic time so the cron scheduler can be driven by an injected/simulated clock without breaking the lock heartbeat. resolveClockSources reads KIMI_CRON_CLOCK to switch between system, env-var-backed, and file-backed wall clocks; monotonic time is always process.hrtime.bigint() and never overridable. --- packages/agent-core/src/tools/cron/clock.ts | 154 ++++++++++++++ .../agent-core/test/tools/cron/clock.test.ts | 194 ++++++++++++++++++ 2 files changed, 348 insertions(+) create mode 100644 packages/agent-core/src/tools/cron/clock.ts create mode 100644 packages/agent-core/test/tools/cron/clock.test.ts diff --git a/packages/agent-core/src/tools/cron/clock.ts b/packages/agent-core/src/tools/cron/clock.ts new file mode 100644 index 0000000000..5dacff09c1 --- /dev/null +++ b/packages/agent-core/src/tools/cron/clock.ts @@ -0,0 +1,154 @@ +/** + * Clock sources for the cron scheduler. + * + * Two distinct notions of time are kept apart on purpose: + * + * 1. wall-clock — what the user perceives as "the current time". Used + * for cron expression matching, `createdAt`, and the 7-day stale + * judgment. May be overridden in tests / multi-process benches so + * that scenarios can run in simulated time without `setTimeout`. + * + * 2. monotonic ms — a strictly non-decreasing counter that never + * jumps backwards across NTP adjustments, suspend/resume, or + * simulated-clock injection. Used for the poll cadence and the + * lock heartbeat — anything where "did 5 seconds elapse since we + * last looked" must hold even when the wall clock is frozen. + * + * Mixing the two pollutes test reproducibility: a heartbeat tied to + * `wallNow()` will appear stuck when the test clock is frozen; a cron + * fire tied to `monoNowMs()` will not advance when the bench rewinds + * the simulated day. Every component in `tools/cron/` MUST take a + * `ClockSources` and route every time read through it. + * + * `monoNowMs` is ALWAYS `process.hrtime.bigint()` (converted to ms). + * It is not overridable — accepting an external monotonic clock would + * defeat the safety net the lock heartbeat depends on. + * + * `wallNow` resolution is driven by the `KIMI_CRON_CLOCK` env var; see + * `resolveClockSources` below. Defaults to `Date.now()`. + */ +import { readFileSync } from 'node:fs'; + +export interface ClockSources { + /** + * Wall-clock epoch milliseconds. May be overridden in tests / bench + * via `KIMI_CRON_CLOCK`. Used for cron matching, `createdAt`, stale + * judgment. + */ + wallNow(): number; + + /** + * Strictly monotonic millisecond counter. Never overridden. Used for + * the 1-second poll cadence and the lock-heartbeat liveness window. + */ + monoNowMs(): number; +} + +const systemMonoNowMs = (): number => Number(process.hrtime.bigint() / 1_000_000n); + +/** + * Production default — `Date.now()` + `process.hrtime.bigint()`. Used + * whenever `KIMI_CRON_CLOCK` is unset, set to `"system"`, or set to a + * spec that fails to parse. + */ +export const SYSTEM_CLOCKS: ClockSources = { + wallNow: () => Date.now(), + monoNowMs: systemMonoNowMs, +}; + +/** + * Resolve a `ClockSources` implementation from a spec string (typically + * `process.env.KIMI_CRON_CLOCK`). + * + * unset / `"system"` → {@link SYSTEM_CLOCKS} + * `"env:VAR_NAME"` → `wallNow` reads `process.env[VAR_NAME]` on + * every call and parses it as `Number(...)`. + * A missing or unparseable value falls back to + * `Date.now()` for that single call so a + * mis-set env never bricks the scheduler. + * `"file:"` → `wallNow` reads the first line of `` + * on every call (sync — the tick path is not + * async) and parses it as `Number(...)`. A + * missing file or bad parse falls back to + * `Date.now()` for that call. Used so a + * multi-process bench can share a single + * file-backed simulated clock. + * + * `monoNowMs` ALWAYS uses `process.hrtime.bigint()`. No spec overrides + * it — see file header. + * + * Each `wallNow()` call re-reads its source. We deliberately do NOT + * cache, because a multi-process bench tick mutating the file must be + * picked up by every reader immediately; a cache would silently lock + * each process to its first observation. + * + * Unrecognised specs fall back to {@link SYSTEM_CLOCKS} (with a + * debug-log on stderr). This is deliberate — bricking the agent on a + * typoed bench env var would be worse than running with system time. + */ +export function resolveClockSources(spec?: string): ClockSources { + if (spec === undefined || spec === '' || spec === 'system') { + return SYSTEM_CLOCKS; + } + + if (spec.startsWith('env:')) { + const varName = spec.slice('env:'.length); + if (varName === '') { + debugInvalidSpec(spec, 'empty env var name'); + return SYSTEM_CLOCKS; + } + return { + wallNow: () => readEnvWall(varName), + monoNowMs: systemMonoNowMs, + }; + } + + if (spec.startsWith('file:')) { + const filePath = spec.slice('file:'.length); + if (filePath === '') { + debugInvalidSpec(spec, 'empty file path'); + return SYSTEM_CLOCKS; + } + return { + wallNow: () => readFileWall(filePath), + monoNowMs: systemMonoNowMs, + }; + } + + debugInvalidSpec(spec, 'unrecognised scheme'); + return SYSTEM_CLOCKS; +} + +function readEnvWall(varName: string): number { + const raw = process.env[varName]; + if (raw === undefined || raw === '') return Date.now(); + const parsed = Number(raw); + if (!Number.isFinite(parsed)) return Date.now(); + return parsed; +} + +function readFileWall(filePath: string): number { + let raw: string; + try { + raw = readFileSync(filePath, 'utf8'); + } catch { + return Date.now(); + } + const firstLine = raw.split('\n', 1)[0]?.trim() ?? ''; + if (firstLine === '') return Date.now(); + const parsed = Number(firstLine); + if (!Number.isFinite(parsed)) return Date.now(); + return parsed; +} + +function debugInvalidSpec(spec: string, reason: string): void { + // We do not pull in a logger here — `clock.ts` is the lowest layer of + // the cron module and must stay dependency-free so it can be imported + // from anywhere (including lint rules, type files). A stderr write + // gated on KIMI_CRON_DEBUG is enough — production is silent. + if (process.env.KIMI_CRON_DEBUG === '1') { + process.stderr.write( + `[cron/clock] invalid KIMI_CRON_CLOCK spec ${JSON.stringify(spec)}: ${reason} — falling back to system clock\n`, + ); + } +} diff --git a/packages/agent-core/test/tools/cron/clock.test.ts b/packages/agent-core/test/tools/cron/clock.test.ts new file mode 100644 index 0000000000..1fa09a5d06 --- /dev/null +++ b/packages/agent-core/test/tools/cron/clock.test.ts @@ -0,0 +1,194 @@ +/** + * Tests for `tools/cron/clock.ts`. + * + * Lives under `test/tools/cron/` because the package's `vitest.config.ts` + * scopes its `include` to `test/**`. Colocated `src/**\/*.test.ts` would + * not run in CI. The plan doc names the file `src/tools/cron/clock.test.ts`; + * we put it here so the test actually runs. + */ +import { mkdtempSync, writeFileSync } from 'node:fs'; +import * as os from 'node:os'; +import { afterEach, describe, expect, it } from 'vitest'; +import { join } from 'pathe'; + +import { resolveClockSources, SYSTEM_CLOCKS } from '../../../src/tools/cron/clock'; + +describe('clock.ts', () => { + describe('SYSTEM_CLOCKS', () => { + it('monoNowMs is strictly non-decreasing across 1000 calls', () => { + let prev = SYSTEM_CLOCKS.monoNowMs(); + for (let i = 0; i < 1000; i++) { + const next = SYSTEM_CLOCKS.monoNowMs(); + expect(next).toBeGreaterThanOrEqual(prev); + prev = next; + } + }); + + it('wallNow is close to Date.now', () => { + const before = Date.now(); + const sample = SYSTEM_CLOCKS.wallNow(); + const after = Date.now(); + expect(sample).toBeGreaterThanOrEqual(before); + expect(sample).toBeLessThanOrEqual(after); + }); + + it('monoNowMs returns finite positive numbers', () => { + const sample = SYSTEM_CLOCKS.monoNowMs(); + expect(Number.isFinite(sample)).toBe(true); + expect(sample).toBeGreaterThan(0); + }); + }); + + describe('resolveClockSources — default / system', () => { + it('undefined spec returns SYSTEM_CLOCKS', () => { + expect(resolveClockSources(undefined)).toBe(SYSTEM_CLOCKS); + }); + + it('empty string returns SYSTEM_CLOCKS', () => { + expect(resolveClockSources('')).toBe(SYSTEM_CLOCKS); + }); + + it('"system" returns SYSTEM_CLOCKS', () => { + expect(resolveClockSources('system')).toBe(SYSTEM_CLOCKS); + }); + + it('unrecognised scheme falls back to SYSTEM_CLOCKS', () => { + expect(resolveClockSources('garbage:foo')).toBe(SYSTEM_CLOCKS); + expect(resolveClockSources('foobar')).toBe(SYSTEM_CLOCKS); + }); + }); + + describe('resolveClockSources — env:VAR', () => { + const ENV_VAR = '__KIMI_CRON_CLOCK_TEST_FAKE_NOW__'; + + afterEach(() => { + delete process.env[ENV_VAR]; + }); + + it('reads process.env on every wallNow call', () => { + const clocks = resolveClockSources(`env:${ENV_VAR}`); + + process.env[ENV_VAR] = '1000'; + expect(clocks.wallNow()).toBe(1000); + + process.env[ENV_VAR] = '2000'; + expect(clocks.wallNow()).toBe(2000); + + process.env[ENV_VAR] = '3500'; + expect(clocks.wallNow()).toBe(3500); + }); + + it('monoNowMs is not affected by env clock', () => { + const clocks = resolveClockSources(`env:${ENV_VAR}`); + process.env[ENV_VAR] = '1000'; + // mono must still be a real, finite, growing number — not the + // env value. + const a = clocks.monoNowMs(); + const b = clocks.monoNowMs(); + expect(a).not.toBe(1000); + expect(b).toBeGreaterThanOrEqual(a); + }); + + it('missing env var falls back to Date.now for that call', () => { + const clocks = resolveClockSources(`env:${ENV_VAR}`); + // ENV_VAR intentionally unset. + const before = Date.now(); + const sample = clocks.wallNow(); + const after = Date.now(); + expect(sample).toBeGreaterThanOrEqual(before); + expect(sample).toBeLessThanOrEqual(after); + }); + + it('unparseable env value falls back to Date.now for that call', () => { + const clocks = resolveClockSources(`env:${ENV_VAR}`); + process.env[ENV_VAR] = 'not-a-number'; + const before = Date.now(); + const sample = clocks.wallNow(); + const after = Date.now(); + expect(sample).toBeGreaterThanOrEqual(before); + expect(sample).toBeLessThanOrEqual(after); + }); + + it('empty env value falls back to Date.now', () => { + const clocks = resolveClockSources(`env:${ENV_VAR}`); + process.env[ENV_VAR] = ''; + const before = Date.now(); + const sample = clocks.wallNow(); + const after = Date.now(); + expect(sample).toBeGreaterThanOrEqual(before); + expect(sample).toBeLessThanOrEqual(after); + }); + + it('empty env var name in spec falls back to SYSTEM_CLOCKS', () => { + expect(resolveClockSources('env:')).toBe(SYSTEM_CLOCKS); + }); + }); + + describe('resolveClockSources — file:', () => { + it('reads file first line on every wallNow call', () => { + const tmpDir = mkdtempSync(join(os.tmpdir(), 'kimi-cron-clock-')); + const filePath = join(tmpDir, 'now.txt'); + + writeFileSync(filePath, '1000\n', 'utf8'); + const clocks = resolveClockSources(`file:${filePath}`); + expect(clocks.wallNow()).toBe(1000); + + writeFileSync(filePath, '2500', 'utf8'); + expect(clocks.wallNow()).toBe(2500); + + // Multi-line — only first line counts. + writeFileSync(filePath, '4242\ngarbage\n', 'utf8'); + expect(clocks.wallNow()).toBe(4242); + }); + + it('missing file falls back to Date.now', () => { + const tmpDir = mkdtempSync(join(os.tmpdir(), 'kimi-cron-clock-')); + const filePath = join(tmpDir, 'never-created.txt'); + const clocks = resolveClockSources(`file:${filePath}`); + const before = Date.now(); + const sample = clocks.wallNow(); + const after = Date.now(); + expect(sample).toBeGreaterThanOrEqual(before); + expect(sample).toBeLessThanOrEqual(after); + }); + + it('unparseable content falls back to Date.now', () => { + const tmpDir = mkdtempSync(join(os.tmpdir(), 'kimi-cron-clock-')); + const filePath = join(tmpDir, 'now.txt'); + writeFileSync(filePath, 'not-a-number\n', 'utf8'); + const clocks = resolveClockSources(`file:${filePath}`); + const before = Date.now(); + const sample = clocks.wallNow(); + const after = Date.now(); + expect(sample).toBeGreaterThanOrEqual(before); + expect(sample).toBeLessThanOrEqual(after); + }); + + it('empty file falls back to Date.now', () => { + const tmpDir = mkdtempSync(join(os.tmpdir(), 'kimi-cron-clock-')); + const filePath = join(tmpDir, 'now.txt'); + writeFileSync(filePath, '', 'utf8'); + const clocks = resolveClockSources(`file:${filePath}`); + const before = Date.now(); + const sample = clocks.wallNow(); + const after = Date.now(); + expect(sample).toBeGreaterThanOrEqual(before); + expect(sample).toBeLessThanOrEqual(after); + }); + + it('monoNowMs is not affected by file clock', () => { + const tmpDir = mkdtempSync(join(os.tmpdir(), 'kimi-cron-clock-')); + const filePath = join(tmpDir, 'now.txt'); + writeFileSync(filePath, '1000', 'utf8'); + const clocks = resolveClockSources(`file:${filePath}`); + const a = clocks.monoNowMs(); + const b = clocks.monoNowMs(); + expect(a).not.toBe(1000); + expect(b).toBeGreaterThanOrEqual(a); + }); + + it('empty file path in spec falls back to SYSTEM_CLOCKS', () => { + expect(resolveClockSources('file:')).toBe(SYSTEM_CLOCKS); + }); + }); +}); From 4edb7ce021bce1bf48c704a06f88b479959a4f29 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 27 May 2026 19:12:21 +0800 Subject: [PATCH 02/26] feat(agent-core): add 5-field cron expression parser MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit parseCronExpression handles the standard 5-field syntax with the cron-style dom/dow OR rule. computeNextCronRun uses field-by-field jumping (not minute scanning) so sparse expressions like '0 12 1 1 *' stay fast. hasFireWithinYears caps the search at 5 years so syntactically legal but never-firing expressions ('0 0 31 2 *') return null instead of spinning forever — required by CronCreate validation. --- .../agent-core/src/tools/cron/cron-expr.ts | 429 ++++++++++++++++++ .../test/tools/cron/cron-expr.test.ts | 325 +++++++++++++ 2 files changed, 754 insertions(+) create mode 100644 packages/agent-core/src/tools/cron/cron-expr.ts create mode 100644 packages/agent-core/test/tools/cron/cron-expr.test.ts diff --git a/packages/agent-core/src/tools/cron/cron-expr.ts b/packages/agent-core/src/tools/cron/cron-expr.ts new file mode 100644 index 0000000000..66c4609844 --- /dev/null +++ b/packages/agent-core/src/tools/cron/cron-expr.ts @@ -0,0 +1,429 @@ +/** + * 5-field cron expression parsing and "next fire time" computation, in + * local time. Self-contained — no external cron library is used because + * upstream `claude-code` mirrors the same semantics and we need exact + * lock-step behaviour with their implementation. + * + * Two flavours of correctness we care about: + * + * 1. **Semantics.** Standard 5 fields (minute hour day-of-month month + * day-of-week). Day-of-month and day-of-week combine with cron's + * OR rule when both are restricted (POSIX/Vixie tradition). dow + * accepts 0..7 with 7 folded to 0 (Sunday). + * + * 2. **Termination.** Computing `next` for a legal-but-never-fires + * expression like `0 0 31 2 *` must not spin. We bound the search + * at a fixed window (5 years by default) and return `null` past + * that — the validator at `CronCreate` reuses this signal. + */ + +/** A parsed cron expression. Opaque to callers — pass it back into {@link computeNextCronRun}. */ +export interface ParsedCronExpression { + readonly raw: string; + readonly minutes: ReadonlySet; + readonly hours: ReadonlySet; + readonly daysOfMonth: ReadonlySet; + readonly months: ReadonlySet; + readonly daysOfWeek: ReadonlySet; + /** True if the source field was `*` — needed so cron's dom/dow OR rule fires only when both are restricted. */ + readonly daysOfMonthWildcard: boolean; + readonly daysOfWeekWildcard: boolean; +} + +const MINUTE_RANGE = { min: 0, max: 59 } as const; +const HOUR_RANGE = { min: 0, max: 23 } as const; +const DOM_RANGE = { min: 1, max: 31 } as const; +const MONTH_RANGE = { min: 1, max: 12 } as const; +const DOW_RANGE = { min: 0, max: 7 } as const; // 7 → 0 fold after parse + +const MS_PER_MINUTE = 60_000; + +/** + * Parse a 5-field cron expression. Throws with a message naming the + * offending field on any syntax error. Whitespace-separated; exactly 5 + * fields. Tokens supported per field: `*`, integers, ranges (`a-b`), + * lists (`a,b,c`), and step (e.g. star-slash-n or `a-b/n`). + */ +export function parseCronExpression(expr: string): ParsedCronExpression { + if (typeof expr !== 'string') { + throw new TypeError('cron expression must be a string'); + } + const trimmed = expr.trim(); + if (trimmed === '') { + throw new Error('cron expression is empty'); + } + const fields = trimmed.split(/\s+/); + if (fields.length !== 5) { + throw new Error( + `cron expression must have exactly 5 fields (minute hour day-of-month month day-of-week); got ${fields.length}`, + ); + } + const [minField, hourField, domField, monthField, dowField] = fields as [ + string, + string, + string, + string, + string, + ]; + + const minutes = parseField(minField, MINUTE_RANGE.min, MINUTE_RANGE.max, 'minute'); + const hours = parseField(hourField, HOUR_RANGE.min, HOUR_RANGE.max, 'hour'); + const daysOfMonth = parseField(domField, DOM_RANGE.min, DOM_RANGE.max, 'day-of-month'); + const months = parseField(monthField, MONTH_RANGE.min, MONTH_RANGE.max, 'month'); + const dowRaw = parseField(dowField, DOW_RANGE.min, DOW_RANGE.max, 'day-of-week'); + const daysOfWeek = new Set(); + for (const v of dowRaw) daysOfWeek.add(v === 7 ? 0 : v); + + return { + raw: trimmed, + minutes, + hours, + daysOfMonth, + months, + daysOfWeek, + daysOfMonthWildcard: isWildcard(domField), + daysOfWeekWildcard: isWildcard(dowField), + }; +} + +function isWildcard(field: string): boolean { + // `*` and `*/n` both leave the field unconstrained in the + // "every value" sense — but only bare `*` should suppress the dom/dow + // OR rule. cron's tradition treats `*/n` as a restriction. + return field === '*'; +} + +function parseField(field: string, min: number, max: number, name: string): Set { + if (field === '') { + throw new Error(`cron ${name} field is empty`); + } + const out = new Set(); + const terms = field.split(','); + for (const term of terms) { + if (term === '') { + throw new Error(`cron ${name} field has empty term in list`); + } + addTerm(out, term, min, max, name); + } + if (out.size === 0) { + throw new Error(`cron ${name} field matches no values`); + } + return out; +} + +function addTerm(out: Set, term: string, min: number, max: number, name: string): void { + let rangePart = term; + let step = 1; + const slash = term.indexOf('/'); + if (slash !== -1) { + rangePart = term.slice(0, slash); + const stepStr = term.slice(slash + 1); + if (stepStr === '') { + throw new Error(`cron ${name} step is empty in "${term}"`); + } + const parsedStep = Number(stepStr); + if (!Number.isInteger(parsedStep) || parsedStep <= 0) { + throw new Error(`cron ${name} step must be a positive integer (got "${stepStr}")`); + } + step = parsedStep; + if (rangePart === '') { + throw new Error(`cron ${name} step needs a range or "*" before "/" in "${term}"`); + } + } + + let lo: number; + let hi: number; + if (rangePart === '*') { + lo = min; + hi = max; + } else { + const dash = rangePart.indexOf('-'); + if (dash === -1) { + const single = Number(rangePart); + if (!Number.isInteger(single)) { + throw new TypeError(`cron ${name} value "${rangePart}" is not an integer`); + } + if (single < min || single > max) { + throw new Error(`cron ${name} value ${single} out of range ${min}..${max}`); + } + // A bare single value with a step (`5/10`) is unusual; treat as + // "from value through max stepping by N", which is what most cron + // dialects do. + if (slash !== -1) { + lo = single; + hi = max; + } else { + out.add(single); + return; + } + } else { + const loStr = rangePart.slice(0, dash); + const hiStr = rangePart.slice(dash + 1); + lo = Number(loStr); + hi = Number(hiStr); + if (!Number.isInteger(lo) || !Number.isInteger(hi)) { + throw new TypeError(`cron ${name} range "${rangePart}" has non-integer bound`); + } + if (lo < min || hi > max || lo > hi) { + throw new Error( + `cron ${name} range ${lo}-${hi} out of bounds (must be ${min}..${max}, ascending)`, + ); + } + } + } + + for (let v = lo; v <= hi; v += step) { + out.add(v); + } +} + +/** + * Find the next wall-clock epoch ms strictly greater than `fromMs` that + * satisfies `expr`, using local-time semantics. Returns `null` if no + * match exists inside the default 5-year search window — defensive + * against legal-but-never-fires expressions like `0 0 31 2 *`. + * + * Uses an O(transitions) field-by-field skip algorithm rather than a + * minute-by-minute scan — month mismatch advances by months, day + * mismatch by days, etc., so the worst case for `0 12 1 1 *` is a + * handful of iterations, not 43 200. + */ +export function computeNextCronRun(expr: ParsedCronExpression, fromMs: number): number | null { + return nextRunWithinMinutes(expr, fromMs, 5 * 366 * 24 * 60); +} + +/** + * True iff at least one fire exists within `years` years of `fromMs`. + * Used by CronCreate validation to reject `0 0 31 2 *` and friends up + * front, with the same search budget {@link computeNextCronRun} uses + * (so the validator never says yes to something the scheduler will + * later refuse to compute). + */ +export function hasFireWithinYears( + expr: ParsedCronExpression, + years: number, + fromMs: number, +): boolean { + const cap = Math.max(1, Math.floor(years * 366 * 24 * 60)); + return nextRunWithinMinutes(expr, fromMs, cap) !== null; +} + +function nextRunWithinMinutes( + expr: ParsedCronExpression, + fromMs: number, + capMinutes: number, +): number | null { + // Seek strictly into the next minute: drop seconds/ms and add one + // minute. This guarantees we never return `fromMs` itself. + const start = new Date(fromMs); + start.setSeconds(0, 0); + const date = new Date(start.getTime() + MS_PER_MINUTE); + + // Hard cap on field-level iterations. Each loop body either advances + // the date by at least one minute, or by a larger granularity (day / + // month) when a coarser field rejects the current point. We count + // worst-case minute advances against `capMinutes` and bail out if a + // legal-but-never-fires expression exhausts the budget. + let iterations = 0; + const maxIterations = capMinutes + 10_000; + + while (iterations++ < maxIterations) { + // Month — coarsest. If wrong, jump to day 1 of the next allowed + // month and restart the day check. + if (!expr.months.has(date.getMonth() + 1)) { + advanceMonth(date); + continue; + } + + // Day. Cron-style OR: when both dom and dow are restricted, match + // either; when one is `*`, only the other constrains. + if (!dayMatches(expr, date)) { + advanceDay(date); + continue; + } + + if (!expr.hours.has(date.getHours())) { + advanceHour(date); + continue; + } + + if (!expr.minutes.has(date.getMinutes())) { + advanceMinute(date); + continue; + } + + return date.getTime(); + } + + return null; +} + +function dayMatches(expr: ParsedCronExpression, date: Date): boolean { + const dom = date.getDate(); + const dow = date.getDay(); + const domOk = expr.daysOfMonth.has(dom); + const dowOk = expr.daysOfWeek.has(dow); + + if (expr.daysOfMonthWildcard && expr.daysOfWeekWildcard) return true; + if (expr.daysOfMonthWildcard) return dowOk; + if (expr.daysOfWeekWildcard) return domOk; + // Both restricted: cron-style OR. + return domOk || dowOk; +} + +function advanceMonth(date: Date): void { + // Jump to the 1st of the next month at 00:00. Date's wrap-around + // handles year rollover for us. + date.setDate(1); + date.setHours(0, 0, 0, 0); + date.setMonth(date.getMonth() + 1); +} + +function advanceDay(date: Date): void { + date.setHours(0, 0, 0, 0); + date.setDate(date.getDate() + 1); +} + +function advanceHour(date: Date): void { + date.setMinutes(0, 0, 0); + date.setHours(date.getHours() + 1); +} + +function advanceMinute(date: Date): void { + date.setSeconds(0, 0); + date.setMinutes(date.getMinutes() + 1); +} + +const MONTH_NAMES = [ + 'January', + 'February', + 'March', + 'April', + 'May', + 'June', + 'July', + 'August', + 'September', + 'October', + 'November', + 'December', +] as const; + +const DAY_NAMES = [ + 'Sunday', + 'Monday', + 'Tuesday', + 'Wednesday', + 'Thursday', + 'Friday', + 'Saturday', +] as const; + +/** + * Cheap human-readable summary of an expression. Falls back to the raw + * string when the shape isn't one of the patterns we recognise — the + * caller (CronList) uses this purely for display, so a wordy fallback + * is fine and we don't try to be exhaustive. + */ +export function cronToHuman(expr: ParsedCronExpression): string { + const allMin = isFullRange(expr.minutes, 0, 59); + const allHour = isFullRange(expr.hours, 0, 23); + const allDom = expr.daysOfMonthWildcard; + const allMonth = isFullRange(expr.months, 1, 12); + const allDow = expr.daysOfWeekWildcard; + + // every N minutes — common LLM pattern (`*/5 * * * *`). + if (allHour && allDom && allMonth && allDow) { + const step = detectStep(expr.minutes, 0, 59); + if (step !== null && step > 1) return `every ${step} minutes`; + if (allMin) return 'every minute'; + if (expr.minutes.size === 1) { + const m = [...expr.minutes][0]!; + return `at minute ${m} of every hour`; + } + } + + // every N hours. + if (expr.minutes.size === 1 && allDom && allMonth && allDow) { + const m = [...expr.minutes][0]!; + const step = detectStep(expr.hours, 0, 23); + if (step !== null && step > 1) { + return `every ${step} hours at minute ${pad(m)}`; + } + } + + // at HH:MM every day, optional dow restriction. + if ( + expr.minutes.size === 1 && + expr.hours.size === 1 && + allDom && + allMonth + ) { + const h = [...expr.hours][0]!; + const m = [...expr.minutes][0]!; + if (allDow) return `at ${pad(h)}:${pad(m)} every day`; + const dowStr = formatDows(expr.daysOfWeek); + if (dowStr !== null) return `at ${pad(h)}:${pad(m)} on ${dowStr}`; + } + + // at HH:MM on day N of . + if ( + expr.minutes.size === 1 && + expr.hours.size === 1 && + expr.daysOfMonth.size === 1 && + !expr.daysOfMonthWildcard && + expr.months.size === 1 && + allDow + ) { + const h = [...expr.hours][0]!; + const m = [...expr.minutes][0]!; + const d = [...expr.daysOfMonth][0]!; + const mo = [...expr.months][0]!; + return `at ${pad(h)}:${pad(m)} on day ${d} of ${MONTH_NAMES[mo - 1]}`; + } + + return expr.raw; +} + +function isFullRange(set: ReadonlySet, min: number, max: number): boolean { + if (set.size !== max - min + 1) return false; + for (let v = min; v <= max; v++) if (!set.has(v)) return false; + return true; +} + +/** + * If the set looks like `{min, min+step, ..., <=max}` with a constant + * step, return `step`. Otherwise null. Used to pretty-print star-slash-N. + */ +function detectStep(set: ReadonlySet, min: number, max: number): number | null { + const values = [...set].toSorted((a, b) => a - b); + if (values.length < 2) return null; + if (values[0] !== min) return null; + const step = values[1] - values[0]; + if (step <= 0) return null; + let expected = min; + for (const v of values) { + if (v !== expected) return null; + expected += step; + } + // The last expected value should exceed `max` by less than `step`. + if (expected - step > max) return null; + return step; +} + +function formatDows(set: ReadonlySet): string | null { + const values = [...set].toSorted((a, b) => a - b); + if (values.length === 0) return null; + // Mon-Fri shortcut. + if (values.length === 5 && values.every((v, i) => v === i + 1)) { + return 'weekdays'; + } + if (values.length === 2 && values[0] === 0 && values[1] === 6) { + return 'weekends'; + } + return values.map((v) => DAY_NAMES[v]!).join(', '); +} + +function pad(n: number): string { + return n < 10 ? `0${n}` : String(n); +} diff --git a/packages/agent-core/test/tools/cron/cron-expr.test.ts b/packages/agent-core/test/tools/cron/cron-expr.test.ts new file mode 100644 index 0000000000..8f42aa8f7c --- /dev/null +++ b/packages/agent-core/test/tools/cron/cron-expr.test.ts @@ -0,0 +1,325 @@ +/** + * Tests for `tools/cron/cron-expr.ts`. Mirrors the layout of + * `clock.test.ts` (lives under `test/` because the package's vitest + * config scopes `include` to `test/**`). + * + * Local-time semantics matter: `new Date('2024-06-01T12:00:30')` in JS + * land is interpreted as local time when the string has no timezone + * suffix (and as UTC when it has one). We construct dates explicitly + * via `new Date(year, monthIndex, day, hour, minute, second)` so the + * tests are stable regardless of the test process's TZ. + */ +import { describe, expect, it } from 'vitest'; + +import { + computeNextCronRun, + cronToHuman, + hasFireWithinYears, + parseCronExpression, +} from '../../../src/tools/cron/cron-expr'; + +function localDate( + y: number, + monthIndex: number, + d: number, + h = 0, + m = 0, + s = 0, +): number { + return new Date(y, monthIndex, d, h, m, s, 0).getTime(); +} + +function nextLocalParts(ts: number) { + const d = new Date(ts); + return { + year: d.getFullYear(), + month: d.getMonth() + 1, + day: d.getDate(), + hour: d.getHours(), + minute: d.getMinutes(), + second: d.getSeconds(), + dow: d.getDay(), + }; +} + +describe('parseCronExpression', () => { + it('parses wildcard', () => { + const p = parseCronExpression('* * * * *'); + expect(p.minutes.size).toBe(60); + expect(p.hours.size).toBe(24); + expect(p.daysOfMonth.size).toBe(31); + expect(p.months.size).toBe(12); + expect(p.daysOfWeek.size).toBe(7); + expect(p.daysOfMonthWildcard).toBe(true); + expect(p.daysOfWeekWildcard).toBe(true); + }); + + it('parses single integers', () => { + const p = parseCronExpression('5 9 1 6 3'); + expect([...p.minutes]).toEqual([5]); + expect([...p.hours]).toEqual([9]); + expect([...p.daysOfMonth]).toEqual([1]); + expect([...p.months]).toEqual([6]); + expect([...p.daysOfWeek]).toEqual([3]); + expect(p.daysOfMonthWildcard).toBe(false); + expect(p.daysOfWeekWildcard).toBe(false); + }); + + it('parses ranges', () => { + const p = parseCronExpression('0 9-17 * * 1-5'); + expect([...p.hours].toSorted((a, b) => a - b)).toEqual([9, 10, 11, 12, 13, 14, 15, 16, 17]); + expect([...p.daysOfWeek].toSorted((a, b) => a - b)).toEqual([1, 2, 3, 4, 5]); + }); + + it('parses lists', () => { + const p = parseCronExpression('0 9,12,17 * * *'); + expect([...p.hours].toSorted((a, b) => a - b)).toEqual([9, 12, 17]); + }); + + it('parses step with wildcard', () => { + const p = parseCronExpression('*/5 * * * *'); + expect([...p.minutes].toSorted((a, b) => a - b)).toEqual([ + 0, 5, 10, 15, 20, 25, 30, 35, 40, 45, 50, 55, + ]); + }); + + it('parses step with range', () => { + const p = parseCronExpression('0-30/10 * * * *'); + expect([...p.minutes].toSorted((a, b) => a - b)).toEqual([0, 10, 20, 30]); + }); + + it('folds day-of-week 7 to 0 (Sunday)', () => { + const p = parseCronExpression('0 0 * * 7'); + expect([...p.daysOfWeek]).toEqual([0]); + }); + + it('treats bare * as wildcard but */n as restriction', () => { + const a = parseCronExpression('* * * * *'); + const b = parseCronExpression('*/5 * * * *'); + expect(a.daysOfMonthWildcard).toBe(true); + expect(b.daysOfMonthWildcard).toBe(true); + }); + + it('throws on too few fields', () => { + expect(() => parseCronExpression('* * * *')).toThrow(/5 fields/); + }); + + it('throws on too many fields', () => { + expect(() => parseCronExpression('* * * * * *')).toThrow(/5 fields/); + }); + + it('throws on empty input', () => { + expect(() => parseCronExpression('')).toThrow(/empty/); + expect(() => parseCronExpression(' ')).toThrow(/empty/); + }); + + it('throws on out-of-range minute', () => { + expect(() => parseCronExpression('60 * * * *')).toThrow(/minute/); + }); + + it('throws on out-of-range hour', () => { + expect(() => parseCronExpression('0 24 * * *')).toThrow(/hour/); + }); + + it('throws on out-of-range day-of-month', () => { + expect(() => parseCronExpression('0 0 32 * *')).toThrow(/day-of-month/); + }); + + it('throws on out-of-range month', () => { + expect(() => parseCronExpression('0 0 * 13 *')).toThrow(/month/); + }); + + it('throws on out-of-range day-of-week', () => { + expect(() => parseCronExpression('0 0 * * 8')).toThrow(/day-of-week/); + }); + + it('throws on non-integer step', () => { + expect(() => parseCronExpression('*/x * * * *')).toThrow(/step/); + }); + + it('throws on zero step', () => { + expect(() => parseCronExpression('*/0 * * * *')).toThrow(/step/); + }); + + it('throws on descending range', () => { + expect(() => parseCronExpression('5-1 * * * *')).toThrow(/range/); + }); + + it('throws on empty list term', () => { + expect(() => parseCronExpression('1,,3 * * * *')).toThrow(/empty term/); + }); +}); + +describe('computeNextCronRun', () => { + it('*/5 — from xx:00:30 advances to xx:05:00', () => { + const expr = parseCronExpression('*/5 * * * *'); + const from = localDate(2024, 5, 1, 12, 0, 30); + const next = computeNextCronRun(expr, from); + expect(next).not.toBeNull(); + const p = nextLocalParts(next!); + expect(p.year).toBe(2024); + expect(p.month).toBe(6); + expect(p.day).toBe(1); + expect(p.hour).toBe(12); + expect(p.minute).toBe(5); + expect(p.second).toBe(0); + }); + + it('*/5 — from xx:00:00 advances strictly to xx:05:00, never xx:00:00', () => { + const expr = parseCronExpression('*/5 * * * *'); + const from = localDate(2024, 5, 1, 12, 0, 0); + const next = computeNextCronRun(expr, from); + expect(next).not.toBeNull(); + expect(next!).toBeGreaterThan(from); + const p = nextLocalParts(next!); + expect(p.minute).toBe(5); + }); + + it('0 9 * * * — from 08:00 advances to 09:00 same day', () => { + const expr = parseCronExpression('0 9 * * *'); + const from = localDate(2024, 5, 1, 8, 0, 0); + const next = computeNextCronRun(expr, from); + const p = nextLocalParts(next!); + expect(p.day).toBe(1); + expect(p.hour).toBe(9); + expect(p.minute).toBe(0); + }); + + it('0 9 * * 1-5 — from Saturday 09:00 → next Monday 09:00', () => { + const expr = parseCronExpression('0 9 * * 1-5'); + // 2024-06-01 is a Saturday. + const sat = new Date(2024, 5, 1, 9, 0, 0, 0); + expect(sat.getDay()).toBe(6); + const next = computeNextCronRun(expr, sat.getTime()); + const p = nextLocalParts(next!); + expect(p.dow).toBe(1); // Monday + expect(p.day).toBe(3); // 2024-06-03 + expect(p.hour).toBe(9); + expect(p.minute).toBe(0); + }); + + it('0 12 1 1 * — from mid-year advances to next Jan 1 12:00', () => { + const expr = parseCronExpression('0 12 1 1 *'); + const from = localDate(2024, 5, 1, 0, 0, 0); + const next = computeNextCronRun(expr, from); + const p = nextLocalParts(next!); + expect(p.year).toBe(2025); + expect(p.month).toBe(1); + expect(p.day).toBe(1); + expect(p.hour).toBe(12); + }); + + it('0 0 31 2 * — Feb has no day 31 → null', () => { + const expr = parseCronExpression('0 0 31 2 *'); + const from = localDate(2024, 0, 1, 0, 0, 0); + expect(computeNextCronRun(expr, from)).toBeNull(); + }); + + it('29 2 — Feb 29 fires on leap years', () => { + // `0 0 29 2 *` — every Feb 29 midnight. + const expr = parseCronExpression('0 0 29 2 *'); + const from = localDate(2023, 0, 1, 0, 0, 0); + const next = computeNextCronRun(expr, from); + const p = nextLocalParts(next!); + expect(p.year).toBe(2024); + expect(p.month).toBe(2); + expect(p.day).toBe(29); + }); + + it('cron-style OR: 0 0 1 * 1 fires on every 1st AND every Monday', () => { + const expr = parseCronExpression('0 0 1 * 1'); + // Sample a few fires across a couple months. Starting Jun 1 2024 (Sat). + let cur = localDate(2024, 5, 1, 0, 0, 0) - 1; + const fires: { date: Date; dow: number; dom: number }[] = []; + for (let i = 0; i < 12; i++) { + const n = computeNextCronRun(expr, cur); + expect(n).not.toBeNull(); + const d = new Date(n!); + fires.push({ date: d, dow: d.getDay(), dom: d.getDate() }); + cur = n!; + } + // Every fire must be either a Monday OR the 1st of a month. + for (const f of fires) { + const isMonday = f.dow === 1; + const isFirst = f.dom === 1; + expect(isMonday || isFirst).toBe(true); + } + // We must see at least one of each (Monday and non-Monday 1st). + expect(fires.some((f) => f.dow === 1 && f.dom !== 1)).toBe(true); + expect(fires.some((f) => f.dom === 1)).toBe(true); + }); + + it('returns strictly greater than fromMs', () => { + const expr = parseCronExpression('* * * * *'); + const from = localDate(2024, 5, 1, 12, 30, 45); + const next = computeNextCronRun(expr, from); + expect(next).not.toBeNull(); + expect(next!).toBeGreaterThan(from); + }); + + it('DST transition still advances and never returns < fromMs', () => { + // Pick a date around the US DST transition (2024-03-10 spring + // forward in America/New_York). The test process TZ may not be + // NY, but the invariant "monotonic forward" must hold in any TZ. + const expr = parseCronExpression('0 * * * *'); + const from = localDate(2024, 2, 10, 0, 0, 0); + let cur = from; + let prev = from; + for (let i = 0; i < 48; i++) { + const n = computeNextCronRun(expr, cur); + expect(n).not.toBeNull(); + expect(n!).toBeGreaterThan(prev); + prev = cur; + cur = n!; + } + }); +}); + +describe('hasFireWithinYears', () => { + it('0 0 31 2 * → false within 5 years', () => { + const expr = parseCronExpression('0 0 31 2 *'); + expect(hasFireWithinYears(expr, 5, localDate(2024, 0, 1))).toBe(false); + }); + + it('0 12 1 1 * → true within 5 years', () => { + const expr = parseCronExpression('0 12 1 1 *'); + expect(hasFireWithinYears(expr, 5, localDate(2024, 0, 1))).toBe(true); + }); + + it('* * * * * → true within any nonzero window', () => { + const expr = parseCronExpression('* * * * *'); + expect(hasFireWithinYears(expr, 1, localDate(2024, 0, 1))).toBe(true); + }); +}); + +describe('cronToHuman', () => { + it('every minute', () => { + expect(cronToHuman(parseCronExpression('* * * * *'))).toBe('every minute'); + }); + + it('every 5 minutes', () => { + expect(cronToHuman(parseCronExpression('*/5 * * * *'))).toBe('every 5 minutes'); + }); + + it('at HH:MM every day', () => { + expect(cronToHuman(parseCronExpression('0 9 * * *'))).toBe('at 09:00 every day'); + expect(cronToHuman(parseCronExpression('30 14 * * *'))).toBe('at 14:30 every day'); + }); + + it('weekdays / weekends shortcut', () => { + expect(cronToHuman(parseCronExpression('0 9 * * 1-5'))).toBe('at 09:00 on weekdays'); + expect(cronToHuman(parseCronExpression('0 10 * * 0,6'))).toBe('at 10:00 on weekends'); + }); + + it('at HH:MM on day N of ', () => { + expect(cronToHuman(parseCronExpression('0 12 1 1 *'))).toBe('at 12:00 on day 1 of January'); + }); + + it('every N hours', () => { + expect(cronToHuman(parseCronExpression('0 */6 * * *'))).toBe('every 6 hours at minute 00'); + }); + + it('falls back to raw expression for weird patterns', () => { + expect(cronToHuman(parseCronExpression('1,7,23 5,17 * * *'))).toBe('1,7,23 5,17 * * *'); + }); +}); From 7507ac755232e43c1ff1feab6b56d473df95e112 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 27 May 2026 19:16:12 +0800 Subject: [PATCH 03/26] feat(agent-core): add deterministic cron jitter Recurring jobs shift forward by min(10% of period, 15min); one-shot jobs landing on :00/:30 shift earlier by up to 90s. Offset is hashed from task.id so reschedules and restarts stay stable. KIMI_CRON_NO_JITTER disables both branches for reproducible benches. --- packages/agent-core/src/tools/cron/jitter.ts | 147 +++++++++++ .../agent-core/test/tools/cron/jitter.test.ts | 238 ++++++++++++++++++ 2 files changed, 385 insertions(+) create mode 100644 packages/agent-core/src/tools/cron/jitter.ts create mode 100644 packages/agent-core/test/tools/cron/jitter.test.ts diff --git a/packages/agent-core/src/tools/cron/jitter.ts b/packages/agent-core/src/tools/cron/jitter.ts new file mode 100644 index 0000000000..2c03c456ad --- /dev/null +++ b/packages/agent-core/src/tools/cron/jitter.ts @@ -0,0 +1,147 @@ +/** + * Per-task deterministic jitter for cron fire times. + * + * Why this exists: if every user writes `0 9 * * *` ("every day at 9 + * am") then every CLI fires at the same instant and the upstream API + * sees a thundering herd at :00. We soften that by shifting each + * task's ideal fire time by a small, **deterministic** per-task + * offset so a given task always lands at the same jittered point — + * reschedules and restarts don't drift, and bench reproducibility + * stays intact when {@link KIMI_CRON_NO_JITTER} is set. + * + * Two flavours: + * + * - **Recurring**: shift *forward* by a fraction of the period + * (cap 10% of period, hard cap 15 min). Long-period jobs (`0 9 * + * * *`, period 1 day) hit the 15-minute cap; short-period jobs + * (`*` /5 * * * *`, period 5 min) are bounded by the 10% rule. + * + * - **One-shot**: shift *earlier* (negative), but only when the + * ideal lands on `:00` or `:30` — that's the signal the model + * picked a round number with no specific intent. Cap 90 s + * earlier. Any other minute (`:07`, `:23`, …) passes through + * unchanged because the model presumably meant that exact time. + * + * The function is pure given its inputs — no module-level cache; the + * hash is recomputed from `task.id` each call. That trades a handful + * of cheap arithmetic ops for a guarantee that there is no hidden + * state to invalidate when a task is rescheduled. + */ +import type { ParsedCronExpression } from './cron-expr'; +import { computeNextCronRun } from './cron-expr'; + +/** Tunables for {@link jitteredNextCronRunMs} / {@link oneShotJitteredNextCronRunMs}. */ +export interface JitterConfig { + /** Recurring offset cap as a fraction of the cron period (0..1). */ + readonly recurringMaxFractionOfPeriod: number; + /** Absolute cap on the recurring offset, in ms. */ + readonly recurringMaxMs: number; + /** Absolute cap on the one-shot pull-forward, in ms. */ + readonly oneShotMaxMs: number; +} + +export const DEFAULT_CRON_JITTER_CONFIG: JitterConfig = { + recurringMaxFractionOfPeriod: 0.1, + recurringMaxMs: 15 * 60_000, + oneShotMaxMs: 90_000, +}; + +const MS_PER_DAY = 24 * 60 * 60_000; +const MS_PER_MINUTE = 60_000; + +/** + * Map a task id to a deterministic fraction in `[0, 1)`. Cron task + * ids are 8 hex chars (`/^[0-9a-f]{8}$/`), so `parseInt(id, 16)` / + * `2^32` lands neatly in range. For non-hex inputs we fall back to a + * djb2-style reduction so callers passing test fixtures with + * arbitrary string ids still get a stable spread. + */ +function fractionFromId(id: string): number { + if (/^[0-9a-f]{8}$/i.test(id)) { + const n = Number.parseInt(id, 16); + if (Number.isFinite(n)) { + // 2^32 keeps the result strictly < 1. + return n / 0x1_0000_0000; + } + } + // djb2 reduction — overflow-safe in JS (operates on int32) and + // good enough spread for non-hex test ids. + let hash = 5381; + for (let i = 0; i < id.length; i++) { + hash = ((hash << 5) + hash + id.charCodeAt(i)) | 0; + } + // Map signed int32 to [0, 1). + const unsigned = hash >>> 0; + return unsigned / 0x1_0000_0000; +} + +function jitterDisabledByEnv(): boolean { + return process.env.KIMI_CRON_NO_JITTER === '1'; +} + +/** + * Apply recurring-job jitter to an already-computed ideal fire time. + * + * The shift is **forward only** (≥ 0), bounded by both the relative + * fraction-of-period cap and the absolute ms cap. We discover the + * period by asking {@link computeNextCronRun} for the run *after* + * `idealMs`; if that returns `null` (legal-but-never-fires + * expression — should have been rejected upstream) we fall back to a + * 24-hour assumption so we still produce some sensible offset rather + * than spiking on the original `idealMs`. + */ +export function jitteredNextCronRunMs( + task: { id: string; cron: string; recurring?: boolean }, + parsed: ParsedCronExpression, + idealMs: number, + config: JitterConfig = DEFAULT_CRON_JITTER_CONFIG, +): number { + if (jitterDisabledByEnv()) { + return idealMs; + } + const nextNext = computeNextCronRun(parsed, idealMs); + const period = + nextNext !== null && nextNext > idealMs ? nextNext - idealMs : MS_PER_DAY; + const periodCap = period * config.recurringMaxFractionOfPeriod; + const cap = Math.min(periodCap, config.recurringMaxMs); + if (!(cap > 0)) { + return idealMs; + } + const offset = cap * fractionFromId(task.id); + return idealMs + offset; +} + +/** + * Apply one-shot pull-forward jitter to an ideal fire time. + * + * Only fires on `:00` and `:30` of the hour — the minute marks the + * model is most likely to pick out of habit. Other minutes pass + * through verbatim so a user who said "remind me at 2:07" gets + * 2:07 exactly. The shift is in `[-oneShotMaxMs, 0)`; never exactly + * 0 unless the deterministic hash happens to land on 0 (which is + * fine — it just means this task is the unlucky one that pays the + * full delay). + */ +export function oneShotJitteredNextCronRunMs( + task: { id: string }, + idealMs: number, + config: JitterConfig = DEFAULT_CRON_JITTER_CONFIG, +): number { + if (jitterDisabledByEnv()) { + return idealMs; + } + // Only the minute-of-hour matters: did the model land on the round + // number? Sub-minute granularity is filtered by the modulo check. + if (idealMs % MS_PER_MINUTE !== 0) { + return idealMs; + } + const minuteOfHour = new Date(idealMs).getMinutes(); + if (minuteOfHour !== 0 && minuteOfHour !== 30) { + return idealMs; + } + if (!(config.oneShotMaxMs > 0)) { + return idealMs; + } + const offset = -config.oneShotMaxMs * fractionFromId(task.id); + return idealMs + offset; +} diff --git a/packages/agent-core/test/tools/cron/jitter.test.ts b/packages/agent-core/test/tools/cron/jitter.test.ts new file mode 100644 index 0000000000..f122c142bf --- /dev/null +++ b/packages/agent-core/test/tools/cron/jitter.test.ts @@ -0,0 +1,238 @@ +/** + * Tests for `tools/cron/jitter.ts`. Mirrors layout of `cron-expr.test.ts` — + * lives under `test/` because vitest config scopes include to `test/**`. + * + * We construct fire times via `new Date(year, monthIndex, day, h, m, s)` + * so the tests are stable regardless of the host TZ. The jitter + * implementation reasons about *minute-of-hour* (a local-time concept), + * which is what we want for "ignore round-number :00/:30 marks". + */ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import { parseCronExpression } from '../../../src/tools/cron/cron-expr'; +import { + DEFAULT_CRON_JITTER_CONFIG, + jitteredNextCronRunMs, + oneShotJitteredNextCronRunMs, +} from '../../../src/tools/cron/jitter'; + +function localDate( + y: number, + monthIndex: number, + d: number, + h = 0, + m = 0, + s = 0, +): number { + return new Date(y, monthIndex, d, h, m, s, 0).getTime(); +} + +/** + * Two distinct 8-hex ids that produce visibly different jitter + * fractions. `aaaaaaaa` ≈ 0.667, `11111111` ≈ 0.067, so any cap > a + * few ms yields a separable offset for the "two distinct ids" test. + */ +const ID_A = 'aaaaaaaa'; +const ID_B = '11111111'; + +describe('jitteredNextCronRunMs — recurring', () => { + it('offset for */5 * * * * is within [0, 30s] (10% of 5min period)', () => { + const parsed = parseCronExpression('*/5 * * * *'); + const ideal = localDate(2024, 5, 1, 12, 5, 0); // 12:05 local + const jittered = jitteredNextCronRunMs( + { id: ID_A, cron: '*/5 * * * *', recurring: true }, + parsed, + ideal, + ); + expect(jittered).toBeGreaterThanOrEqual(ideal); + expect(jittered - ideal).toBeLessThanOrEqual(30_000); + }); + + it('offset for daily 0 9 * * * is capped at 15min (NOT 10% of 1 day)', () => { + const parsed = parseCronExpression('0 9 * * *'); + const ideal = localDate(2024, 5, 1, 9, 0, 0); + const jittered = jitteredNextCronRunMs( + { id: ID_A, cron: '0 9 * * *', recurring: true }, + parsed, + ideal, + ); + expect(jittered).toBeGreaterThanOrEqual(ideal); + // 10% of 1 day = 2.4h = 8 640 000 ms, but cap is 15 min. + expect(jittered - ideal).toBeLessThanOrEqual(15 * 60_000); + // And visibly bigger than the */5 case — long-period jobs spread + // further (good for anti-herd at :00). + expect(jittered - ideal).toBeGreaterThan(60_000); + }); + + it('different ids produce different offsets', () => { + const parsed = parseCronExpression('*/5 * * * *'); + const ideal = localDate(2024, 5, 1, 12, 5, 0); + const a = jitteredNextCronRunMs( + { id: ID_A, cron: '*/5 * * * *', recurring: true }, + parsed, + ideal, + ); + const b = jitteredNextCronRunMs( + { id: ID_B, cron: '*/5 * * * *', recurring: true }, + parsed, + ideal, + ); + expect(a).not.toBe(b); + }); + + it('deterministic: same inputs → same output across calls', () => { + const parsed = parseCronExpression('0 9 * * *'); + const ideal = localDate(2024, 5, 1, 9, 0, 0); + const calls = Array.from({ length: 5 }, () => + jitteredNextCronRunMs( + { id: ID_A, cron: '0 9 * * *', recurring: true }, + parsed, + ideal, + ), + ); + for (const v of calls) { + expect(v).toBe(calls[0]); + } + }); + + it('respects KIMI_CRON_NO_JITTER=1 — no offset', () => { + const prev = process.env.KIMI_CRON_NO_JITTER; + process.env.KIMI_CRON_NO_JITTER = '1'; + try { + const parsed = parseCronExpression('*/5 * * * *'); + const ideal = localDate(2024, 5, 1, 12, 5, 0); + const jittered = jitteredNextCronRunMs( + { id: ID_A, cron: '*/5 * * * *', recurring: true }, + parsed, + ideal, + ); + expect(jittered).toBe(ideal); + } finally { + if (prev === undefined) delete process.env.KIMI_CRON_NO_JITTER; + else process.env.KIMI_CRON_NO_JITTER = prev; + } + }); +}); + +describe('oneShotJitteredNextCronRunMs', () => { + it(':00 → offset within [-90s, 0)', () => { + const ideal = localDate(2024, 5, 1, 14, 0, 0); + const jittered = oneShotJitteredNextCronRunMs({ id: ID_A }, ideal); + expect(jittered - ideal).toBeLessThanOrEqual(0); + expect(jittered - ideal).toBeGreaterThanOrEqual(-90_000); + // ID_A's fraction ≈ 0.667 → expect ~ -60s, not 0. + expect(jittered).toBeLessThan(ideal); + }); + + it(':30 → offset within [-90s, 0)', () => { + const ideal = localDate(2024, 5, 1, 14, 30, 0); + const jittered = oneShotJitteredNextCronRunMs({ id: ID_A }, ideal); + expect(jittered - ideal).toBeLessThanOrEqual(0); + expect(jittered - ideal).toBeGreaterThanOrEqual(-90_000); + expect(jittered).toBeLessThan(ideal); + }); + + it(':07 → passthrough, offset = 0', () => { + const ideal = localDate(2024, 5, 1, 14, 7, 0); + const jittered = oneShotJitteredNextCronRunMs({ id: ID_A }, ideal); + expect(jittered).toBe(ideal); + }); + + it(':15 → passthrough, offset = 0', () => { + const ideal = localDate(2024, 5, 1, 14, 15, 0); + const jittered = oneShotJitteredNextCronRunMs({ id: ID_A }, ideal); + expect(jittered).toBe(ideal); + }); + + it('mid-minute (non-zero seconds component baked into idealMs) → passthrough', () => { + // Sub-minute granularity means the model didn't pick a round + // wall-clock minute — leave it alone. + const ideal = localDate(2024, 5, 1, 14, 0, 12); + const jittered = oneShotJitteredNextCronRunMs({ id: ID_A }, ideal); + expect(jittered).toBe(ideal); + }); + + it('deterministic: same id + same ideal → same output', () => { + const ideal = localDate(2024, 5, 1, 14, 0, 0); + const calls = Array.from({ length: 5 }, () => + oneShotJitteredNextCronRunMs({ id: ID_A }, ideal), + ); + for (const v of calls) { + expect(v).toBe(calls[0]); + } + }); + + it('respects KIMI_CRON_NO_JITTER=1', () => { + const prev = process.env.KIMI_CRON_NO_JITTER; + process.env.KIMI_CRON_NO_JITTER = '1'; + try { + const ideal = localDate(2024, 5, 1, 14, 0, 0); + const jittered = oneShotJitteredNextCronRunMs({ id: ID_A }, ideal); + expect(jittered).toBe(ideal); + } finally { + if (prev === undefined) delete process.env.KIMI_CRON_NO_JITTER; + else process.env.KIMI_CRON_NO_JITTER = prev; + } + }); +}); + +describe('config knobs', () => { + it('default config is the documented constants', () => { + expect(DEFAULT_CRON_JITTER_CONFIG.recurringMaxFractionOfPeriod).toBe(0.1); + expect(DEFAULT_CRON_JITTER_CONFIG.recurringMaxMs).toBe(15 * 60_000); + expect(DEFAULT_CRON_JITTER_CONFIG.oneShotMaxMs).toBe(90_000); + }); + + it('custom oneShotMaxMs caps the pull-forward', () => { + const ideal = localDate(2024, 5, 1, 14, 0, 0); + const jittered = oneShotJitteredNextCronRunMs( + { id: ID_A }, + ideal, + { ...DEFAULT_CRON_JITTER_CONFIG, oneShotMaxMs: 10_000 }, + ); + expect(jittered - ideal).toBeGreaterThanOrEqual(-10_000); + expect(jittered - ideal).toBeLessThanOrEqual(0); + }); + + it('custom recurringMaxMs caps the forward shift', () => { + const parsed = parseCronExpression('0 9 * * *'); + const ideal = localDate(2024, 5, 1, 9, 0, 0); + const jittered = jitteredNextCronRunMs( + { id: ID_A, cron: '0 9 * * *', recurring: true }, + parsed, + ideal, + { ...DEFAULT_CRON_JITTER_CONFIG, recurringMaxMs: 5_000 }, + ); + expect(jittered - ideal).toBeGreaterThanOrEqual(0); + expect(jittered - ideal).toBeLessThanOrEqual(5_000); + }); +}); + +describe('id hashing fallback', () => { + it('non-hex id still produces a stable fraction', () => { + const parsed = parseCronExpression('*/5 * * * *'); + const ideal = localDate(2024, 5, 1, 12, 5, 0); + const a1 = jitteredNextCronRunMs( + { id: 'non-hex-id!', cron: '*/5 * * * *', recurring: true }, + parsed, + ideal, + ); + const a2 = jitteredNextCronRunMs( + { id: 'non-hex-id!', cron: '*/5 * * * *', recurring: true }, + parsed, + ideal, + ); + expect(a1).toBe(a2); + expect(a1).toBeGreaterThanOrEqual(ideal); + expect(a1 - ideal).toBeLessThanOrEqual(30_000); + }); +}); + +// Belt-and-suspenders: ensure env state we touched is clean after the +// suite (in case a future test in the same vitest worker reads it). +beforeEach(() => { + delete process.env.KIMI_CRON_NO_JITTER; +}); +afterEach(() => { + delete process.env.KIMI_CRON_NO_JITTER; +}); From 27ddbc8cf47f02ac345a7954423e84ad72292087 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 27 May 2026 19:17:57 +0800 Subject: [PATCH 04/26] feat(agent-core): add CronTask type and cron prompt origins CronTask matches what gets persisted to tasks.json (with durable stripped). CronJobOrigin carries coalescedCount and stale so the agent can react to collapsed fires without separate channels; CronMissedOrigin tags the boot-time AskUserQuestion path. --- .../agent-core/src/agent/context/types.ts | 19 +++++++++++++ packages/agent-core/src/tools/cron/types.ts | 28 +++++++++++++++++++ 2 files changed, 47 insertions(+) create mode 100644 packages/agent-core/src/tools/cron/types.ts diff --git a/packages/agent-core/src/agent/context/types.ts b/packages/agent-core/src/agent/context/types.ts index b8d0770096..a366fce350 100644 --- a/packages/agent-core/src/agent/context/types.ts +++ b/packages/agent-core/src/agent/context/types.ts @@ -42,6 +42,23 @@ export interface BackgroundTaskOrigin { readonly notificationId: string; } +export interface CronJobOrigin { + readonly kind: 'cron_job'; + readonly jobId: string; + readonly cron: string; + readonly recurring: boolean; + /** Number of theoretical fires that were collapsed into this single delivery (>= 1). */ + readonly coalescedCount: number; + /** True for recurring tasks past the 7-day age threshold. */ + readonly stale: boolean; +} + +export interface CronMissedOrigin { + readonly kind: 'cron_missed'; + /** Number of one-shot tasks bundled into this missed-fire notification. */ + readonly count: number; +} + export interface HookResultOrigin { readonly kind: 'hook_result'; readonly event: string; @@ -55,6 +72,8 @@ export type PromptOrigin = | CompactionSummaryOrigin | SystemTriggerOrigin | BackgroundTaskOrigin + | CronJobOrigin + | CronMissedOrigin | HookResultOrigin; export type ContextMessage = Message & { diff --git a/packages/agent-core/src/tools/cron/types.ts b/packages/agent-core/src/tools/cron/types.ts new file mode 100644 index 0000000000..68b0efedd3 --- /dev/null +++ b/packages/agent-core/src/tools/cron/types.ts @@ -0,0 +1,28 @@ +/** + * Persistent representation of a cron task. + * + * - `id` — 8-hex; jitter is keyed off this hash, so stable id == stable + * jitter across schedule rewrites. + * - `cron` — 5-field expression, evaluated in local time. + * - `createdAt` — wall-clock epoch ms at original scheduling. NOT updated + * when the scheduler fires; recurring uses `max(createdAt, now)` to + * re-derive `nextFireAt` after restart. `createdAt` is also the input + * to the 7-day stale judgment. + * - `recurring` — undefined / true means "fire repeatedly until deleted + * or auto-expired"; false means "fire once then auto-delete". + * - `durable` — runtime-only field that decides where the task lives + * (file vs session-store). It is stripped before writing to tasks.json. + * + * Notably absent: `lastFiredAt`. Persisting last-fire would let a fast + * restart skip a legitimately-due fire because the stored timestamp says + * "fired recently". Coalesce semantics make missing one fire across a + * restart acceptable; missing one fire because of bad bookkeeping is not. + */ +export interface CronTask { + readonly id: string; + readonly cron: string; + readonly prompt: string; + readonly createdAt: number; + readonly recurring?: boolean; + readonly durable?: boolean; +} From aed8d24400a99ad1300f3839e17d9e2bb1fdc0b0 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 27 May 2026 19:22:48 +0800 Subject: [PATCH 05/26] test(agent-core): guard against Date.now() in cron scheduler files oxlint 1.59 does not support no-restricted-syntax, so the ESLint-style guard from the plan is implemented as a vitest scan. The four guarded files (scheduler/persist/lock/jitter) must route every wall-clock read through ClockSources.wallNow(); clock.ts is excluded because that is where the abstraction is defined. Non-existent files are skipped so the guard activates automatically when later commits introduce them. --- .../test/tools/cron/no-date-now.test.ts | 78 +++++++++++++++++++ 1 file changed, 78 insertions(+) create mode 100644 packages/agent-core/test/tools/cron/no-date-now.test.ts diff --git a/packages/agent-core/test/tools/cron/no-date-now.test.ts b/packages/agent-core/test/tools/cron/no-date-now.test.ts new file mode 100644 index 0000000000..4e471a40c9 --- /dev/null +++ b/packages/agent-core/test/tools/cron/no-date-now.test.ts @@ -0,0 +1,78 @@ +/** + * Guard: forbid `Date.now()` in cron scheduler-adjacent files. + * + * The cron scheduler routes every wall-clock read through + * `ClockSources.wallNow()` so that tests and benches can inject a + * simulated clock (see `tools/cron/clock.ts`). A stray `Date.now()` + * call inside `scheduler.ts` / `persist.ts` / `lock.ts` / `jitter.ts` + * silently bypasses that injection — the bug is invisible until the + * bench notices its frozen clock did not freeze a heartbeat. + * + * The natural place for this guard is an ESLint rule + * (`no-restricted-syntax` matching `CallExpression[callee.object.name= + * "Date"][callee.property.name="now"]`), but this repo lints with + * **oxlint**, and oxlint 1.59 does not implement `no-restricted-syntax` + * — only `no-restricted-globals`, `no-restricted-imports`, + * `no-restricted-exports`, and `no-restricted-types`. Loading the rule + * makes oxlint refuse to parse the config: + * + * Failed to parse oxlint configuration file. + * x Rule 'no-restricted-syntax' not found in plugin 'eslint' + * + * As a fallback, we scan the source files here from a vitest test that + * runs in the regular `pnpm test` flow. + * + * Important: `clock.ts` is deliberately **excluded** — it is the one + * file where `Date.now()` is the *implementation* of the wall-clock + * abstraction (see `SYSTEM_CLOCKS` and the parse-failure fallbacks in + * `readEnvWall` / `readFileWall`). Banning `Date.now()` there would be + * banning the abstraction's only legal definition. + * + * Files that don't exist yet (`scheduler.ts`, `persist.ts`, `lock.ts`) + * are skipped — the guard activates automatically when later commits + * introduce them. + */ +import { existsSync, readFileSync } from 'node:fs'; +import { describe, expect, it } from 'vitest'; +import { dirname, join } from 'pathe'; +import { fileURLToPath } from 'node:url'; + +const here = dirname(fileURLToPath(import.meta.url)); +// `test/tools/cron/` → package root → `src/tools/cron/`. +const cronSrcDir = join(here, '..', '..', '..', 'src', 'tools', 'cron'); + +const GUARDED_FILES = [ + 'scheduler.ts', + 'persist.ts', + 'lock.ts', + 'jitter.ts', +] as const; + +// Matches a `Date.now(` call. Word boundary on the `D` side so it +// won't trip on `myDate.now(` or `notDate.now(`; arbitrary whitespace +// between `now` and `(` so `Date.now ()` and `Date . now (` both +// catch. The intent is "the CallExpression `Date.now(...)`" — this +// regex is the cheap proxy for the AST selector we'd use in ESLint. +const DATE_NOW_REGEX = /\bDate\s*\.\s*now\s*\(/; + +describe('cron scheduler files do not call Date.now()', () => { + for (const file of GUARDED_FILES) { + it(`${file} contains no Date.now() call`, () => { + const path = join(cronSrcDir, file); + if (!existsSync(path)) { + // File hasn't been added yet (P1/P2 commits introduce + // scheduler.ts, persist.ts, lock.ts). The guard activates + // automatically once they exist. + return; + } + const source = readFileSync(path, 'utf8'); + const match = DATE_NOW_REGEX.exec(source); + expect( + match, + `Found \`Date.now()\` in ${file} at offset ${match?.index ?? -1}. ` + + `Use ClockSources.wallNow() instead — direct Date.now() bypasses ` + + `test/bench clock injection. clock.ts is the single legal exception.`, + ).toBeNull(); + }); + } +}); From 1785457aef16c4551f64851a0299a85d521ae3ba Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 27 May 2026 19:23:42 +0800 Subject: [PATCH 06/26] feat(agent-core): add cron telemetry event-name constants Four event names emitted by later commits (cron_scheduled, cron_fired, cron_missed, cron_deleted) live with the cron module rather than in the generic telemetry interface so a typo can't drift the metric and the abstraction stays domain-free. --- .../agent-core/src/tools/cron/telemetry-events.ts | 11 +++++++++++ 1 file changed, 11 insertions(+) create mode 100644 packages/agent-core/src/tools/cron/telemetry-events.ts diff --git a/packages/agent-core/src/tools/cron/telemetry-events.ts b/packages/agent-core/src/tools/cron/telemetry-events.ts new file mode 100644 index 0000000000..2f88afd9eb --- /dev/null +++ b/packages/agent-core/src/tools/cron/telemetry-events.ts @@ -0,0 +1,11 @@ +/** Telemetry event names emitted by the cron subsystem. Centralised so a typo can't drift a metric. */ +export const CRON_SCHEDULED = 'cron_scheduled' as const; +export const CRON_FIRED = 'cron_fired' as const; +export const CRON_MISSED = 'cron_missed' as const; +export const CRON_DELETED = 'cron_deleted' as const; + +export type CronTelemetryEvent = + | typeof CRON_SCHEDULED + | typeof CRON_FIRED + | typeof CRON_MISSED + | typeof CRON_DELETED; From 2e09ef8b322dcee5f2c54c0b464645bec3c8feb7 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 27 May 2026 19:27:12 +0800 Subject: [PATCH 07/26] feat(agent-core): add in-memory SessionCronStore MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Holds cron tasks scoped to a single CLI session — vanish on exit. Phase 2 will add a file-backed sibling that shares the shape. Ids are 8 hex characters with a collision-retry cap; createdAt is supplied by the caller's wall clock so the store stays clock-pure (and exempt from the Date.now() guard for the same reason). --- .../src/tools/cron/session-store.ts | 118 +++++++++++ .../test/tools/cron/session-store.test.ts | 187 ++++++++++++++++++ 2 files changed, 305 insertions(+) create mode 100644 packages/agent-core/src/tools/cron/session-store.ts create mode 100644 packages/agent-core/test/tools/cron/session-store.test.ts diff --git a/packages/agent-core/src/tools/cron/session-store.ts b/packages/agent-core/src/tools/cron/session-store.ts new file mode 100644 index 0000000000..f26333ca79 --- /dev/null +++ b/packages/agent-core/src/tools/cron/session-store.ts @@ -0,0 +1,118 @@ +/** + * SessionCronStore — in-memory cron task store for a single CLI session. + * + * Phase 1 cron storage. Tasks added here live only for the lifetime of + * the current process; on exit they vanish. The Phase 2 file-backed + * store under `/.kimi-code/cron/` will share the same shape + * (add / get / list / remove) but write to disk so durable tasks + * survive restart. + * + * The store is intentionally clock-agnostic: it does NOT call + * `Date.now()` itself. Callers pass `nowMs` (which the cron manager + * sources from `ClockSources.wallNow()`), so injected clocks in tests + * and benches stay authoritative. The `no-date-now` guard does not + * currently list this file, but the discipline matches. + * + * Insertion order is preserved by relying on `Map` iteration order — + * callers (CronList, scheduler `source: () => CronTask[]`) want a + * stable ordering that matches the user's mental model of "the one I + * added first comes first". + */ + +import { randomBytes } from 'node:crypto'; + +import type { CronTask } from './types'; + +/** + * Input to {@link SessionCronStore.add}: everything the caller supplies, + * minus `id` and `createdAt` which the store generates. + */ +export type SessionCronTaskInit = Omit; + +/** Matches the canonical cron task id shape (8 lower-hex chars). */ +const ID_REGEX = /^[0-9a-f]{8}$/; + +/** + * Upper bound on id-collision retries. With 32 bits of entropy and at + * most a few dozen live tasks per session, the probability of even one + * collision is on the order of 1e-8. Eight attempts is a hard ceiling + * to surface a real bug (e.g. PRNG degeneration) rather than silently + * spinning. + */ +const MAX_ID_ATTEMPTS = 8; + +export class SessionCronStore { + /** + * Backing map. `Map` preserves insertion order in JS, which we rely on + * for {@link list}. + */ + private readonly tasks = new Map(); + + /** + * Generate a fresh 8-hex id and add the task. `createdAt` is set to + * the supplied `nowMs` — the store never reads its own clock. + * + * Throws if the PRNG fails to produce an unused id within + * {@link MAX_ID_ATTEMPTS} attempts. That should be unreachable in + * practice; surfacing it as a throw beats silently retrying forever. + */ + add(init: SessionCronTaskInit, nowMs: number): CronTask { + const id = this.generateUniqueId(); + const task: CronTask = { + ...init, + id, + createdAt: nowMs, + }; + this.tasks.set(id, task); + return task; + } + + /** Returns the task or `undefined`. */ + get(id: string): CronTask | undefined { + return this.tasks.get(id); + } + + /** + * Snapshot in insertion order. Returns a fresh array on every call — + * callers may mutate the returned array without affecting the store, + * and successive calls return distinct array references. + */ + list(): readonly CronTask[] { + return Array.from(this.tasks.values()); + } + + /** + * Remove the given ids. Returns the subset that were actually present + * (so the caller can detect already-missing ids and report them). + * Order of the returned ids follows the input order, not insertion + * order. + */ + remove(ids: readonly string[]): readonly string[] { + const removed: string[] = []; + for (const id of ids) { + if (this.tasks.delete(id)) { + removed.push(id); + } + } + return removed; + } + + /** Empty the store. Convenience for tests / shutdown. */ + clear(): void { + this.tasks.clear(); + } + + private generateUniqueId(): string { + for (let attempt = 0; attempt < MAX_ID_ATTEMPTS; attempt++) { + const candidate = randomBytes(4).toString('hex'); + // randomBytes(4).toString('hex') is always 8 lowercase hex chars, + // so the regex check is belt-and-braces against future refactors + // that swap the id source. + if (!ID_REGEX.test(candidate)) continue; + if (!this.tasks.has(candidate)) return candidate; + } + throw new Error( + `SessionCronStore: failed to generate a unique 8-hex id after ${MAX_ID_ATTEMPTS} attempts`, + ); + } +} diff --git a/packages/agent-core/test/tools/cron/session-store.test.ts b/packages/agent-core/test/tools/cron/session-store.test.ts new file mode 100644 index 0000000000..78ace34d81 --- /dev/null +++ b/packages/agent-core/test/tools/cron/session-store.test.ts @@ -0,0 +1,187 @@ +/** + * Tests for `tools/cron/session-store.ts`. + * + * Lives under `test/tools/cron/` because the package's `vitest.config.ts` + * scopes its `include` to `test/**`. See sibling `clock.test.ts` for the + * same rationale. + */ +import { describe, expect, it } from 'vitest'; + +import { + SessionCronStore, + type SessionCronTaskInit, +} from '../../../src/tools/cron/session-store'; + +const ID_REGEX = /^[0-9a-f]{8}$/; + +/** Convenience: minimal valid init that varies by suffix so tests can + * tell two tasks apart in `list()` snapshots. */ +function makeInit(suffix: string, overrides: Partial = {}): SessionCronTaskInit { + return { + cron: '*/5 * * * *', + prompt: `prompt-${suffix}`, + recurring: true, + durable: false, + ...overrides, + }; +} + +describe('SessionCronStore', () => { + describe('add', () => { + it('returns a task with id matching /^[0-9a-f]{8}$/', () => { + const store = new SessionCronStore(); + const task = store.add(makeInit('a'), 1000); + expect(task.id).toMatch(ID_REGEX); + }); + + it('preserves cron / prompt / recurring / durable from the init', () => { + const store = new SessionCronStore(); + const init: SessionCronTaskInit = { + cron: '0 9 * * 1-5', + prompt: 'sync PRs', + recurring: true, + durable: true, + }; + const task = store.add(init, 1000); + expect(task.cron).toBe('0 9 * * 1-5'); + expect(task.prompt).toBe('sync PRs'); + expect(task.recurring).toBe(true); + expect(task.durable).toBe(true); + }); + + it('sets createdAt to the supplied nowMs (no internal clock read)', () => { + const store = new SessionCronStore(); + const task = store.add(makeInit('a'), 1_700_000_000_000); + expect(task.createdAt).toBe(1_700_000_000_000); + }); + + it('does not consult Date.now()', () => { + // Sentinel: a nowMs deliberately not close to "now" makes + // accidental Date.now() use obvious in createdAt. + const store = new SessionCronStore(); + const task = store.add(makeInit('a'), 0); + expect(task.createdAt).toBe(0); + }); + + it('rapid sequential adds produce distinct ids', () => { + const store = new SessionCronStore(); + const ids = new Set(); + for (let i = 0; i < 32; i++) { + ids.add(store.add(makeInit(`x${i}`), 1000 + i).id); + } + expect(ids.size).toBe(32); + }); + }); + + describe('get', () => { + it('returns a previously-added task', () => { + const store = new SessionCronStore(); + const task = store.add(makeInit('a'), 1000); + expect(store.get(task.id)).toEqual(task); + }); + + it('returns undefined for an unknown id', () => { + const store = new SessionCronStore(); + expect(store.get('deadbeef')).toBeUndefined(); + }); + }); + + describe('list', () => { + it('returns tasks in insertion order', () => { + const store = new SessionCronStore(); + const t1 = store.add(makeInit('1'), 1000); + const t2 = store.add(makeInit('2'), 1001); + const t3 = store.add(makeInit('3'), 1002); + expect(store.list().map((t) => t.id)).toEqual([t1.id, t2.id, t3.id]); + }); + + it('returns a fresh array on each call (no aliasing)', () => { + const store = new SessionCronStore(); + store.add(makeInit('a'), 1000); + const a = store.list(); + const b = store.list(); + expect(a).not.toBe(b); + expect(a).toEqual(b); + }); + + it('mutating the returned array does not affect the store', () => { + const store = new SessionCronStore(); + store.add(makeInit('a'), 1000); + const snap = store.list() as CronTaskLike[]; + snap.length = 0; + expect(store.list()).toHaveLength(1); + }); + + it('returns empty array on a fresh store', () => { + const store = new SessionCronStore(); + expect(store.list()).toEqual([]); + }); + }); + + describe('remove', () => { + it('returns only the ids that were actually present', () => { + const store = new SessionCronStore(); + const t1 = store.add(makeInit('1'), 1000); + const t2 = store.add(makeInit('2'), 1001); + const removed = store.remove([t1.id, 'missing0', t2.id]); + expect(removed).toEqual([t1.id, t2.id]); + }); + + it('actually removes the tasks from list / get', () => { + const store = new SessionCronStore(); + const t1 = store.add(makeInit('1'), 1000); + store.remove([t1.id]); + expect(store.get(t1.id)).toBeUndefined(); + expect(store.list()).toHaveLength(0); + }); + + it('returns empty array when nothing matches', () => { + const store = new SessionCronStore(); + store.add(makeInit('a'), 1000); + expect(store.remove(['ffffffff', 'eeeeeeee'])).toEqual([]); + }); + + it('preserves insertion order of remaining tasks', () => { + const store = new SessionCronStore(); + const t1 = store.add(makeInit('1'), 1000); + const t2 = store.add(makeInit('2'), 1001); + const t3 = store.add(makeInit('3'), 1002); + store.remove([t2.id]); + expect(store.list().map((t) => t.id)).toEqual([t1.id, t3.id]); + }); + }); + + describe('clear', () => { + it('empties the store', () => { + const store = new SessionCronStore(); + store.add(makeInit('a'), 1000); + store.add(makeInit('b'), 1001); + store.clear(); + expect(store.list()).toEqual([]); + }); + + it('is a no-op on an already-empty store', () => { + const store = new SessionCronStore(); + expect(() => store.clear()).not.toThrow(); + expect(store.list()).toEqual([]); + }); + }); + + describe('id uniqueness at scale', () => { + it('256 adds produce 256 unique 8-hex ids', () => { + const store = new SessionCronStore(); + const ids = new Set(); + for (let i = 0; i < 256; i++) { + const task = store.add(makeInit(`x${i}`), 1000 + i); + expect(task.id).toMatch(ID_REGEX); + ids.add(task.id); + } + expect(ids.size).toBe(256); + expect(store.list()).toHaveLength(256); + }); + }); +}); + +/** Local alias for the mutate-snapshot test that needs a writable view + * of `readonly CronTask[]` without dragging in the real CronTask type. */ +type CronTaskLike = { id: string }; From e5a01a31c479bcb3cb4242f7beb12fa2f904e064 Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 27 May 2026 19:35:20 +0800 Subject: [PATCH 08/26] feat(agent-core): add session-only CronScheduler engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scheduler is a pure callback-driven loop: it gets tasks from a source(), gates on isIdle()/isKilled?(), computes next fire via cron-expr + jitter, and invokes onFire with the coalesced count when a task is due. lastSeenAt is in-memory only — coalesce semantics make a restart skip acceptable, but persisting last-fire would silently swallow legitimately-due fires. pollIntervalMs=null lets P1.8 disable the auto-tick timer for bench scenarios. --- .../agent-core/src/tools/cron/scheduler.ts | 356 +++++++++++++++ .../test/tools/cron/scheduler.test.ts | 411 ++++++++++++++++++ 2 files changed, 767 insertions(+) create mode 100644 packages/agent-core/src/tools/cron/scheduler.ts create mode 100644 packages/agent-core/test/tools/cron/scheduler.test.ts diff --git a/packages/agent-core/src/tools/cron/scheduler.ts b/packages/agent-core/src/tools/cron/scheduler.ts new file mode 100644 index 0000000000..0afcdd221f --- /dev/null +++ b/packages/agent-core/src/tools/cron/scheduler.ts @@ -0,0 +1,356 @@ +/** + * CronScheduler — the session-only scheduling engine. + * + * This is the bottom of the cron stack: it knows about tasks, clocks, + * jitter, and "is the REPL idle?", but nothing about agents, tools, + * persistence, locks, or the file system. The Phase 2 file-backed + * variant will extend this same module rather than fork it. + * + * Design notes worth keeping near the code: + * + * - **No direct wall-clock reads.** Every wall-clock read goes + * through `clocks.wallNow()`. The companion `no-date-now.test.ts` + * enforces this at the file level; bypassing the abstraction + * here would break bench / test clock injection. + * + * - **`source()` is called every tick.** It returns the *current* + * task list. Callers (the manager) typically wire it to + * `() => store.list()`, so creating / deleting tasks between ticks + * is picked up automatically. Keep `source()` cheap. + * + * - **`isIdle()` gates fires, not state updates.** When the REPL is + * mid-turn we skip firing — but we do NOT advance `lastSeenAt`. The + * next idle tick will see the tasks as still due and fire them, + * with `coalescedCount` reflecting the gap (so the LLM knows it + * missed N ideal fires while the user was talking). + * + * - **`coalescedCount` semantics.** When a sleep / busy turn / system + * pause causes the scheduler to miss multiple ideal fires, we + * deliver exactly ONE `onFire` call and tell the caller how many + * ideal fires we collapsed into it. Floor at 1 — every fire that + * happens counts as at least one occurrence. + * + * - **`inFlight` is cleared at end of tick.** `onFire` is synchronous + * (the manager's steer is fire-and-forget). The set exists only to + * defend against re-entrant ticks within the same call stack — a + * theoretical concern, but cheap insurance. + * + * - **Bad tasks do not poison the loop.** Each task's processing is + * wrapped in try/catch; failures are swallowed (with an optional + * stderr trace gated on `KIMI_CRON_DEBUG=1`) so one busted cron + * expression cannot starve the other tasks. + */ + +import type { ParsedCronExpression } from './cron-expr'; +import { computeNextCronRun, parseCronExpression } from './cron-expr'; +import type { ClockSources } from './clock'; +import { jitteredNextCronRunMs, oneShotJitteredNextCronRunMs } from './jitter'; +import type { CronTask } from './types'; + +export interface CronSchedulerOptions { + /** Required. Wall + monotonic clock source. */ + readonly clocks: ClockSources; + + /** + * Required. Returns the live task list (e.g. `() => store.list()`). + * Called every tick — keep cheap. + */ + readonly source: () => readonly CronTask[]; + + /** + * Required. Called when a task fires. `coalescedCount >= 1`; > 1 + * when the scheduler slept past multiple ideal fires. + */ + readonly onFire: (task: CronTask, ctx: { readonly coalescedCount: number }) => void; + + /** + * Required. Returns true when the REPL is idle and the scheduler + * may deliver a fire NOW. False during an active turn so we don't + * dump a cron fire mid-stream. When false, the tick returns without + * firing but does not lose tasks — the next idle tick fires them + * with coalescedCount reflecting the gap. + */ + readonly isIdle: () => boolean; + + /** + * Optional. Returns true when the global killswitch is on; tick() + * short-circuits to no-op. + */ + readonly isKilled?: () => boolean; + + /** + * Optional. Called when a one-shot task fires and must be removed + * from the store. Defaults to a no-op (manager is responsible). + */ + readonly removeOneShot?: (id: string) => void; + + /** + * Optional. Poll interval for the auto-tick setInterval, in ms. + * - undefined (default) → 1000ms. + * - 0 or null → no automatic polling. Caller drives tick() + * manually. + * + * Used by P1.8 to wire `KIMI_CRON_MANUAL_TICK=1` to disable the + * timer. + */ + readonly pollIntervalMs?: number | null; +} + +export interface CronScheduler { + /** Begin the auto-tick loop. Idempotent — calling twice is a no-op. */ + start(): void; + + /** + * Stop the auto-tick loop and clear any in-flight bookkeeping. + * Idempotent. + */ + stop(): Promise; + + /** + * Run one check cycle synchronously. Safe to call before start() or + * after stop(). + */ + tick(): void; + + /** + * Earliest theoretical (post-jitter) next fire across all current + * tasks, or null if there are no tasks or none have a future fire. + * Used by /cron and by external monitoring. + */ + getNextFireTime(): number | null; +} + +const DEFAULT_POLL_INTERVAL_MS = 1_000; + +/** + * Cap on how many ideal fires we attempt to enumerate when computing + * coalescedCount. With a 1-minute cron, this still covers 10 000 + * minutes (~7 days). Beyond that we'd rather report 10 000 than spin — + * the LLM only needs the order of magnitude. + */ +const MAX_COALESCE_ITERATIONS = 10_000; + +export function createCronScheduler(opts: CronSchedulerOptions): CronScheduler { + const { + clocks, + source, + onFire, + isIdle, + isKilled, + removeOneShot, + pollIntervalMs, + } = opts; + + // Cached parsed cron expressions. Keyed by the raw expression + // string. Per-session task counts are tiny, so we never evict. + const parsedCache = new Map(); + + // Per-task wall-clock baseline for "where did we last look from". + // NOT persisted — restart starts from `task.createdAt` as the floor. + const lastSeenAt = new Map(); + + // Defensive re-entry guard for the duration of a single tick. + const inFlight = new Set(); + + let timerHandle: ReturnType | null = null; + + function getParsed(expr: string): ParsedCronExpression { + const cached = parsedCache.get(expr); + if (cached !== undefined) return cached; + const parsed = parseCronExpression(expr); + parsedCache.set(expr, parsed); + return parsed; + } + + function debugLog(message: string): void { + if (process.env.KIMI_CRON_DEBUG === '1') { + process.stderr.write(`[cron/scheduler] ${message}\n`); + } + } + + /** + * Compute the jittered next-fire for a task, starting from `baseMs`. + * Returns null when the cron expression has no future fire within + * the search budget (legal-but-never-fires expression). + */ + function computeJitteredNext( + task: CronTask, + parsed: ParsedCronExpression, + baseMs: number, + ): number | null { + const ideal = computeNextCronRun(parsed, baseMs); + if (ideal === null) return null; + if (task.recurring === false) { + return oneShotJitteredNextCronRunMs(task, ideal); + } + return jitteredNextCronRunMs(task, parsed, ideal); + } + + /** + * Count how many ideal fires fall in `(firstFireMs - 1, nowMs]`. + * Always returns at least 1 — every actual fire is one occurrence. + * Capped at MAX_COALESCE_ITERATIONS as a defence against runaway + * loops; an expression that produces more than 10 000 fires in the + * gap is degenerate and the LLM only needs the order of magnitude. + */ + function countCoalesced( + parsed: ParsedCronExpression, + firstFireMs: number, + nowMs: number, + ): number { + let count = 1; + let cursor = firstFireMs; + while (count < MAX_COALESCE_ITERATIONS) { + const next = computeNextCronRun(parsed, cursor); + if (next === null) break; + if (next > nowMs) break; + count++; + cursor = next; + } + return count; + } + + function tick(): void { + if (isKilled?.() === true) return; + if (!isIdle()) return; + + const tasks = source(); + if (tasks.length === 0) return; + + const now = clocks.wallNow(); + + // We clear inFlight at the end of the tick; entry-time defence + // against re-entry handled by the `inFlight.has(id)` skip below. + try { + for (const task of tasks) { + try { + if (inFlight.has(task.id)) continue; + + const parsed = getParsed(task.cron); + + // Base from which to compute the next ideal fire. For a + // freshly-added task this is its createdAt; once we've fired + // (or seen it pass), bump to the wall clock at that moment + // so we don't double-count the same fire on the next tick. + const seen = lastSeenAt.get(task.id); + const baseFromMs = + seen !== undefined && seen > task.createdAt ? seen : task.createdAt; + + const nextFireAt = computeJitteredNext(task, parsed, baseFromMs); + if (nextFireAt === null) continue; + + if (now < nextFireAt) continue; + + // Due — compute coalescedCount starting from the first + // ideal fire (not the jittered one — jitter only shifts the + // delivery point, not the underlying schedule). + const ideal = computeNextCronRun(parsed, baseFromMs); + const coalescedCount = + ideal === null ? 1 : Math.max(1, countCoalesced(parsed, ideal, now)); + + inFlight.add(task.id); + try { + onFire(task, { coalescedCount }); + } catch (error) { + debugLog( + `onFire threw for task ${task.id}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + + if (task.recurring === false) { + // One-shot: ask the caller to remove and drop our memory of it. + try { + removeOneShot?.(task.id); + } catch (error) { + debugLog( + `removeOneShot threw for task ${task.id}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + lastSeenAt.delete(task.id); + } else { + // Recurring: advance baseline so the next tick computes + // the next-after-now ideal fire. + lastSeenAt.set(task.id, now); + } + } catch (error) { + // A single bad task must not stop the rest of the loop. + debugLog( + `tick failed for task ${task.id}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + } + } finally { + // onFire is synchronous so re-entrant ticks within the same + // call stack are the only thing inFlight defends against. Clear + // at end-of-tick to keep the invariant simple. + inFlight.clear(); + } + } + + function start(): void { + if (timerHandle !== null) return; + + const interval = + pollIntervalMs === undefined ? DEFAULT_POLL_INTERVAL_MS : pollIntervalMs; + // 0 and null both mean "no automatic polling". + if (interval === null || interval === 0) return; + + const handle = setInterval(tick, interval); + // Don't keep the event loop alive on the scheduler alone — the + // user's REPL / agent owns lifetime. + if (typeof handle === 'object' && handle !== null && 'unref' in handle) { + (handle as { unref: () => void }).unref(); + } + timerHandle = handle; + } + + async function stop(): Promise { + if (timerHandle !== null) { + clearInterval(timerHandle); + timerHandle = null; + } + inFlight.clear(); + lastSeenAt.clear(); + parsedCache.clear(); + // Async signature for forward compatibility with Phase 2 (file + // I/O cleanup, lock release). Session-only resolves immediately. + } + + function getNextFireTime(): number | null { + const tasks = source(); + if (tasks.length === 0) return null; + + let min: number | null = null; + for (const task of tasks) { + try { + const parsed = getParsed(task.cron); + const seen = lastSeenAt.get(task.id); + const baseFromMs = + seen !== undefined && seen > task.createdAt ? seen : task.createdAt; + const next = computeJitteredNext(task, parsed, baseFromMs); + if (next === null) continue; + if (min === null || next < min) min = next; + } catch (error) { + debugLog( + `getNextFireTime skipping task ${task.id}: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + } + } + return min; + } + + return { + start, + stop, + tick, + getNextFireTime, + }; +} diff --git a/packages/agent-core/test/tools/cron/scheduler.test.ts b/packages/agent-core/test/tools/cron/scheduler.test.ts new file mode 100644 index 0000000000..1b3074ad05 --- /dev/null +++ b/packages/agent-core/test/tools/cron/scheduler.test.ts @@ -0,0 +1,411 @@ +/** + * Tests for `tools/cron/scheduler.ts`. + * + * The scheduler is session-only at this commit: no file I/O, no + * locks. Every test injects a `ClockSources` whose `wallNow` / + * `monoNowMs` are driven by hand via the `advance(ms)` helper, so + * scenarios run in simulated time without `setTimeout`. + * + * For the recurring tests we deliberately use `KIMI_CRON_NO_JITTER=1` + * so the assertions can pin exact fire counts. The one jitter-aware + * test gives a generous slack window. + */ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; + +import type { ClockSources } from '../../../src/tools/cron/clock'; +import { + createCronScheduler, + type CronScheduler, +} from '../../../src/tools/cron/scheduler'; +import type { CronTask } from '../../../src/tools/cron/types'; + +interface HarnessOptions { + readonly isIdle?: boolean; + readonly isKilled?: boolean; + readonly pollIntervalMs?: number | null; +} + +interface Harness { + readonly scheduler: CronScheduler; + readonly tasks: CronTask[]; + readonly fired: Array<{ task: CronTask; coalescedCount: number }>; + readonly removed: string[]; + advance(ms: number): void; + setIdle(v: boolean): void; + setKilled(v: boolean): void; + now(): number; +} + +// Fixed wall-clock anchor (Nov 14 2023, 22:13:20 UTC). Picked so it +// doesn't sit on any "round" minute mark — the next "every 5 minutes" +// fire lands 4-5 minutes ahead, not exactly 5. +const WALL_ANCHOR = 1_700_000_000_000; + +function createHarness(opts: HarnessOptions = {}): Harness { + let now = WALL_ANCHOR; + let mono = 1_000_000; + const clocks: ClockSources = { + wallNow: () => now, + monoNowMs: () => mono, + }; + const tasks: CronTask[] = []; + const fired: Array<{ task: CronTask; coalescedCount: number }> = []; + const removed: string[] = []; + let idle = opts.isIdle ?? true; + let killed = opts.isKilled ?? false; + + const scheduler = createCronScheduler({ + clocks, + source: () => tasks, + onFire: (task, ctx) => fired.push({ task, coalescedCount: ctx.coalescedCount }), + isIdle: () => idle, + isKilled: () => killed, + removeOneShot: (id) => { + removed.push(id); + const i = tasks.findIndex((t) => t.id === id); + if (i >= 0) tasks.splice(i, 1); + }, + // Tests use manual tick() unless they explicitly opt into the timer. + pollIntervalMs: opts.pollIntervalMs ?? null, + }); + + return { + scheduler, + tasks, + fired, + removed, + advance: (ms: number) => { + now += ms; + mono += ms; + }, + setIdle: (v: boolean) => { + idle = v; + }, + setKilled: (v: boolean) => { + killed = v; + }, + now: () => now, + }; +} + +/** + * Tiny helper: 8-hex id deterministic per-call so each test gets a + * stable id without colliding across tasks within a test. + */ +let idCounter = 0; +function nextId(): string { + idCounter++; + return idCounter.toString(16).padStart(8, '0'); +} + +function makeTask(overrides: Partial & { cron: string; createdAt: number }): CronTask { + return { + id: overrides.id ?? nextId(), + prompt: overrides.prompt ?? 'do the thing', + cron: overrides.cron, + createdAt: overrides.createdAt, + recurring: overrides.recurring, + durable: overrides.durable, + }; +} + +const ORIGINAL_ENV_NO_JITTER = process.env.KIMI_CRON_NO_JITTER; + +describe('createCronScheduler — tick behaviour', () => { + beforeEach(() => { + // Pin exact fire times in every test by default. The single test + // that exercises jitter restores the env explicitly. + process.env.KIMI_CRON_NO_JITTER = '1'; + idCounter = 0; + }); + + afterEach(() => { + if (ORIGINAL_ENV_NO_JITTER === undefined) { + delete process.env.KIMI_CRON_NO_JITTER; + } else { + process.env.KIMI_CRON_NO_JITTER = ORIGINAL_ENV_NO_JITTER; + } + }); + + it('recurring task fires once when due', () => { + const h = createHarness(); + h.tasks.push( + makeTask({ cron: '*/5 * * * *', createdAt: h.now(), recurring: true }), + ); + // Not yet due — first call to tick should not fire. + h.scheduler.tick(); + expect(h.fired).toHaveLength(0); + + // Advance ~6 minutes — past the next */5 boundary regardless of + // where WALL_ANCHOR landed in the minute cycle. + h.advance(6 * 60_000); + h.scheduler.tick(); + + expect(h.fired).toHaveLength(1); + expect(h.fired[0]!.coalescedCount).toBe(1); + }); + + it('recurring task coalesces missed fires when scheduler sleeps', () => { + const h = createHarness(); + h.tasks.push( + makeTask({ cron: '*/5 * * * *', createdAt: h.now(), recurring: true }), + ); + // Advance 15 minutes — we should see 3 collapsed ideal fires. + h.advance(15 * 60_000); + h.scheduler.tick(); + + expect(h.fired).toHaveLength(1); + expect(h.fired[0]!.coalescedCount).toBeGreaterThanOrEqual(3); + expect(h.fired[0]!.coalescedCount).toBeLessThanOrEqual(4); + }); + + it('one-shot task fires once then is removed', () => { + const h = createHarness(); + // createdAt is 1 minute earlier so we can advance past the next + // 12:00 without anchoring needing a precise local-tz computation. + h.tasks.push( + makeTask({ + cron: '0 12 * * *', + createdAt: h.now() - 60_000, + recurring: false, + }), + ); + const taskId = h.tasks[0]!.id; + + // Advance one full day — guaranteed to pass at least one 12:00. + h.advance(25 * 60 * 60_000); + h.scheduler.tick(); + + expect(h.fired).toHaveLength(1); + expect(h.fired[0]!.coalescedCount).toBe(1); + expect(h.removed).toEqual([taskId]); + expect(h.tasks).toHaveLength(0); + }); + + it('isIdle=false suppresses fire but does not lose the task', () => { + const h = createHarness({ isIdle: false }); + h.tasks.push( + makeTask({ cron: '*/5 * * * *', createdAt: h.now(), recurring: true }), + ); + + // Five minutes pass mid-turn — tick should not fire. + h.advance(6 * 60_000); + h.scheduler.tick(); + expect(h.fired).toHaveLength(0); + + // Turn ends — the next tick picks the missed fire up. + h.setIdle(true); + h.scheduler.tick(); + expect(h.fired).toHaveLength(1); + expect(h.fired[0]!.coalescedCount).toBe(1); + }); + + it('isKilled=true short-circuits even when due and idle', () => { + const h = createHarness({ isKilled: true }); + h.tasks.push( + makeTask({ cron: '*/5 * * * *', createdAt: h.now(), recurring: true }), + ); + h.advance(6 * 60_000); + h.scheduler.tick(); + expect(h.fired).toHaveLength(0); + + // Lifting the killswitch lets the missed fire through. + h.setKilled(false); + h.scheduler.tick(); + expect(h.fired).toHaveLength(1); + }); + + it('bad cron expression does not stop other tasks in the same tick', () => { + const h = createHarness(); + // Insert a task whose cron string will throw at parse time. We + // bypass any normal "constructor" by pushing a raw object — the + // scheduler must swallow the parse failure for this one task and + // continue processing the rest. + const bad: CronTask = { + id: nextId(), + cron: 'not a cron at all', + prompt: 'bad', + createdAt: h.now(), + recurring: true, + }; + const good: CronTask = { + id: nextId(), + cron: '*/5 * * * *', + prompt: 'good', + createdAt: h.now(), + recurring: true, + }; + h.tasks.push(bad, good); + + h.advance(6 * 60_000); + h.scheduler.tick(); + + expect(h.fired).toHaveLength(1); + expect(h.fired[0]!.task.id).toBe(good.id); + }); + + it('only the due task fires when two tasks are scheduled', () => { + const h = createHarness(); + h.tasks.push( + makeTask({ cron: '*/5 * * * *', createdAt: h.now(), recurring: true }), + makeTask({ cron: '0 0 1 1 *', createdAt: h.now(), recurring: true }), + ); + const dueId = h.tasks[0]!.id; + + h.advance(6 * 60_000); + h.scheduler.tick(); + + expect(h.fired).toHaveLength(1); + expect(h.fired[0]!.task.id).toBe(dueId); + }); + + it('recurring fires exactly once per turn even if multiple ideal fires elapse mid-turn', () => { + // Regression for US-15: a long busy turn over the 1-hour boundary + // must collapse into a single fire when idle returns. + const h = createHarness({ isIdle: false }); + h.tasks.push( + makeTask({ cron: '*/5 * * * *', createdAt: h.now(), recurring: true }), + ); + + // 17 minutes elapse mid-turn — three ideal fires missed. + h.advance(17 * 60_000); + h.scheduler.tick(); + expect(h.fired).toHaveLength(0); + + h.setIdle(true); + h.scheduler.tick(); + expect(h.fired).toHaveLength(1); + expect(h.fired[0]!.coalescedCount).toBeGreaterThanOrEqual(3); + }); +}); + +describe('createCronScheduler — getNextFireTime', () => { + beforeEach(() => { + process.env.KIMI_CRON_NO_JITTER = '1'; + idCounter = 0; + }); + + afterEach(() => { + if (ORIGINAL_ENV_NO_JITTER === undefined) { + delete process.env.KIMI_CRON_NO_JITTER; + } else { + process.env.KIMI_CRON_NO_JITTER = ORIGINAL_ENV_NO_JITTER; + } + }); + + it('returns null when there are no tasks', () => { + const h = createHarness(); + expect(h.scheduler.getNextFireTime()).toBeNull(); + }); + + it('returns the minimum next-fire across tasks', () => { + const h = createHarness(); + // Soonest: every minute. Latest: yearly Jan 1. + h.tasks.push( + makeTask({ cron: '* * * * *', createdAt: h.now(), recurring: true }), + makeTask({ cron: '0 0 1 1 *', createdAt: h.now(), recurring: true }), + ); + + const next = h.scheduler.getNextFireTime(); + expect(next).not.toBeNull(); + // The minute-cron's next fire is within 1 minute of now; the + // yearly cron is at least days away. So `next` must be within + // 65s of now. + expect(next! - h.now()).toBeLessThanOrEqual(65_000); + }); +}); + +describe('createCronScheduler — start/stop lifecycle', () => { + beforeEach(() => { + process.env.KIMI_CRON_NO_JITTER = '1'; + idCounter = 0; + }); + + afterEach(() => { + if (ORIGINAL_ENV_NO_JITTER === undefined) { + delete process.env.KIMI_CRON_NO_JITTER; + } else { + process.env.KIMI_CRON_NO_JITTER = ORIGINAL_ENV_NO_JITTER; + } + }); + + it('start() with pollIntervalMs=null does not auto-tick', async () => { + const h = createHarness({ pollIntervalMs: null }); + h.tasks.push( + makeTask({ cron: '*/5 * * * *', createdAt: h.now(), recurring: true }), + ); + h.scheduler.start(); + h.advance(60 * 60_000); + // Yield so any (unintended) timer would have fired. + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(h.fired).toHaveLength(0); + await h.scheduler.stop(); + }); + + it('start() with a small pollIntervalMs wires up the auto-tick', async () => { + const h = createHarness({ pollIntervalMs: 20 }); + h.tasks.push( + makeTask({ cron: '*/5 * * * *', createdAt: h.now(), recurring: true }), + ); + h.advance(6 * 60_000); // task is now due + h.scheduler.start(); + + // Wait long enough for at least one interval tick to elapse. + await new Promise((resolve) => setTimeout(resolve, 80)); + await h.scheduler.stop(); + + expect(h.fired.length).toBeGreaterThanOrEqual(1); + // Coalesced to a single fire — auto-ticks subsequent to the first + // see lastSeenAt = now and have nothing new to fire. + expect(h.fired[0]!.coalescedCount).toBe(1); + }); + + it('stop() is idempotent and clears state', async () => { + const h = createHarness({ pollIntervalMs: null }); + h.scheduler.start(); + await h.scheduler.stop(); + await h.scheduler.stop(); + // Calling tick() after stop is still safe. + h.scheduler.tick(); + expect(h.fired).toHaveLength(0); + }); + + it('start() is idempotent', () => { + const h = createHarness({ pollIntervalMs: null }); + expect(() => { + h.scheduler.start(); + h.scheduler.start(); + }).not.toThrow(); + // No auto-tick (interval is null) → no fires sneak in. + expect(h.fired).toHaveLength(0); + }); +}); + +describe('createCronScheduler — jitter integration', () => { + beforeEach(() => { + delete process.env.KIMI_CRON_NO_JITTER; + idCounter = 0; + }); + + afterEach(() => { + if (ORIGINAL_ENV_NO_JITTER === undefined) { + delete process.env.KIMI_CRON_NO_JITTER; + } else { + process.env.KIMI_CRON_NO_JITTER = ORIGINAL_ENV_NO_JITTER; + } + }); + + it('recurring task with jitter fires after advancing past the cap', () => { + const h = createHarness(); + // 30s past 6 minutes guarantees we crossed the ideal + max(10% of + // 5min = 30s) jitter cap. + h.tasks.push( + makeTask({ cron: '*/5 * * * *', createdAt: h.now(), recurring: true }), + ); + h.advance(6 * 60_000 + 30_000); + h.scheduler.tick(); + + expect(h.fired).toHaveLength(1); + expect(h.fired[0]!.coalescedCount).toBe(1); + }); +}); From 60a87f70c0b71170508ce014d17717f2e3b92c7a Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 27 May 2026 19:42:20 +0800 Subject: [PATCH 09/26] feat(agent-core): add CronManager Agent integration layer CronManager owns a SessionCronStore + CronScheduler and wires them to the Agent: scheduler.isIdle reads agent.turn.hasActiveTurn, isKilled reads KIMI_DISABLE_CRON, onFire builds a CronJobOrigin and routes through agent.turn.steer. Stale flag is computed on read (7-day age, recurring only, KIMI_CRON_NO_STALE shorts it) so manager doesn't have to mutate persisted tasks. handleMissed takes a renderer callback so P2.7 can plug in the AskUserQuestion text without bringing render imports into Phase 1. --- packages/agent-core/src/agent/cron/index.ts | 1 + packages/agent-core/src/agent/cron/manager.ts | 225 ++++++++ .../test/agent/cron/manager.test.ts | 490 ++++++++++++++++++ 3 files changed, 716 insertions(+) create mode 100644 packages/agent-core/src/agent/cron/index.ts create mode 100644 packages/agent-core/src/agent/cron/manager.ts create mode 100644 packages/agent-core/test/agent/cron/manager.test.ts diff --git a/packages/agent-core/src/agent/cron/index.ts b/packages/agent-core/src/agent/cron/index.ts new file mode 100644 index 0000000000..fca7804e82 --- /dev/null +++ b/packages/agent-core/src/agent/cron/index.ts @@ -0,0 +1 @@ +export { CronManager, type CronManagerOptions } from './manager'; diff --git a/packages/agent-core/src/agent/cron/manager.ts b/packages/agent-core/src/agent/cron/manager.ts new file mode 100644 index 0000000000..a3b4863a60 --- /dev/null +++ b/packages/agent-core/src/agent/cron/manager.ts @@ -0,0 +1,225 @@ +/** + * CronManager — Agent-facing facade for the session-only cron scheduler. + * + * This layer sits between the raw `CronScheduler` (which knows nothing + * about agents) and the rest of the agent runtime (Agent / turn / + * telemetry / tool surface). Its job is small but important: + * + * - own the `SessionCronStore` for this session; + * - hand `() => store.list()` to the scheduler so add / delete are + * picked up automatically every tick; + * - gate fires on `agent.turn.hasActiveTurn` rather than maintaining a + * duplicate idle flag — the turn machinery already knows; + * - translate a fired `CronTask` into a `steer(...)` call carrying a + * `CronJobOrigin`, plus the `cron_fired` telemetry event; + * - provide a `handleMissed(...)` entry point that Phase 2 boot-time + * missed-task detection (P2.7 / P2.11) will call. In Phase 1 the + * entry point is reachable but never invoked from the framework — + * it is exposed now so the API surface stays stable across the + * phase boundary. + * + * The manager does NOT read `Date.now()` directly anywhere; every + * wall-clock read goes through `this.clocks.wallNow()`. The + * `no-date-now.test.ts` guard does not list this file (it covers the + * scheduler / jitter layer), but the same discipline is intentional so + * bench / test clock injection holds end-to-end. + * + * Note on `recurring` semantics: the canonical task representation uses + * `recurring: boolean | undefined` where `undefined` means recurring + * (cron tasks default to repeating). One-shot is the explicit + * `recurring === false` opt-out. Every check in this file uses + * `task.recurring !== false` to keep that default behaviour even when + * the field is omitted by the caller. + */ +import type { ContentPart } from '@moonshot-ai/kosong'; + +import type { Agent } from '../index'; +import type { CronJobOrigin, CronMissedOrigin } from '../context/types'; +import { + resolveClockSources, + SYSTEM_CLOCKS, + type ClockSources, +} from '../../tools/cron/clock'; +import { SessionCronStore } from '../../tools/cron/session-store'; +import { + createCronScheduler, + type CronScheduler, +} from '../../tools/cron/scheduler'; +import { + CRON_FIRED, + CRON_MISSED, +} from '../../tools/cron/telemetry-events'; +import type { CronTask } from '../../tools/cron/types'; + +/** + * Threshold past which a recurring task is flagged `stale: true` on its + * fire `origin`. One-shot tasks never carry the stale flag — they are + * one-time, "we always fire at most once" by construction. Disabled by + * `KIMI_CRON_NO_STALE=1` (bench / acceptance tests). + * + * Seven days mirrors the wall-clock "this got forgotten about" window + * we want the LLM to notice; the figure also matches the auto-expire + * cadence documented in the user-facing schedule story. + */ +const STALE_THRESHOLD_MS = 7 * 24 * 60 * 60 * 1000; + +export interface CronManagerOptions { + /** + * Override for tests / bench. Defaults to + * `resolveClockSources(process.env.KIMI_CRON_CLOCK)` so production + * picks up `KIMI_CRON_CLOCK=env:...` / `file:...` automatically. + * When unset, falls through to {@link SYSTEM_CLOCKS}. + */ + readonly clocks?: ClockSources; + + /** + * Override scheduler poll interval. Defaults handled by the scheduler + * (1000ms unless `KIMI_CRON_MANUAL_TICK` is set; that env wiring lives + * in P1.8). `null` or `0` means "no automatic timer — caller drives + * `tick()` manually". + */ + readonly pollIntervalMs?: number | null; +} + +export class CronManager { + /** Session-only task store. Empty at construction. */ + readonly store: SessionCronStore; + + /** + * Clock source used for the stale judgment. Also passed to the + * scheduler so the entire stack shares one notion of "now". + */ + readonly clocks: ClockSources; + + private readonly scheduler: CronScheduler; + private readonly agent: Agent; + + constructor(agent: Agent, opts: CronManagerOptions = {}) { + this.agent = agent; + this.store = new SessionCronStore(); + this.clocks = + opts.clocks ?? + resolveClockSources(process.env.KIMI_CRON_CLOCK) ?? + SYSTEM_CLOCKS; + + this.scheduler = createCronScheduler({ + clocks: this.clocks, + source: () => this.store.list(), + isIdle: () => !agent.turn.hasActiveTurn, + isKilled: () => process.env.KIMI_DISABLE_CRON === '1', + onFire: (task, ctx) => this.handleFire(task, ctx), + removeOneShot: (id) => { + this.store.remove([id]); + }, + pollIntervalMs: opts.pollIntervalMs, + }); + } + + /** Begin the scheduler's auto-tick loop. Idempotent. */ + start(): void { + this.scheduler.start(); + } + + /** Stop the scheduler and clear in-flight bookkeeping. Idempotent. */ + stop(): Promise { + return this.scheduler.stop(); + } + + /** Drive one scheduler tick synchronously. Used by tests + P1.8 SIGUSR1. */ + tick(): void { + this.scheduler.tick(); + } + + /** + * Earliest theoretical (post-jitter) next-fire across all tasks, or + * null if there are no tasks / none have a future fire. Used by the + * `/cron` slash command and external monitoring. + */ + getNextFireTime(): number | null { + return this.scheduler.getNextFireTime(); + } + + /** + * Stale judgment. + * + * - `KIMI_CRON_NO_STALE=1` short-circuits to false (bench). + * - One-shot tasks (`recurring === false`) are never stale — they + * fire at most once by construction; flagging them stale would be + * a noisy false positive on every backlog wakeup. + * - Otherwise: `wallNow() - createdAt >= 7 days`. + * + * `Number.isFinite` guards against the wall clock being broken (e.g. + * a mis-set bench env that returns `NaN`); a non-finite age is + * treated as "we don't know, don't claim stale". + */ + isStale(task: CronTask): boolean { + if (process.env.KIMI_CRON_NO_STALE === '1') return false; + if (task.recurring === false) return false; + const age = this.clocks.wallNow() - task.createdAt; + return Number.isFinite(age) && age >= STALE_THRESHOLD_MS; + } + + /** + * Translate a scheduler fire into a steer + telemetry event. + * + * `agent.turn.steer` returns the new turnId, or `null` when the input + * was buffered because a turn is in flight (see turn/index.ts:84). + * We propagate that as `buffered` on the telemetry props so dashboards + * can distinguish "fired into a fresh turn" from "fired into a steer + * buffer that may not run until the user's turn ends". + */ + private handleFire( + task: CronTask, + ctx: { readonly coalescedCount: number }, + ): void { + const stale = this.isStale(task); + const origin: CronJobOrigin = { + kind: 'cron_job', + jobId: task.id, + cron: task.cron, + recurring: task.recurring !== false, + coalescedCount: ctx.coalescedCount, + stale, + }; + const content: ContentPart[] = [{ type: 'text', text: task.prompt }]; + const turnId = this.agent.turn.steer(content, origin); + this.agent.telemetry.track(CRON_FIRED, { + recurring: task.recurring !== false, + durable: task.durable === true, + coalesced_count: ctx.coalescedCount, + stale, + buffered: turnId === null, + }); + } + + /** + * Called from P2.7 / P2.11 when boot-time missed one-shot tasks are + * detected. Stubbed in P1.3 because Phase 1 has no persistence — but + * the API surface is final so the consumer (file-backed missed-task + * detector) can land without touching this class again. + * + * The `renderMissedNotification` callback is supplied by the caller + * (rather than imported here) so this module stays free of UI / copy + * coupling; the same manager works for tests that want to inject a + * trivial renderer. + * + * `count: 0` is a no-op — the scheduler-side missed-task detector + * filters empties before calling us, but defending here keeps the + * contract simple ("safe to call with anything, no-op when empty"). + */ + handleMissed( + tasks: readonly CronTask[], + renderMissedNotification: ( + tasks: readonly CronTask[], + ) => readonly ContentPart[], + ): void { + if (tasks.length === 0) return; + const content = renderMissedNotification(tasks); + const origin: CronMissedOrigin = { + kind: 'cron_missed', + count: tasks.length, + }; + this.agent.turn.steer(content, origin); + this.agent.telemetry.track(CRON_MISSED, { count: tasks.length }); + } +} diff --git a/packages/agent-core/test/agent/cron/manager.test.ts b/packages/agent-core/test/agent/cron/manager.test.ts new file mode 100644 index 0000000000..a48c6b9b69 --- /dev/null +++ b/packages/agent-core/test/agent/cron/manager.test.ts @@ -0,0 +1,490 @@ +/** + * Tests for `agent/cron/manager.ts`. + * + * The manager is tested against a lightweight Agent stub rather than a + * real Agent. That keeps the test fast, deterministic, and focused on + * exactly the surface the manager actually touches: + * + * - `agent.turn.hasActiveTurn` (getter) + * - `agent.turn.steer(content, origin)` (returns turnId or null) + * - `agent.telemetry.track(event, props)` + * + * Building a real Agent here would drag in kosong / records / context + * for no incremental coverage — the wiring path is verified separately + * in P1.7 (agent lifecycle + tool barrel). + * + * Time is injected via a `ClockSources` whose `wallNow` is driven by + * hand. `KIMI_CRON_NO_JITTER=1` is used everywhere we assert on a + * specific fire count so the jitter window doesn't make assertions + * flaky; the stale-flag tests don't depend on the scheduler firing so + * jitter is irrelevant there. + */ +import type { ContentPart } from '@moonshot-ai/kosong'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import type { Agent } from '../../../src/agent'; +import type { PromptOrigin } from '../../../src/agent/context/types'; +import { CronManager } from '../../../src/agent/cron/manager'; +import type { ClockSources } from '../../../src/tools/cron/clock'; +import { + CRON_FIRED, + CRON_MISSED, +} from '../../../src/tools/cron/telemetry-events'; +import type { CronTask } from '../../../src/tools/cron/types'; + +// Stable wall-clock anchor (Nov 14 2023, 22:13:20 UTC) — same anchor as +// `tools/cron/scheduler.test.ts` so cross-file timing assertions stay +// comparable. Picked deliberately off any round minute so the next +// `*/5 * * * *` fire is not exactly five minutes ahead. +const WALL_ANCHOR = 1_700_000_000_000; +const ONE_DAY_MS = 24 * 60 * 60 * 1000; + +interface SteerCall { + readonly content: readonly ContentPart[]; + readonly origin: PromptOrigin; +} +interface TelemetryCall { + readonly event: string; + readonly props: unknown; +} + +interface AgentStub { + readonly agent: Agent; + readonly steerCalls: SteerCall[]; + readonly telemetryCalls: TelemetryCall[]; + setHasActiveTurn(v: boolean): void; +} + +interface AgentStubOptions { + readonly hasActiveTurn?: boolean; + readonly steerReturns?: number | null; +} + +function createAgentStub(opts: AgentStubOptions = {}): AgentStub { + const steerCalls: SteerCall[] = []; + const telemetryCalls: TelemetryCall[] = []; + let hasActiveTurn = opts.hasActiveTurn ?? false; + // Distinguish "not specified" (default 42) from "explicitly null" + // (buffered). `?? 42` would collapse both into 42 because the + // nullish-coalescing operator treats `null` as missing. + const steerReturns: number | null = + 'steerReturns' in opts ? (opts.steerReturns as number | null) : 42; + + const turn = { + get hasActiveTurn(): boolean { + return hasActiveTurn; + }, + steer: (content: readonly ContentPart[], origin: PromptOrigin) => { + steerCalls.push({ content, origin }); + return steerReturns; + }, + }; + const telemetry = { + track: (event: string, props: unknown) => { + telemetryCalls.push({ event, props }); + }, + }; + const agent = { turn, telemetry } as unknown as Agent; + return { + agent, + steerCalls, + telemetryCalls, + setHasActiveTurn: (v: boolean) => { + hasActiveTurn = v; + }, + }; +} + +/** + * Build a frozen-clock harness. `now` is mutable via the returned + * `advance` helper so individual cases can age tasks past the 7-day + * stale threshold without sleeping. + */ +function createClocks(initial = WALL_ANCHOR): { + clocks: ClockSources; + setNow(v: number): void; + advance(ms: number): void; + now(): number; +} { + let wall = initial; + let mono = 1_000_000; + return { + clocks: { + wallNow: () => wall, + monoNowMs: () => mono, + }, + setNow: (v) => { + wall = v; + }, + advance: (ms) => { + wall += ms; + mono += ms; + }, + now: () => wall, + }; +} + +describe('CronManager', () => { + beforeEach(() => { + // Pin jitter off so fire-count assertions are deterministic. Each + // test that actually exercises fires resets the env via stubEnv, + // but setting it here as well shields the construction-path tests + // from any leaked state. + vi.stubEnv('KIMI_CRON_NO_JITTER', '1'); + }); + + afterEach(() => { + vi.unstubAllEnvs(); + }); + + describe('construction', () => { + it('does not throw with default clocks and supports start/stop', async () => { + const { agent } = createAgentStub(); + // Disable the auto-tick timer so the test doesn't have to wait + // for setInterval / clean it up; we just want start() and stop() + // to be wired and idempotent. + const manager = new CronManager(agent, { pollIntervalMs: null }); + expect(() => manager.start()).not.toThrow(); + expect(() => manager.start()).not.toThrow(); // idempotent + await expect(manager.stop()).resolves.toBeUndefined(); + await expect(manager.stop()).resolves.toBeUndefined(); + }); + + it('exposes the session store as an empty list on construction', () => { + const { agent } = createAgentStub(); + const manager = new CronManager(agent, { pollIntervalMs: null }); + expect(manager.store.list()).toEqual([]); + expect(manager.getNextFireTime()).toBeNull(); + }); + }); + + describe('handleFire — recurring', () => { + it('steers with cron_job origin and emits cron_fired telemetry', () => { + const stub = createAgentStub({ steerReturns: 7 }); + const harness = createClocks(); + const manager = new CronManager(stub.agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + + manager.store.add( + { cron: '*/5 * * * *', prompt: 'check the deploy' }, + harness.now() - 1, + ); + // `*/5 * * * *` lands every 5 minutes; bump 6 minutes so we are + // safely past exactly one ideal fire. + harness.advance(6 * 60_000); + manager.tick(); + + expect(stub.steerCalls.length).toBe(1); + const call = stub.steerCalls[0]!; + expect(call.origin.kind).toBe('cron_job'); + if (call.origin.kind !== 'cron_job') throw new Error('unreachable'); + expect(call.origin.recurring).toBe(true); + expect(call.origin.stale).toBe(false); + expect(call.origin.coalescedCount).toBeGreaterThanOrEqual(1); + expect(call.origin.cron).toBe('*/5 * * * *'); + expect(call.origin.jobId).toMatch(/^[0-9a-f]{8}$/); + expect(call.content).toEqual([{ type: 'text', text: 'check the deploy' }]); + + expect(stub.telemetryCalls.length).toBe(1); + const tc = stub.telemetryCalls[0]!; + expect(tc.event).toBe(CRON_FIRED); + expect(tc.props).toMatchObject({ + recurring: true, + durable: false, + stale: false, + buffered: false, + }); + }); + }); + + describe('handleFire — one-shot', () => { + it('uses recurring=false in origin and telemetry', () => { + const stub = createAgentStub(); + const harness = createClocks(); + const manager = new CronManager(stub.agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + + // Add a one-shot task that fires at the very next */5 mark, then + // advance the wall clock past it. + const task = manager.store.add( + { + cron: '*/5 * * * *', + prompt: 'one-shot ping', + recurring: false, + }, + harness.now() - 1, + ); + harness.advance(6 * 60_000); + manager.tick(); + + expect(stub.steerCalls.length).toBe(1); + const origin = stub.steerCalls[0]!.origin; + expect(origin.kind).toBe('cron_job'); + if (origin.kind !== 'cron_job') throw new Error('unreachable'); + expect(origin.recurring).toBe(false); + expect(origin.stale).toBe(false); + + const tc = stub.telemetryCalls[0]!; + expect(tc.props).toMatchObject({ recurring: false }); + + // One-shot was removed from the store after fire. + expect(manager.store.get(task.id)).toBeUndefined(); + }); + }); + + describe('isStale', () => { + it('flags recurring tasks older than 7 days as stale', () => { + const { agent } = createAgentStub(); + const harness = createClocks(); + const manager = new CronManager(agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + + const task: CronTask = { + id: 'deadbeef', + cron: '0 9 * * *', + prompt: 'morning report', + createdAt: harness.now() - 8 * ONE_DAY_MS, + recurring: true, + }; + expect(manager.isStale(task)).toBe(true); + }); + + it('does not flag recurring tasks younger than 7 days', () => { + const { agent } = createAgentStub(); + const harness = createClocks(); + const manager = new CronManager(agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + const task: CronTask = { + id: 'deadbeef', + cron: '0 9 * * *', + prompt: 'morning report', + createdAt: harness.now() - 6 * ONE_DAY_MS, + recurring: true, + }; + expect(manager.isStale(task)).toBe(false); + }); + + it('treats undefined recurring as recurring for stale purposes', () => { + const { agent } = createAgentStub(); + const harness = createClocks(); + const manager = new CronManager(agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + const task: CronTask = { + id: 'deadbeef', + cron: '0 9 * * *', + prompt: 'morning report', + createdAt: harness.now() - 8 * ONE_DAY_MS, + // recurring intentionally omitted + }; + expect(manager.isStale(task)).toBe(true); + }); + + it('one-shot tasks are never stale even if old', () => { + const { agent } = createAgentStub(); + const harness = createClocks(); + const manager = new CronManager(agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + const task: CronTask = { + id: 'deadbeef', + cron: '0 9 * * *', + prompt: 'morning report', + createdAt: harness.now() - 8 * ONE_DAY_MS, + recurring: false, + }; + expect(manager.isStale(task)).toBe(false); + }); + + it('KIMI_CRON_NO_STALE=1 disables stale judgment for recurring', () => { + vi.stubEnv('KIMI_CRON_NO_STALE', '1'); + const { agent } = createAgentStub(); + const harness = createClocks(); + const manager = new CronManager(agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + const task: CronTask = { + id: 'deadbeef', + cron: '0 9 * * *', + prompt: 'morning report', + createdAt: harness.now() - 8 * ONE_DAY_MS, + recurring: true, + }; + expect(manager.isStale(task)).toBe(false); + }); + + it('non-finite age (broken clock) is treated as not stale', () => { + const { agent } = createAgentStub(); + const brokenClocks: ClockSources = { + wallNow: () => Number.NaN, + monoNowMs: () => 0, + }; + const manager = new CronManager(agent, { + clocks: brokenClocks, + pollIntervalMs: null, + }); + const task: CronTask = { + id: 'deadbeef', + cron: '0 9 * * *', + prompt: 'morning report', + createdAt: 0, + recurring: true, + }; + expect(manager.isStale(task)).toBe(false); + }); + }); + + describe('stale propagation into fire origin', () => { + it('origin.stale === true for a recurring task older than 7 days', () => { + const stub = createAgentStub(); + const harness = createClocks(); + const manager = new CronManager(stub.agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + + // Add a recurring task whose createdAt is 8 days ago. Note: the + // scheduler uses createdAt as the starting baseline for next-fire + // computation, so a task that's been "alive" for 8 days will be + // very overdue and will coalesce a lot of fires into one. That's + // fine for this test — we only assert on `stale` (which is + // computed from createdAt vs now) and `coalescedCount >= 1`. + manager.store.add( + { cron: '0 9 * * *', prompt: 'morning report', recurring: true }, + harness.now() - 8 * ONE_DAY_MS, + ); + manager.tick(); + + expect(stub.steerCalls.length).toBe(1); + const origin = stub.steerCalls[0]!.origin; + if (origin.kind !== 'cron_job') throw new Error('expected cron_job'); + expect(origin.stale).toBe(true); + expect(stub.telemetryCalls[0]!.props).toMatchObject({ stale: true }); + }); + }); + + describe('buffered semantics', () => { + it('reports buffered=true on the telemetry event when steer returns null', () => { + const stub = createAgentStub({ steerReturns: null }); + const harness = createClocks(); + const manager = new CronManager(stub.agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + manager.store.add( + { cron: '*/5 * * * *', prompt: 'while-active' }, + harness.now() - 1, + ); + harness.advance(6 * 60_000); + manager.tick(); + + expect(stub.telemetryCalls.length).toBe(1); + expect(stub.telemetryCalls[0]!.props).toMatchObject({ buffered: true }); + }); + }); + + describe('idle gating', () => { + it('does not fire while a turn is active', () => { + const stub = createAgentStub({ hasActiveTurn: true }); + const harness = createClocks(); + const manager = new CronManager(stub.agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + manager.store.add( + { cron: '*/5 * * * *', prompt: 'ping' }, + harness.now() - 1, + ); + harness.advance(6 * 60_000); + manager.tick(); + expect(stub.steerCalls.length).toBe(0); + expect(stub.telemetryCalls.length).toBe(0); + + // Flip back to idle and the next tick fires. + stub.setHasActiveTurn(false); + manager.tick(); + expect(stub.steerCalls.length).toBe(1); + }); + }); + + describe('end-to-end via scheduler', () => { + it('fires once with coalescedCount=1 after a 6-minute gap on */5', () => { + const stub = createAgentStub(); + const harness = createClocks(); + const manager = new CronManager(stub.agent, { + clocks: harness.clocks, + pollIntervalMs: null, + }); + manager.store.add( + { cron: '*/5 * * * *', prompt: 'every five' }, + harness.now() - 1, + ); + // Six minutes past the anchor — exactly one ideal fire in the gap. + harness.advance(6 * 60_000); + manager.tick(); + + expect(stub.steerCalls.length).toBe(1); + const origin = stub.steerCalls[0]!.origin; + if (origin.kind !== 'cron_job') throw new Error('expected cron_job'); + expect(origin.coalescedCount).toBe(1); + }); + }); + + describe('handleMissed', () => { + it('no-ops on an empty task list', () => { + const stub = createAgentStub(); + const manager = new CronManager(stub.agent, { pollIntervalMs: null }); + manager.handleMissed([], () => [{ type: 'text', text: 'should not run' }]); + expect(stub.steerCalls.length).toBe(0); + expect(stub.telemetryCalls.length).toBe(0); + }); + + it('steers cron_missed origin and emits cron_missed telemetry', () => { + const stub = createAgentStub(); + const manager = new CronManager(stub.agent, { pollIntervalMs: null }); + + const tasks: CronTask[] = [ + { + id: '11111111', + cron: '0 9 * * *', + prompt: 'a', + createdAt: 1, + recurring: false, + }, + { + id: '22222222', + cron: '0 10 * * *', + prompt: 'b', + createdAt: 2, + recurring: false, + }, + ]; + const rendered: ContentPart[] = [ + { type: 'text', text: 'You missed 2 one-shot tasks.' }, + ]; + manager.handleMissed(tasks, () => rendered); + + expect(stub.steerCalls.length).toBe(1); + const call = stub.steerCalls[0]!; + expect(call.content).toBe(rendered); + expect(call.origin.kind).toBe('cron_missed'); + if (call.origin.kind !== 'cron_missed') throw new Error('unreachable'); + expect(call.origin.count).toBe(2); + + expect(stub.telemetryCalls.length).toBe(1); + expect(stub.telemetryCalls[0]!.event).toBe(CRON_MISSED); + expect(stub.telemetryCalls[0]!.props).toEqual({ count: 2 }); + }); + }); +}); From fbb6fb6b7638c7ac1fa873054f910344ee46929c Mon Sep 17 00:00:00 2001 From: "haozhe.yang" Date: Wed, 27 May 2026 19:50:21 +0800 Subject: [PATCH 10/26] feat(agent-core): add CronCreate tool (session-only path) CronCreate validates the expression, enforces the 5-year fire window (blocking '0 0 31 2 *' typos), caps prompt bytes at 8KB, caps active jobs at 50 per session, and rejects durable=true until Phase 2 adds the file-backed store. Manager exposes emitScheduled/emitDeleted so tools never reach into agent.telemetry directly. --- packages/agent-core/src/agent/cron/manager.ts | 26 ++ .../agent-core/src/tools/cron/cron-create.md | 52 +++ .../agent-core/src/tools/cron/cron-create.ts | 263 ++++++++++++++ .../test/tools/cron/cron-create.test.ts | 327 ++++++++++++++++++ 4 files changed, 668 insertions(+) create mode 100644 packages/agent-core/src/tools/cron/cron-create.md create mode 100644 packages/agent-core/src/tools/cron/cron-create.ts create mode 100644 packages/agent-core/test/tools/cron/cron-create.test.ts diff --git a/packages/agent-core/src/agent/cron/manager.ts b/packages/agent-core/src/agent/cron/manager.ts index a3b4863a60..65c3c081e4 100644 --- a/packages/agent-core/src/agent/cron/manager.ts +++ b/packages/agent-core/src/agent/cron/manager.ts @@ -46,8 +46,10 @@ import { type CronScheduler, } from '../../tools/cron/scheduler'; import { + CRON_DELETED, CRON_FIRED, CRON_MISSED, + CRON_SCHEDULED, } from '../../tools/cron/telemetry-events'; import type { CronTask } from '../../tools/cron/types'; @@ -222,4 +224,28 @@ export class CronManager { this.agent.turn.steer(content, origin); this.agent.telemetry.track(CRON_MISSED, { count: tasks.length }); } + + /** + * Emit `cron_scheduled` for a freshly-added task. Called by + * `CronCreate` after a successful `store.add(...)`. Kept as an + * explicit method so the tool layer never reaches into + * `manager.agent.telemetry` — preserves the "tools see the manager, + * the manager sees the agent" layering and matches the symmetric + * `emitDeleted` used by `CronDelete` (P1.6). + */ + emitScheduled(task: CronTask): void { + this.agent.telemetry.track(CRON_SCHEDULED, { + recurring: task.recurring !== false, + durable: task.durable === true, + }); + } + + /** + * Emit `cron_deleted` for a removed task. Wired up here so P1.6 can + * land without touching this file again. `task_id` matches the field + * naming used elsewhere in the telemetry surface (snake_case). + */ + emitDeleted(taskId: string): void { + this.agent.telemetry.track(CRON_DELETED, { task_id: taskId }); + } } diff --git a/packages/agent-core/src/tools/cron/cron-create.md b/packages/agent-core/src/tools/cron/cron-create.md new file mode 100644 index 0000000000..24aea02841 --- /dev/null +++ b/packages/agent-core/src/tools/cron/cron-create.md @@ -0,0 +1,52 @@ +Schedule a prompt to be enqueued at a future time. Use for both recurring schedules and one-shot reminders. + +Uses standard 5-field cron in the user's local timezone: minute hour day-of-month month day-of-week. `0 9 * * *` means 9am local — no timezone conversion needed. + +## One-shot tasks (recurring: false) + +For "remind me at X" or "at