diff --git a/.changeset/cron-tools.md b/.changeset/cron-tools.md new file mode 100644 index 0000000000..98576443d6 --- /dev/null +++ b/.changeset/cron-tools.md @@ -0,0 +1,6 @@ +--- +"@moonshot-ai/agent-core": minor +"@moonshot-ai/kimi-code": minor +--- + +Add session-scoped scheduled prompts so the assistant can register, list, and cancel recurring or one-shot reminders that fire later in the same session. diff --git a/apps/kimi-code/src/tui/controllers/session-replay.ts b/apps/kimi-code/src/tui/controllers/session-replay.ts index 106b384bf0..f806cd8145 100644 --- a/apps/kimi-code/src/tui/controllers/session-replay.ts +++ b/apps/kimi-code/src/tui/controllers/session-replay.ts @@ -203,6 +203,12 @@ export class SessionReplayRenderer { if (message.origin?.kind === 'injection') { return; } + // WHY: cron fires are not user turns (see isReplayUserTurnRecord); skip + // visual render and turn advance so the raw envelope never + // surfaces in the resumed transcript. + if (message.origin?.kind === 'cron_job' || message.origin?.kind === 'cron_missed') { + return; + } this.flushAssistant(context); const skill = skillActivationFromOrigin(message.origin); diff --git a/apps/kimi-code/src/tui/utils/message-replay.ts b/apps/kimi-code/src/tui/utils/message-replay.ts index 510b5769e4..8b83186dd7 100644 --- a/apps/kimi-code/src/tui/utils/message-replay.ts +++ b/apps/kimi-code/src/tui/utils/message-replay.ts @@ -246,6 +246,8 @@ function isReplayUserTurnRecord(record: AgentReplayRecord): boolean { return message.origin.trigger === 'user-slash'; case 'background_task': case 'compaction_summary': + case 'cron_job': + case 'cron_missed': case 'hook_result': case 'injection': case 'system_trigger': diff --git a/apps/kimi-code/test/tui/message-replay.test.ts b/apps/kimi-code/test/tui/message-replay.test.ts index c4a64fc99c..905fae2adc 100644 --- a/apps/kimi-code/test/tui/message-replay.test.ts +++ b/apps/kimi-code/test/tui/message-replay.test.ts @@ -371,6 +371,54 @@ describe('KimiTUI resume message replay', () => { ]); }); + it('skips cron_job origin records during replay', async () => { + const cronFire = + '\nrun nightly\n'; + const driver = await replayIntoDriver([ + message('user', [{ type: 'text', text: 'real prompt' }]), + message('assistant', [{ type: 'text', text: 'real answer' }]), + message('user', [{ type: 'text', text: cronFire }], { + origin: { + kind: 'cron_job', + jobId: 'job-1', + cron: '*/5 * * * *', + recurring: true, + coalescedCount: 1, + stale: false, + }, + }), + ]); + + const transcript = driver.state.transcriptContainer.render(120).join('\n'); + expect(transcript).not.toContain(' entry.kind === 'user') + .map((entry) => entry.content), + ).toEqual(['real prompt']); + }); + + it('skips cron_missed origin records during replay', async () => { + const cronMissed = + '\n3 one-shot tasks missed while offline\n'; + const driver = await replayIntoDriver([ + message('user', [{ type: 'text', text: 'real prompt' }]), + message('assistant', [{ type: 'text', text: 'real answer' }]), + message('user', [{ type: 'text', text: cronMissed }], { + origin: { kind: 'cron_missed', count: 3 }, + }), + ]); + + const transcript = driver.state.transcriptContainer.render(120).join('\n'); + expect(transcript).not.toContain(' entry.kind === 'user') + .map((entry) => entry.content), + ).toEqual(['real prompt']); + }); + it('renders user-slash skill activation once without exposing injected prompt text', async () => { const activation = message( 'user', diff --git a/packages/agent-core/src/agent/context/projector.ts b/packages/agent-core/src/agent/context/projector.ts index 3b5ac10a7a..d085a791fe 100644 --- a/packages/agent-core/src/agent/context/projector.ts +++ b/packages/agent-core/src/agent/context/projector.ts @@ -115,6 +115,7 @@ function isInjectionUserMessage(message: Message): boolean { if (trimmed.startsWith('')) return true; if (trimmed.startsWith('= 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/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..005a2f8caa --- /dev/null +++ b/packages/agent-core/src/agent/cron/manager.ts @@ -0,0 +1,386 @@ +/** + * 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 { renderCronFireXml } from '../../tools/cron/cron-fire-xml'; +import { SessionCronStore } from '../../tools/cron/session-store'; +import { + createCronScheduler, + 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'; + +/** + * 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=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=1`, which forces `null` here + * so the auto-tick `setInterval` is never installed). `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; + /** + * Tracks whether `start()` has been called without a matching `stop()`. + * Used to keep `start()` / `stop()` idempotent and — more importantly + * for P1.8 — to gate SIGUSR1 binding so we don't accumulate handlers + * across repeated start() calls. + */ + private started = false; + /** + * Reference to the bound SIGUSR1 listener while the manager is + * running. Held so `stop()` can call `process.off('SIGUSR1', handler)` + * with the same function reference and not leak handlers across vitest + * files. `null` whenever the manager is not started, or when running + * on a platform that does not support SIGUSR1 (Windows). + */ + private sigusr1Handler: NodeJS.SignalsListener | null = null; + + 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]); + }, + // P1.8: `KIMI_CRON_MANUAL_TICK=1` forces the scheduler into + // manual-drive mode (no setInterval), so bench / time-injected + // tests can step time forward and call `tick()` explicitly without + // racing a 1-second auto-tick. Explicit caller overrides + // (`opts.pollIntervalMs`) lose to the env so a bench can flip the + // switch from the outside without rebuilding the manager wiring. + pollIntervalMs: + process.env['KIMI_CRON_MANUAL_TICK'] === '1' + ? null + : opts.pollIntervalMs, + }); + } + + /** + * Begin the scheduler's auto-tick loop and bind the SIGUSR1 manual-tick + * hook (P1.8). Idempotent: a second call is a no-op so the boot + * sequence and tests can opt into "ensure started" without bookkeeping. + */ + start(): void { + if (this.started) return; + this.started = true; + this.scheduler.start(); + this.bindSigusr1(); + } + + /** + * Stop the scheduler, clear in-flight bookkeeping, and unbind the + * SIGUSR1 handler. Idempotent and signal-handler-safe — multiple + * vitest files exercising the manager must not leave a SIGUSR1 listener + * dangling on the shared process. + */ + async stop(): Promise { + this.unbindSigusr1(); + await this.scheduler.stop(); + this.started = false; + } + + /** 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(); + } + + /** + * Per-task post-jitter next-fire. Forwards to the scheduler so + * CronList renders the same instant the scheduler will fire — even + * when an already-past ideal still has a pending jittered delivery + * in the current period. + */ + getNextFireForTask(taskId: string): number | null { + return this.scheduler.getNextFireForTask(taskId); + } + + /** + * 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". + * + * Honours the documented 7-day auto-expire contract for recurring + * tasks: a stale recurring task gets exactly one final delivery + * (already issued above) and is then removed from the store. The + * scheduler picks up the deletion on its next tick via `source()` + * and stops re-firing the task. One-shots are not affected — they + * are deleted by the scheduler immediately after delivery via the + * `removeOneShot` callback. + */ + 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: renderCronFireXml(origin, task.prompt), + }, + ]; + const turnId = this.agent.turn.steer(content, origin); + this.agent.telemetry.track(CRON_FIRED, { + recurring: task.recurring !== false, + coalesced_count: ctx.coalescedCount, + stale, + buffered: turnId === null, + }); + + // 7-day auto-expire — the recurring branch of CronCreate's tool + // description promises this contract to the model. Without the + // removal a long-lived session keeps re-injecting a multi-day-old + // cron prompt forever; with it, the task fires one last time + // (above) and is then dropped. Emit `cron_deleted` symmetrically + // with manual deletion so dashboards see the lifecycle close. + if (stale && task.recurring !== false) { + this.store.remove([task.id]); + this.emitDeleted(task.id); + } + } + + /** + * 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 }); + } + + /** + * 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, + }); + } + + /** + * 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 }); + } + + /** + * Wire `SIGUSR1` to a manual `tick()` so bench scripts can advance the + * scheduler with `kill -USR1 ` without a custom RPC. + * + * Gated on `KIMI_CRON_MANUAL_TICK=1` for two reasons: + * + * 1. SIGUSR1 only makes sense when auto-tick is off. When the 1s + * interval is running, it already advances the scheduler — a + * manual signal is redundant. + * 2. In production a single CLI process can host one main agent plus + * many subagents. Each Agent unconditionally binding a SIGUSR1 + * listener would put us over Node's 10-listener default cap and + * print a `MaxListenersExceededWarning`. Coupling the binding to + * the same env that disables auto-tick keeps the production path + * at zero listeners while still giving benches the affordance. + * + * Skipped on Windows because Node's signal layer does not deliver + * POSIX signals there; attempting to `process.on('SIGUSR1', ...)` is a + * silent no-op but we avoid the call entirely so the bookkeeping + * (`sigusr1Handler !== null` means "we did bind") stays accurate. + * + * Idempotent — repeated calls keep the same listener registered once, + * so `start() → start()` does not stack handlers. + * + * The handler swallows any throw from `tick()` because a signal-driven + * bench tool must never crash the host process; the tick failure mode + * is already surfaced via telemetry / logs inside the scheduler. + * Set `KIMI_CRON_DEBUG=1` to surface the swallowed error to stderr — + * mirrors `scheduler.ts`'s debugLog pattern so bench debugging can + * see a bad tick. + */ + private bindSigusr1(): void { + if (process.platform === 'win32') return; + if (process.env['KIMI_CRON_MANUAL_TICK'] !== '1') return; + if (this.sigusr1Handler !== null) return; + const handler: NodeJS.SignalsListener = () => { + try { + this.tick(); + } catch (error) { + if (process.env['KIMI_CRON_DEBUG'] === '1') { + const msg = error instanceof Error ? error.message : String(error); + process.stderr.write( + `[cron/manager] SIGUSR1 tick threw: ${msg}\n`, + ); + } + } + }; + this.sigusr1Handler = handler; + process.on('SIGUSR1', handler); + } + + /** + * Detach the SIGUSR1 listener registered by `bindSigusr1`. Safe to + * call when nothing is bound (no-op). Pair this with `stop()` so + * vitest files don't leak signal handlers across the shared process — + * `process.listenerCount('SIGUSR1')` should return to its pre-`start()` + * value once `stop()` resolves. + */ + private unbindSigusr1(): void { + if (this.sigusr1Handler === null) return; + process.off('SIGUSR1', this.sigusr1Handler); + this.sigusr1Handler = null; + } +} diff --git a/packages/agent-core/src/agent/index.ts b/packages/agent-core/src/agent/index.ts index 88f95d9097..d4b19f23d5 100644 --- a/packages/agent-core/src/agent/index.ts +++ b/packages/agent-core/src/agent/index.ts @@ -34,6 +34,7 @@ import { import type { PromisableMethods } from '../utils/types'; import { BackgroundManager } from './background'; import { FullCompaction, type CompactionStrategy } from './compaction'; +import { CronManager } from './cron'; import { ConfigState } from './config'; import { ContextMemory } from './context'; import { HookEngine } from './hooks'; @@ -111,6 +112,7 @@ export class Agent { readonly usage: UsageRecorder; readonly tools: ToolManager; readonly background: BackgroundManager; + readonly cron: CronManager; readonly replayBuilder: ReplayBuilder; readonly log: Logger; @@ -161,6 +163,18 @@ export class Agent { maxRunningTasks: config.backgroundMaxRunningTasks, sessionDir: config.backgroundSessionDir, }); + this.cron = new CronManager(this); + if (this.type !== 'sub') { + // Skip auto-tick for subagents: each session can spawn many + // subagents, and stacking 1s setInterval timers + SIGUSR1 + // listeners per subagent serves no purpose — the default subagent + // profiles don't expose Cron tools, so the store stays empty. + // The scheduler unref()'s its setInterval so the cron timer never + // keeps the process alive on its own, and isKilled (reading + // KIMI_DISABLE_CRON) short-circuits every tick — no need to + // delay start when the killswitch is set. + this.cron.start(); + } this.replayBuilder = new ReplayBuilder(this); } diff --git a/packages/agent-core/src/agent/permission/policies/default-tool-approve.ts b/packages/agent-core/src/agent/permission/policies/default-tool-approve.ts index c2944fa36e..7e5a5c2f2c 100644 --- a/packages/agent-core/src/agent/permission/policies/default-tool-approve.ts +++ b/packages/agent-core/src/agent/permission/policies/default-tool-approve.ts @@ -9,6 +9,7 @@ const DEFAULT_APPROVE_TOOLS = new Set([ 'TodoList', 'TaskList', 'TaskOutput', + 'CronList', 'WebSearch', 'FetchURL', 'Agent', diff --git a/packages/agent-core/src/agent/permission/policies/plan-mode-guard-deny.ts b/packages/agent-core/src/agent/permission/policies/plan-mode-guard-deny.ts index b25e294c93..0e02f9e421 100644 --- a/packages/agent-core/src/agent/permission/policies/plan-mode-guard-deny.ts +++ b/packages/agent-core/src/agent/permission/policies/plan-mode-guard-deny.ts @@ -28,12 +28,23 @@ export class PlanModeGuardDenyPermissionPolicy implements PermissionPolicy { }; } - if (toolName !== 'TaskStop') return; - return { - kind: 'deny', - message: - 'TaskStop is not available in plan mode. Call ExitPlanMode to exit plan mode before stopping a background task.', - }; + if (toolName === 'TaskStop') { + return { + kind: 'deny', + message: + 'TaskStop is not available in plan mode. Call ExitPlanMode to exit plan mode before stopping a background task.', + }; + } + + if (toolName === 'CronCreate' || toolName === 'CronDelete') { + return { + kind: 'deny', + message: + `${toolName} is not available in plan mode because it would mutate scheduled work that runs after plan exit. Call ExitPlanMode first.`, + }; + } + + return; } } diff --git a/packages/agent-core/src/agent/tool/index.ts b/packages/agent-core/src/agent/tool/index.ts index df928d43c0..c329ccaf42 100644 --- a/packages/agent-core/src/agent/tool/index.ts +++ b/packages/agent-core/src/agent/tool/index.ts @@ -375,6 +375,9 @@ export class ToolManager { new b.TaskListTool(background), new b.TaskOutputTool(background), new b.TaskStopTool(background), + this.agent.type !== 'sub' && new b.CronCreateTool(this.agent.cron), + this.agent.type !== 'sub' && new b.CronListTool(this.agent.cron), + this.agent.type !== 'sub' && new b.CronDeleteTool(this.agent.cron), this.agent.skills !== undefined && this.agent.skills.registry.listInvocableSkills().length > 0 && new b.SkillTool(this.agent), diff --git a/packages/agent-core/src/profile/default/agent.yaml b/packages/agent-core/src/profile/default/agent.yaml index 0c0490becb..82b81bd3ef 100644 --- a/packages/agent-core/src/profile/default/agent.yaml +++ b/packages/agent-core/src/profile/default/agent.yaml @@ -15,6 +15,9 @@ tools: - TaskList - TaskOutput - TaskStop + - CronCreate + - CronList + - CronDelete - ReadMediaFile - TodoList - Skill diff --git a/packages/agent-core/src/session/index.ts b/packages/agent-core/src/session/index.ts index ad621208ad..9ca223f033 100644 --- a/packages/agent-core/src/session/index.ts +++ b/packages/agent-core/src/session/index.ts @@ -183,6 +183,7 @@ export class Session { async close(): Promise { try { await this.stopBackgroundTasksOnExit(); + await this.stopCronOnExit(); await this.flushMetadata(); await this.triggerSessionEnd('exit'); } finally { @@ -194,6 +195,15 @@ export class Session { } } + // Stop every agent's cron scheduler on close. No keepAlive notion — cron + // intervals reference the agent/session graph and must die with the session. + // `allSettled` keeps one agent's failure from blocking the rest. + private async stopCronOnExit(): Promise { + await Promise.allSettled( + Array.from(this.agents.values(), (agent) => agent.cron.stop()), + ); + } + private async stopBackgroundTasksOnExit(): Promise { const keepAliveOnExit = resolveConfigValue({ env: process.env, diff --git a/packages/agent-core/src/tools/builtin/index.ts b/packages/agent-core/src/tools/builtin/index.ts index 046d475576..ebbe0dc71e 100644 --- a/packages/agent-core/src/tools/builtin/index.ts +++ b/packages/agent-core/src/tools/builtin/index.ts @@ -2,6 +2,9 @@ export * from '../background/manager'; export * from '../background/task-list'; export * from '../background/task-output'; export * from '../background/task-stop'; +export * from '../cron/cron-create'; +export * from '../cron/cron-delete'; +export * from '../cron/cron-list'; export * from './collaboration/agent'; export * from './collaboration/ask-user'; export * from './collaboration/skill-tool'; 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..98b784a87b --- /dev/null +++ b/packages/agent-core/src/tools/cron/clock.ts @@ -0,0 +1,148 @@ +/** + * 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 { closeSync, openSync, readSync } 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} + * `"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('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; +} + +// Epoch-ms is always under 20 characters in practice; 64 bytes leaves +// slack for a leading newline / `\r` and prevents OOM on a hostile or +// accidentally-huge clock file (e.g. a `/dev/zero` redirect). +const MAX_CLOCK_FILE_BYTES = 64; + +function readFileWall(filePath: string): number { + let bytesRead = 0; + const buf = Buffer.alloc(MAX_CLOCK_FILE_BYTES); + let fd: number; + try { + fd = openSync(filePath, 'r'); + } catch { + return Date.now(); + } + try { + bytesRead = readSync(fd, buf, 0, MAX_CLOCK_FILE_BYTES, 0); + } catch { + return Date.now(); + } finally { + try { + closeSync(fd); + } catch { + /* swallow close errors */ + } + } + const raw = buf.subarray(0, bytesRead).toString('utf8'); + 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/src/tools/cron/cron-create.md b/packages/agent-core/src/tools/cron/cron-create.md new file mode 100644 index 0000000000..9b6d6850d9 --- /dev/null +++ b/packages/agent-core/src/tools/cron/cron-create.md @@ -0,0 +1,79 @@ +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