From 971fce6e528c2b210df1852d7cd12bcda71014fd Mon Sep 17 00:00:00 2001 From: Haozhe Date: Thu, 28 May 2026 21:05:46 +0800 Subject: [PATCH] feat(agent-core): add session-scoped cron tools with persistence (#157) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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. * feat(agent-core): add 5-field cron expression parser 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. * 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. * 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. * 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. * 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. * feat(agent-core): add in-memory SessionCronStore 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). * feat(agent-core): add session-only CronScheduler engine 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. * 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. * 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. * feat(agent-core): add CronList tool Read-only tool surfacing every scheduled cron job for the session. Each record carries id / cron / humanSchedule / nextFireAt / recurring / durable / ageDays / stale, formatted in the same key: value\n---\n shape as TaskList. nextFireAt is the post-jitter timestamp so the model sees what the scheduler will actually fire on. Malformed cron strings render with null nextFireAt instead of throwing — defends against any future direct store inserts. * feat(agent-core): add CronDelete tool Validates the 8-hex id shape up front and routes deletion through the manager so cron_deleted telemetry stays consistent with cron_scheduled. Not-found is reported as an error so the model corrects itself rather than silently believing the delete succeeded — the next CronList would still show whatever id was missed. * feat(agent-core): wire CronManager + cron tools into Agent Agent gains a public cron field constructed and started in the constructor. The scheduler's setInterval is .unref()'d so the cron timer never keeps the process alive, and isKilled (KIMI_DISABLE_CRON) short-circuits every tick, so eager start is safe. ToolManager registers CronCreate/CronList/CronDelete next to the background tools. initializeBuiltinTools runs lazily after the Agent constructor finishes, so this.agent.cron is already defined when the tools are constructed. * feat(agent-core): add manual-tick env + SIGUSR1 bench hook KIMI_CRON_MANUAL_TICK=1 forces the scheduler into manual-drive mode (pollIntervalMs: null), and in the same gate SIGUSR1 binds to a no-throw manager.tick() so bench scripts can advance the scheduler with kill -USR1 without a custom RPC. SIGUSR1 binding is opt-in (rather than always-on) for two reasons: the auto-tick interval already advances the scheduler, and a CLI with many subagents would otherwise pile up listeners and trip Node's 10-listener default. Tests cover the env gate, signal swallowing, listener cleanup, and the no-bind path when the env is unset. The AgentTestContext harness gains an onTestFinished cleanup so the auto-started cron manager never leaks across test files. * test(agent-core): add end-to-end session cron smoke Exercises the full Agent → ToolManager → CronScheduler stack through the production CronCreateTool surface. Local-time anchor + injected ClockSources make coalescedCount=3 deterministic across host timezones. A second case walks the Create → List → Delete tool cycle to confirm the three-tool surface composes round-trip. * test(agent-core): extract shared cron test harness Pulls the duplicated createAgentStub + createClocks helpers out of the five cron test files into test/agent/cron/harness/stub.ts. The shared stub keeps the lightweight-Agent shape (only turn + telemetry surfaces need to look real) while letting individual tests opt into the options that mattered locally (hasActiveTurn / steerReturns). * test(agent-core): trim cron test file headers Compresses each cron test header to a couple of lines describing what the file covers. The long rationale blocks were process documentation (why-this-file-exists, why-stub-vs-real-agent, plan-doc references) that didn't help anyone reading the test later. The few details that mattered (local-time anchor, coalescedCount math) stayed inline next to the code that needs them. E2E test also picks up the shared createClocks helper instead of defining its own. * test(agent-core): convert cron tool output assertions to inline snapshots The 21 multi-field toMatch / toContain assertions across CronCreate / CronList / CronDelete tests covered the same ground a snapshot would have but cost more diff churn when a format detail changes. Errors become single-line snapshots; success outputs go through a small scrubCronOutput helper that replaces the random 8-hex id and ISO timestamp with stable placeholders so the snapshot is deterministic across TZ and run. * refactor(cron): remove durable flag, env clock source, and enable cron tools in default profile - Remove the unimplemented durable persistence field from the entire cron stack (types, tools, manager, tests, docs) to avoid misleading the model into promising cross-session persistence that does not exist yet. - Drop the env:VAR clock source in favour of the file:path source for test/bench control. - Register CronCreate, CronList, and CronDelete in the default agent profile. * fix(agent-core): address PR #136 typecheck and Codex review for cron tools Typecheck: - Bracket-access KIMI_CRON_* env reads under TS4111 in cron source/tests. - Add `approvalRule` to CronCreate/CronList/CronDelete tool executions. - Guard `cron-expr.detectStep` against undefined array elements; cast the readonly snapshot in session-store.test through `unknown`; import the missing `ClockSources`/`ContentPart` symbols in manager.test. - Extend `isReplayUserTurnRecord` switch to cover the new `cron_job` and `cron_missed` origins so kimi-code stays exhaustive. Semantics (from Codex review): - CronCreate re-reads `wallNow()` and re-checks the session cap inside `execute()`, so manual-approval delays and concurrent prepared calls can't backdate the schedule or breach the cap. - One-shot jitter floors the pull-forward at `task.createdAt`, so a brand-new `:00`/`:30` reminder can't end up before its scheduling time. - Scheduler coalesce loop reapplies the same jitter as the delivery path and advances `lastSeenAt` to the last actually-delivered ideal fire; a not-yet-due jittered slot is no longer lost. One-shot fires always report `coalescedCount: 1`. - Manager removes recurring tasks after the first stale fire and emits a `cron_deleted` event, matching the 7-day auto-expire contract. - CronList anchors one-shot `nextFireAt` at `createdAt`, so a pending today's slot isn't rendered as tomorrow. Tests cover each of the above and a changeset is added. * fix(agent-core,tui): address deep review + Codex review on PR #136 Closes the deep-review pass and the four Codex review rounds that followed on the cron tools feature. Consolidated rather than landed as a series so the PR history reads as one fix wave on top of the original Phase-1 cron implementation. ## Lifecycle + structural - Session.close() now stops every agent's CronManager via `Promise.allSettled` (mirroring `stopBackgroundTasksOnExit`). Without this the 1s setInterval and its closure-captured Agent/Session graph leaked on every closed session. - Agent constructor gates `cron.start()` and the three Cron tool instantiations on `type !== 'sub'`. Subagents no longer pile up empty 1Hz timers or duplicate SIGUSR1 listeners under `KIMI_CRON_MANUAL_TICK=1`. ## Permission + plan-mode parity - `PlanModeGuardDenyPermissionPolicy` denies CronCreate and CronDelete during plan mode (CronList stays allowed); matches the TaskStop precedent. - `DEFAULT_APPROVE_TOOLS` includes CronList for parity with TaskList / TaskOutput so manual-approval mode doesn't prompt on read-only listings. ## Cron-fire envelope + projector - Cron fires wrap `task.prompt` in a `` XML envelope (mirrors `notification-xml.ts`). Without this the `coalescedCount` and `stale` cues documented in cron-create.md were invisible to the LLM. - Projector `isInjectionUserMessage` recognises `` reads at most 64 bytes via `openSync` + `readSync` so a stray-large file can't OOM. - SIGUSR1 handler logs swallowed `tick()` exceptions to stderr when `KIMI_CRON_DEBUG=1` (matches scheduler's debug pattern); silent in production. - Documentation rewrite across cron-list.md / cron-create.md / cron-delete.md so the documented stale + nextFireAt behaviour matches the implementation (recurring stale tasks auto-delete after the final fire; `nextFireAt` is an ISO timestamp; refresh ritual is "just CronCreate again — the old id is already gone"). Tests cover each of the above. Suite at 2090+ passing across agent-core, kosong, and the kimi-code app; typecheck clean across all workspace packages. * feat(cron): persist scheduled tasks across kimi resume Add per-id JSON persistence so cron tasks survive a kimi resume of the same session. Core changes: - CronManager: addTask / removeTasks mirror mutations to /cron/.json - CronManager.loadFromDisk() rehydrates the in-memory store on resume - CronManager.flushPersist() drains pending writes for graceful shutdown - SessionCronStore.adopt() inserts persisted tasks with original id and createdAt - Extract shared createPerIdJsonStore utility from background/persist.ts - Refactor background/persist.ts to use createPerIdJsonStore (no behavior change) - New tests: resume.test.ts, persist.test.ts, per-id-json-store.test.ts - Update cron-create / cron-list / cron-delete docs to reflect session lifetime * fix(agent-core): fix cron-stop-on-close test import for tsgo compat * feat(cron): persist lastFiredAt cursor and scope approval rules - persist lastFiredAt across resume so recurring tasks don't replay - add one-shot pinned-date guard to reject >1-year-out first fires - scope CronCreate approvalRule to exact payload (cron, prompt, recurring) - add resume replay tests and corrupt-cursor fallback test - stop cron before awaiting background shutdown so due ticks cannot start a fresh turn while session.close() is mid-flight - filter cron_job / cron_missed origins from markdown export so the internal envelope no longer leaks into user-facing exports --- .changeset/cron-tools.md | 6 + .../src/tui/controllers/session-replay.ts | 6 + .../src/tui/utils/export-markdown.ts | 6 + .../kimi-code/src/tui/utils/message-replay.ts | 2 + .../test/tui/export-markdown.test.ts | 21 + .../kimi-code/test/tui/message-replay.test.ts | 48 ++ .../agent-core/src/agent/context/projector.ts | 1 + .../agent-core/src/agent/context/types.ts | 19 + packages/agent-core/src/agent/cron/index.ts | 1 + packages/agent-core/src/agent/cron/manager.ts | 566 ++++++++++++++++ packages/agent-core/src/agent/index.ts | 34 + .../policies/default-tool-approve.ts | 1 + .../policies/plan-mode-guard-deny.ts | 23 +- packages/agent-core/src/agent/tool/index.ts | 3 + .../agent-core/src/profile/default/agent.yaml | 3 + packages/agent-core/src/session/index.ts | 17 + .../src/tools/background/persist.ts | 125 ++-- .../agent-core/src/tools/builtin/index.ts | 3 + packages/agent-core/src/tools/cron/clock.ts | 148 +++++ .../agent-core/src/tools/cron/cron-create.md | 84 +++ .../agent-core/src/tools/cron/cron-create.ts | 327 +++++++++ .../agent-core/src/tools/cron/cron-delete.md | 39 ++ .../agent-core/src/tools/cron/cron-delete.ts | 119 ++++ .../agent-core/src/tools/cron/cron-expr.ts | 451 +++++++++++++ .../src/tools/cron/cron-fire-xml.ts | 44 ++ .../agent-core/src/tools/cron/cron-list.md | 53 ++ .../agent-core/src/tools/cron/cron-list.ts | 170 +++++ packages/agent-core/src/tools/cron/jitter.ts | 170 +++++ packages/agent-core/src/tools/cron/persist.ts | 64 ++ .../agent-core/src/tools/cron/scheduler.ts | 516 +++++++++++++++ .../src/tools/cron/session-store.ts | 151 +++++ .../src/tools/cron/telemetry-events.ts | 11 + packages/agent-core/src/tools/cron/types.ts | 28 + .../agent-core/src/utils/per-id-json-store.ts | 147 +++++ .../agent-core/test/agent/context.test.ts | 18 + .../test/agent/cron/agent-integration.test.ts | 81 +++ .../test/agent/cron/cron.e2e.test.ts | 207 ++++++ .../test/agent/cron/harness/stub.ts | 123 ++++ .../test/agent/cron/manager.test.ts | 452 +++++++++++++ .../test/agent/cron/manual-tick.test.ts | 229 +++++++ .../agent-core/test/agent/cron/resume.test.ts | 361 ++++++++++ .../test/agent/cron/subagent-skip.test.ts | 88 +++ .../agent-core/test/agent/harness/agent.ts | 12 +- .../permission/default-tool-approve.test.ts | 47 ++ .../test/session/cron-stop-on-close.test.ts | 94 +++ .../agent-core/test/tools/cron/clock.test.ts | 147 +++++ .../test/tools/cron/cron-create.test.ts | 582 +++++++++++++++++ .../test/tools/cron/cron-delete.test.ts | 177 +++++ .../test/tools/cron/cron-expr.test.ts | 398 +++++++++++ .../test/tools/cron/cron-fire-xml.test.ts | 92 +++ .../test/tools/cron/cron-list.test.ts | 381 +++++++++++ .../agent-core/test/tools/cron/jitter.test.ts | 277 ++++++++ .../test/tools/cron/no-date-now.test.ts | 53 ++ .../test/tools/cron/persist.test.ts | 74 +++ .../test/tools/cron/scheduler.test.ts | 618 ++++++++++++++++++ .../test/tools/cron/session-store.test.ts | 180 +++++ .../test/tools/plan-mode-hard-block.test.ts | 29 + .../test/utils/per-id-json-store.test.ts | 176 +++++ 58 files changed, 8230 insertions(+), 73 deletions(-) create mode 100644 .changeset/cron-tools.md 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/src/tools/cron/clock.ts 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/src/tools/cron/cron-delete.md create mode 100644 packages/agent-core/src/tools/cron/cron-delete.ts create mode 100644 packages/agent-core/src/tools/cron/cron-expr.ts create mode 100644 packages/agent-core/src/tools/cron/cron-fire-xml.ts create mode 100644 packages/agent-core/src/tools/cron/cron-list.md create mode 100644 packages/agent-core/src/tools/cron/cron-list.ts create mode 100644 packages/agent-core/src/tools/cron/jitter.ts create mode 100644 packages/agent-core/src/tools/cron/persist.ts create mode 100644 packages/agent-core/src/tools/cron/scheduler.ts create mode 100644 packages/agent-core/src/tools/cron/session-store.ts create mode 100644 packages/agent-core/src/tools/cron/telemetry-events.ts create mode 100644 packages/agent-core/src/tools/cron/types.ts create mode 100644 packages/agent-core/src/utils/per-id-json-store.ts create mode 100644 packages/agent-core/test/agent/cron/agent-integration.test.ts create mode 100644 packages/agent-core/test/agent/cron/cron.e2e.test.ts create mode 100644 packages/agent-core/test/agent/cron/harness/stub.ts create mode 100644 packages/agent-core/test/agent/cron/manager.test.ts create mode 100644 packages/agent-core/test/agent/cron/manual-tick.test.ts create mode 100644 packages/agent-core/test/agent/cron/resume.test.ts create mode 100644 packages/agent-core/test/agent/cron/subagent-skip.test.ts create mode 100644 packages/agent-core/test/agent/permission/default-tool-approve.test.ts create mode 100644 packages/agent-core/test/session/cron-stop-on-close.test.ts create mode 100644 packages/agent-core/test/tools/cron/clock.test.ts create mode 100644 packages/agent-core/test/tools/cron/cron-create.test.ts create mode 100644 packages/agent-core/test/tools/cron/cron-delete.test.ts create mode 100644 packages/agent-core/test/tools/cron/cron-expr.test.ts create mode 100644 packages/agent-core/test/tools/cron/cron-fire-xml.test.ts create mode 100644 packages/agent-core/test/tools/cron/cron-list.test.ts create mode 100644 packages/agent-core/test/tools/cron/jitter.test.ts create mode 100644 packages/agent-core/test/tools/cron/no-date-now.test.ts create mode 100644 packages/agent-core/test/tools/cron/persist.test.ts create mode 100644 packages/agent-core/test/tools/cron/scheduler.test.ts create mode 100644 packages/agent-core/test/tools/cron/session-store.test.ts create mode 100644 packages/agent-core/test/utils/per-id-json-store.test.ts 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