diff --git a/.changeset/cron-tools.md b/.changeset/cron-tools.md new file mode 100644 index 000000000..98576443d --- /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 106b384bf..f806cd814 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/export-markdown.ts b/apps/kimi-code/src/tui/utils/export-markdown.ts index 6eac05322..9531efa10 100644 --- a/apps/kimi-code/src/tui/utils/export-markdown.ts +++ b/apps/kimi-code/src/tui/utils/export-markdown.ts @@ -97,6 +97,12 @@ const INTERNAL_ORIGINS = new Set([ 'system_trigger', 'compaction_summary', 'hook_result', + // Cron fires are stored as user-role records carrying a `` + // XML envelope meant only for the model. Replay and the TUI projector + // already hide them; the markdown exporter must do the same or the raw + // protocol XML leaks into the user-facing export. + 'cron_job', + 'cron_missed', ]); export function isInternalMessage(msg: ContextMessage): boolean { diff --git a/apps/kimi-code/src/tui/utils/message-replay.ts b/apps/kimi-code/src/tui/utils/message-replay.ts index 510b5769e..8b83186dd 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/export-markdown.test.ts b/apps/kimi-code/test/tui/export-markdown.test.ts index 1c59cd6e6..05a6eb8ad 100644 --- a/apps/kimi-code/test/tui/export-markdown.test.ts +++ b/apps/kimi-code/test/tui/export-markdown.test.ts @@ -181,6 +181,27 @@ describe('isInternalMessage', () => { ).toBe(true); }); + it('marks cron_job origin as internal', () => { + expect( + isInternalMessage( + userMsg('x', { + kind: 'cron_job', + jobId: 'a1b2c3d4', + cron: '0 9 * * *', + recurring: true, + coalescedCount: 1, + stale: false, + }), + ), + ).toBe(true); + }); + + it('marks cron_missed origin as internal', () => { + expect( + isInternalMessage(userMsg('x', { kind: 'cron_missed', count: 2 })), + ).toBe(true); + }); + it('keeps real user messages', () => { expect(isInternalMessage(userMsg('hello', { kind: 'user' }))).toBe(false); }); diff --git a/apps/kimi-code/test/tui/message-replay.test.ts b/apps/kimi-code/test/tui/message-replay.test.ts index e9b347c27..6ca45971b 100644 --- a/apps/kimi-code/test/tui/message-replay.test.ts +++ b/apps/kimi-code/test/tui/message-replay.test.ts @@ -372,6 +372,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 3b5ac10a7..d085a791f 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 000000000..fca7804e8 --- /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 000000000..058ccba8c --- /dev/null +++ b/packages/agent-core/src/agent/cron/manager.ts @@ -0,0 +1,566 @@ +/** + * CronManager — Agent-facing facade for the 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; + * - mirror every store mutation to `/cron/.json` + * (via {@link addTask} / {@link removeTasks}) so that `kimi resume` + * can call {@link loadFromDisk} to rehydrate previously-scheduled + * tasks. When no `sessionDir` is supplied (subagents, tests, + * ephemeral sessions) the manager stays purely in-memory. + * - provide a `handleMissed(...)` entry point that future boot-time + * missed-task notification will call. Today the scheduler's + * `coalescedCount` semantics handle missed fires inline, so this + * entry point is not wired by the framework — it stays exposed so + * adding a banner later does not require API churn here. + * + * 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 { createCronPersistStore } from '../../tools/cron/persist'; +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'; +import type { PerIdJsonStore } from '../../utils/per-id-json-store'; + +import type { SessionCronTaskInit } from '../../tools/cron/session-store'; + +/** + * 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; + + /** + * Per-session directory used to persist cron tasks across + * `kimi resume`. When set, `addTask` / `removeTasks` mirror the + * in-memory store to `/cron/.json` and + * `loadFromDisk()` re-populates the store on resume. When omitted + * (subagents, tests, ephemeral sessions), the manager stays purely + * in-memory and `loadFromDisk()` is a no-op. + */ + readonly sessionDir?: string; +} + +export class CronManager { + /** In-memory task store. Empty at construction; populated by + * {@link addTask} (and {@link loadFromDisk} on resume). */ + 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; + + /** + * File-backed mirror of {@link store}. `undefined` when no + * `sessionDir` was supplied — the manager then behaves as pure + * in-memory, matching pre-persistence semantics. When defined, + * `addTask` / `removeTasks` schedule fire-and-forget writes so a + * later `kimi resume` can reload via {@link loadFromDisk}. + */ + private readonly persistStore: PerIdJsonStore | undefined; + + /** + * Per-id serializer for persistence writes. Prevents a fast + * `add` → `remove` sequence on the same id from racing each other on + * the rename — the rm must observe the prior write's renamed file. + * Empty between bursts; entries are deleted once their tail promise + * settles so the map cannot grow unboundedly with churn. + */ + private readonly persistQueues: Map> = new Map(); + + 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.persistStore = + opts.sessionDir === undefined + ? undefined + : createCronPersistStore(opts.sessionDir); + + 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.removeTasks([id]); + }, + onAdvanceCursor: (id, lastFiredAt) => { + this.advanceCursor(id, lastFiredAt); + }, + // 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, + }); + } + + /** + * Add a fresh task to the in-memory store and, when persistence is + * attached, mirror the new record to `/cron/.json`. + * + * The store call is synchronous (CronCreate needs the id for its + * response); the on-disk write is fire-and-forget so a slow disk + * never blocks the tool's reply. Per-id queueing serializes + * concurrent writes on the same id (e.g. add → stale auto-expire) so + * the rm cannot race the rename. + * + * Persistence failures are logged via `agent.log.warn` and swallowed + * — a flaky disk drops cross-resume durability but must not crash + * the agent loop. + */ + addTask(init: SessionCronTaskInit): CronTask { + const task = this.store.add(init, this.clocks.wallNow()); + this.persistEnqueue(task.id, () => + this.persistStore!.write(task.id, task), + ); + return task; + } + + /** + * Remove a batch of tasks from the in-memory store and mirror each + * deletion to disk (when persistence is attached). Returns the + * subset of ids that were actually present, matching + * `SessionCronStore.remove`'s contract — callers (CronDelete / + * scheduler one-shot cleanup / stale auto-expire) read this to + * decide whether to emit telemetry. + * + * Persistence failures are logged and swallowed; cross-resume the + * worst case is a ghost entry that gets dropped on the next + * `list()` shape-guard pass. + */ + removeTasks(ids: readonly string[]): readonly string[] { + const removed = this.store.remove(ids); + for (const id of removed) { + this.persistEnqueue(id, () => this.persistStore!.remove(id)); + } + return removed; + } + + /** + * Persist the scheduler's `lastFiredAt` cursor for a recurring task + * so a `kimi resume` does not coalesce-replay an already-delivered + * fire. Called by the scheduler's `onAdvanceCursor` callback after a + * successful recurring fire. + * + * No-op when the task has already been removed between fire and + * callback (concurrent CronDelete is the canonical case). When + * persistence is detached (subagent / ephemeral session) we still + * update the in-memory record — same-session stale checks read off + * the in-memory store. The on-disk write is fire-and-forget via + * `persistEnqueue`; a flaky disk drops cross-resume durability but + * never blocks the scheduler. + */ + private advanceCursor(id: string, lastFiredAt: number): void { + const updated = this.store.markFired(id, lastFiredAt); + if (updated === undefined) return; + if (this.persistStore === undefined) return; + this.persistEnqueue(id, () => this.persistStore!.write(id, updated)); + } + + /** + * Rehydrate the in-memory store from `/cron/` after + * `kimi resume`. No-op when persistence is not attached. Idempotent: + * clears the in-memory map and re-inserts every record on disk. + * + * Tasks are inserted via {@link SessionCronStore.adopt} so the + * original `id` and `createdAt` survive — `createdAt` is the + * scheduler's recurring baseline and the 7-day stale judgment's + * input, so a regenerated value would corrupt both. + */ + async loadFromDisk(): Promise { + if (this.persistStore === undefined) return; + const tasks = await this.persistStore.list(); + this.store.clear(); + for (const task of tasks) { + this.store.adopt(task); + } + } + + /** + * Serialize per-id persistence writes. Concurrent mutations on the + * same id (uncommon but reachable via `add` immediately followed by + * stale auto-expire) would otherwise race on the rename — atomicWrite + * is per-call atomic, not per-id ordered. Each id's chain is dropped + * from the map once it settles so the map size tracks live in-flight + * writes, not lifetime churn. + */ + private persistEnqueue(id: string, work: () => Promise): void { + if (this.persistStore === undefined) return; + const prev = this.persistQueues.get(id) ?? Promise.resolve(); + const next = prev + .catch(() => {}) + .then(() => work()) + .catch((err: unknown) => { + this.agent.log?.warn?.('cron persist failed', err); + }) + .finally(() => { + if (this.persistQueues.get(id) === next) { + this.persistQueues.delete(id); + } + }); + this.persistQueues.set(id, next); + } + + /** + * Wait for every pending persistence write / remove scheduled via + * {@link addTask} / {@link removeTasks} to settle. Called from + * {@link stop} for graceful session shutdown and exposed publicly so + * tests can synchronise on disk-visible state without polling. + * + * Errors are already swallowed by `persistEnqueue`, so this never + * rejects. + */ + async flushPersist(): Promise { + // Snapshot the chain promises rather than the map itself — the + // `.finally` cleanup deletes entries while we await, and a live + // map iteration would observe the deletions and miss tails. + const inFlight = Array.from(this.persistQueues.values()); + await Promise.allSettled(inFlight); + } + + /** + * 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, drain pending persistence writes, 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. + * + * Draining persistence on shutdown matters for production: a session + * `close()` immediately after a CronCreate would otherwise tear the + * process down before the JSON file lands on disk, and the task + * would be missing from the resume's `loadFromDisk()`. + */ + async stop(): Promise { + this.unbindSigusr1(); + await this.scheduler.stop(); + await this.flushPersist(); + 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.removeTasks([task.id]); + this.emitDeleted(task.id); + } + } + + /** + * Reserved hook for an explicit "you missed N fires while offline" + * banner. Today the scheduler's `coalescedCount` semantics already + * communicate missed fires inside the `cron_job` envelope (and + * recurring tasks past 7 days arrive with `stale: true`), so the + * resume path does NOT invoke this from the framework. The method + * stays exposed because adding a separate user-facing banner later + * — e.g. for one-shots whose fire times all landed during a long + * outage — should not require an API change here. + * + * 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 0c01d7c83..30f283c49 100644 --- a/packages/agent-core/src/agent/index.ts +++ b/packages/agent-core/src/agent/index.ts @@ -29,6 +29,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'; @@ -74,6 +75,13 @@ export interface AgentConfig { readonly hookEngine?: HookEngine; readonly backgroundMaxRunningTasks?: number; readonly backgroundSessionDir?: string; + /** + * Per-session directory used by `CronManager` to persist scheduled + * tasks across `kimi resume`. When omitted the cron stack stays + * purely in-memory (subagents, ephemeral sessions). Set in parallel + * with {@link backgroundSessionDir} from the session homedir. + */ + readonly cronSessionDir?: string; readonly permission?: PermissionManagerOptions | undefined; readonly log?: Logger; readonly telemetry?: TelemetryClient | undefined; @@ -106,6 +114,7 @@ export class Agent { readonly usage: UsageRecorder; readonly tools: ToolManager; readonly background: BackgroundManager; + readonly cron: CronManager; readonly replayBuilder: ReplayBuilder; readonly log: Logger; @@ -157,6 +166,25 @@ export class Agent { maxRunningTasks: config.backgroundMaxRunningTasks, sessionDir: config.backgroundSessionDir, }); + this.cron = new CronManager(this, { + // Subagents stay in-memory: their default profiles don't expose + // cron tools, so attaching a sessionDir would only litter the + // subagent homedir with an empty `cron/` directory on first + // mkdir. Main / non-sub agents persist when the session supplied + // a directory. + sessionDir: this.type !== 'sub' ? config.cronSessionDir : undefined, + }); + 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); } @@ -254,6 +282,12 @@ export class Agent { const result = await this.records.replay(); await this.background.loadFromDisk(); await this.background.reconcile(); + // Rehydrate cron tasks scheduled in the previous CLI invocation. + // No-op when this agent was constructed without a `cronSessionDir` + // (subagents, ephemeral sessions). The scheduler's `createdAt`-based + // baseline then handles any fire times missed during downtime via + // `coalescedCount` (and the 7-day stale flag) without further wiring. + await this.cron.loadFromDisk(); this.turn.finishResume(); return result; } 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 c2944fa36..7e5a5c2f2 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 b25e294c9..0e02f9e42 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 13ecdadbb..4a0bb925f 100644 --- a/packages/agent-core/src/agent/tool/index.ts +++ b/packages/agent-core/src/agent/tool/index.ts @@ -374,6 +374,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 0c0490bec..82b81bd3e 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 379e83d15..2e2751a09 100644 --- a/packages/agent-core/src/session/index.ts +++ b/packages/agent-core/src/session/index.ts @@ -181,6 +181,13 @@ export class Session { async close(): Promise { try { + // Stop cron FIRST. `stopBackgroundTasksOnExit()` can await + // long-running background workers (especially with + // `background.keepAliveOnExit=false`); while we are waiting, a due + // cron tick would otherwise still call `turn.steer()` and start a + // fresh turn after shutdown has already begun, racing the + // metadata flush below. + await this.stopCronOnExit(); await this.stopBackgroundTasksOnExit(); await this.flushMetadata(); await this.triggerSessionEnd('exit'); @@ -193,6 +200,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, @@ -415,6 +431,7 @@ export class Session { mcp: this.mcp, backgroundMaxRunningTasks: this.config.background?.maxRunningTasks, backgroundSessionDir: homedir, + cronSessionDir: homedir, permission: this.permissionOptions(parentAgentId, config.permission), telemetry: this.telemetry, log: this.log.createChild({ agentId: id }), diff --git a/packages/agent-core/src/tools/background/persist.ts b/packages/agent-core/src/tools/background/persist.ts index a6f95aa66..91725868a 100644 --- a/packages/agent-core/src/tools/background/persist.ts +++ b/packages/agent-core/src/tools/background/persist.ts @@ -5,15 +5,19 @@ * restart can list previously-running tasks (now lost) and emit terminal * notifications. * - * Writes use `atomicWrite` (write-tmp-fsync-rename) so a crash mid-write - * never leaves a half-truncated file. + * The per-id JSON layer (write / read / list / remove) is delegated to + * `createPerIdJsonStore`, which centralises atomic-write + + * path-traversal-guarded readdir for cron / background / anything else + * that needs session-scoped per-id JSON. This module keeps the + * background-specific shape, the output.log helpers, and the named + * exports (`writeTask`, …) the rest of `background/` already imports. */ import { statSync } from 'node:fs'; -import { appendFile, mkdir, open, readFile, readdir, rm, stat, unlink } from 'node:fs/promises'; +import { appendFile, mkdir, open, readFile, rm, stat } from 'node:fs/promises'; import { dirname, join } from 'pathe'; -import { atomicWrite } from '../../utils/fs'; +import { createPerIdJsonStore, type PerIdJsonStore } from '../../utils/per-id-json-store'; import type { BackgroundTaskStatus } from './manager'; /** @@ -67,13 +71,6 @@ function tasksDirOf(sessionDir: string): string { return join(sessionDir, 'tasks'); } -function taskFile(sessionDir: string, taskId: string): string { - if (!VALID_TASK_ID.test(taskId)) { - throw new Error(`Invalid task id: "${taskId}"`); - } - return join(tasksDirOf(sessionDir), `${taskId}.json`); -} - function taskOutputDir(sessionDir: string, taskId: string): string { if (!VALID_TASK_ID.test(taskId)) { throw new Error(`Invalid task id: "${taskId}"`); @@ -85,11 +82,33 @@ export function taskOutputFile(sessionDir: string, taskId: string): string { return join(taskOutputDir(sessionDir, taskId), 'output.log'); } +/** + * Cache of `createPerIdJsonStore` instances keyed by sessionDir. + * + * Per-id stores hold no state beyond their options object, so reusing + * the same instance across calls into this module is purely an allocation + * micro-optimisation. The cache is unbounded; the number of distinct + * session directories per process is small (typically 1) and the + * lifetime matches the process. + */ +const storeCache = new Map>(); +function storeFor(sessionDir: string): PerIdJsonStore { + const cached = storeCache.get(sessionDir); + if (cached !== undefined) return cached; + const store = createPerIdJsonStore({ + rootDir: sessionDir, + subdir: 'tasks', + idRegex: VALID_TASK_ID, + isValid: isValidPersistedTask, + entityName: 'task id', + }); + storeCache.set(sessionDir, store); + return store; +} + /** Atomically write a task's persisted state. Creates dirs as needed. */ export async function writeTask(sessionDir: string, task: PersistedTask): Promise { - await mkdir(tasksDirOf(sessionDir), { recursive: true, mode: 0o700 }); - const target = taskFile(sessionDir, task.task_id); - await atomicWrite(target, JSON.stringify(task, null, 2)); + await storeFor(sessionDir).write(task.task_id, task); } /** Read a single task file. Returns undefined when missing/corrupt. */ @@ -97,22 +116,7 @@ export async function readTask( sessionDir: string, taskId: string, ): Promise { - // Path-traversal validation runs before the try/catch so callers see - // an explicit error instead of a misleading "missing" return. - const path = taskFile(sessionDir, taskId); - let raw: string; - try { - raw = await readFile(path, 'utf-8'); - } catch { - return undefined; - } - try { - const parsed = JSON.parse(raw) as Record; - if (typeof parsed !== 'object' || parsed === null) return undefined; - return parsed as unknown as PersistedTask; - } catch { - return undefined; - } + return storeFor(sessionDir).read(taskId); } export async function appendTaskOutput( @@ -206,34 +210,22 @@ export async function readTaskOutputBytes( } } -/** Enumerate all persisted tasks for a session. Skips corrupt entries. */ -export async function listTasks(sessionDir: string): Promise { - let entries: string[]; - try { - entries = await readdir(tasksDirOf(sessionDir)); - } catch { - return []; - } - const out: PersistedTask[] = []; - for (const entry of entries) { - if (!entry.endsWith('.json')) continue; - const taskId = entry.slice(0, -'.json'.length); - // Silently drop: filename basename is not a valid task id (stray files, - // legacy bg_* leftovers, etc.). - if (!VALID_TASK_ID.test(taskId)) continue; - const task = await readTask(sessionDir, taskId); - // Silently drop: JSON parse failed or file disappeared between readdir - // and readTask. writeTask uses an atomic temp+rename pattern so a - // genuinely truncated file in production is rare; if it happens we - // accept the loss rather than emit a ghost with no recoverable - // metadata beyond the filename. - if (task === undefined) continue; - // Silently drop: parsed JSON is missing one or more required fields - // for a PersistedTask. Treated the same as a missing file. - if (!isValidPersistedTask(task)) continue; - out.push(task); - } - return out; +/** + * Enumerate all persisted tasks for a session. + * + * Skips, silently: + * - basenames that don't match `VALID_TASK_ID` (stray files, legacy + * `bg_*` leftovers, partially-written temp files); + * - files that fail to read / parse; + * - records that fail `isValidPersistedTask` (canonical "spec with + * missing fields" failure mode). + * + * `writeTask` uses atomic temp+rename so a genuinely truncated file in + * production is rare; if it happens we accept the loss rather than + * emit a ghost with no recoverable metadata beyond the filename. + */ +export async function listTasks(sessionDir: string): Promise { + return storeFor(sessionDir).list(); } /** @@ -256,14 +248,15 @@ function isValidPersistedTask(obj: unknown): obj is PersistedTask { ); } -/** Remove a task file (idempotent). */ +/** + * Remove a task — both the per-id JSON spec and the task's `output.log` + * directory. Idempotent: missing spec or missing output dir is not an + * error. Throws for an invalid `taskId` (path-traversal guard fires + * before any FS call). + */ export async function removeTask(sessionDir: string, taskId: string): Promise { - // Path-traversal validation outside try/catch. - const path = taskFile(sessionDir, taskId); - try { - await unlink(path); - } catch (error) { - if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error; - } + await storeFor(sessionDir).remove(taskId); + // `taskOutputDir` re-validates the id, so a malformed id throws here + // even if the spec rm above silently returned; matches prior behavior. await rm(taskOutputDir(sessionDir, taskId), { recursive: true, force: true }); } diff --git a/packages/agent-core/src/tools/builtin/index.ts b/packages/agent-core/src/tools/builtin/index.ts index 046d47557..ebbe0dc71 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 000000000..98b784a87 --- /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 000000000..fbd6598b5 --- /dev/null +++ b/packages/agent-core/src/tools/cron/cron-create.md @@ -0,0 +1,84 @@ +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