import { ulid } from 'ulid'; import { assign, fromCallback, sendTo, setup, type Snapshot } from 'xstate'; import { createDecorator, IInstantiationService } from '#/_base/di/instantiation'; import { IntervalTimer } from '#/_base/utils/timer'; import type { CronJobOrigin, CronMissedOrigin } from '#/agent/contextMemory/types'; import { IAgentLoopService, type Turn } from '#/agent/loop/loop'; import { AgentActorService, type AgentActorContext, type AgentActorRestoreEvent, } from '#/agent/actorService/agentActorService'; import { IAgentScopeContext } from '#/agent/scopeContext/scopeContext'; import { registerEvent2Class } from '#/app/event/event2'; import { IConfigService } from '#/app/config/config'; import { type ClockSources, resolveClockSources, SYSTEM_CLOCKS } from '#/features/cron/internal/clock'; import { type CronConfig, CRON_SECTION, DEFAULT_CRON_CONFIG } from '#/features/cron/configSection'; import { computeNextCronRun, parseCronExpression, type ParsedCronExpression } from '#/features/cron/internal/cron-expr'; import type { CronTask, CronTaskInit } from '#/features/cron/cronTask'; import { renderCronFireXml } from '#/features/cron/internal/format'; import { jitteredNextCronRunMs, oneShotJitteredNextCronRunMs } from '#/features/cron/internal/jitter'; import type { CronDeletedEvent, CronScheduledEvent } from '#/app/telemetry/events'; import { ITelemetryService } from '#/app/telemetry/telemetry'; import { BugIndicatingError } from '#/errors'; import type { ContentPart } from '#human/llm/message'; import { MAIN_AGENT_ID } from '#/session/agentLifecycle/agentLifecycle'; import { IEventDispatcher } from '#/state/eventDispatcher'; import { CronAdd, CronCursor, CronDelete, CronFired, type CronModelState } from './cronOps'; registerEvent2Class(CronAdd); registerEvent2Class(CronDelete); registerEvent2Class(CronCursor); const STALE_THRESHOLD_MS = 7 * 24 * 60 * 60 * 1000; const DEFAULT_POLL_INTERVAL_MS = 1_000; const MAX_COALESCE_ITERATIONS = 10_000; const CRON_ID_REGEX: RegExp = /^(?:[0-9a-f]{8}|[0-9A-HJKMNP-TV-Z]{26})$/i; const MAX_ID_ATTEMPTS = 8; export const CRON_SCHEDULED = 'cron_scheduled' as const; export const CRON_FIRED = 'cron_fired' as const; export const CRON_MISSED = 'cron_missed' as const; export const CRON_DELETED = 'cron_deleted' as const; interface CronActorContext { readonly tasks: CronModelState; readonly runtime: AgentActorContext; } interface CronCommitEvent { readonly type: 'cron.commit'; readonly tasks: CronModelState; } interface CronTickEvent { readonly type: 'cron.tick'; readonly resolve?: () => void; readonly reject?: (error: unknown) => void; } type CronActorEvent = CronCommitEvent | AgentActorRestoreEvent | CronTickEvent; type CronActorSnapshot = Snapshot & { readonly context: CronActorContext }; interface CronEffectState { clocks: ClockSources; readonly parsedCache: Map; readonly lastSeenAt: Map; readonly seededFromStore: Set; readonly inFlight: Set; } function configOf(runtime: AgentActorContext): IConfigService { return runtime.get(IConfigService); } function cronConfigOf(runtime: AgentActorContext): CronConfig { return configOf(runtime).get(CRON_SECTION) ?? DEFAULT_CRON_CONFIG; } function clocksOf(runtime: AgentActorContext): ClockSources { const config = cronConfigOf(runtime); return resolveClockSources(config.clock, config.debug) ?? SYSTEM_CLOCKS; } function telemetryOf(runtime: AgentActorContext): ITelemetryService { return runtime.get(ITelemetryService); } function debugLog(runtime: AgentActorContext, message: string): void { if (cronConfigOf(runtime).debug) process.stderr.write(`[cron/session] ${message}\n`); } function isStaleAt( runtime: AgentActorContext, task: CronTask, now: number, ): boolean { if (cronConfigOf(runtime).noStale) return false; if (task.recurring === false) return false; const age = now - task.createdAt; return Number.isFinite(age) && age >= STALE_THRESHOLD_MS; } function computeJitteredNext( runtime: AgentActorContext, task: CronTask, parsed: ParsedCronExpression, baseMs: number, ): number | null { const ideal = computeNextCronRun(parsed, baseMs); if (ideal === null) return null; const noJitter = cronConfigOf(runtime).noJitter; if (task.recurring === false) { return oneShotJitteredNextCronRunMs(task, ideal, undefined, noJitter); } return jitteredNextCronRunMs(task, parsed, ideal, undefined, noJitter); } function parsedCron(state: CronEffectState, expression: string): ParsedCronExpression { const cached = state.parsedCache.get(expression); if (cached !== undefined) return cached; const parsed = parseCronExpression(expression); state.parsedCache.set(expression, parsed); return parsed; } function countCoalesced( runtime: AgentActorContext, task: CronTask, parsed: ParsedCronExpression, firstFireMs: number, nowMs: number, ): { count: number; lastDueMs: number } { let count = 1; let cursor = firstFireMs; let lastDueMs = firstFireMs; const noJitter = cronConfigOf(runtime).noJitter; while (count < MAX_COALESCE_ITERATIONS) { const next = computeNextCronRun(parsed, cursor); if (next === null || next > nowMs) break; const jitteredNext = task.recurring === false ? oneShotJitteredNextCronRunMs(task, next, undefined, noJitter) : jitteredNextCronRunMs(task, parsed, next, undefined, noJitter); if (jitteredNext > nowMs) break; count += 1; cursor = next; lastDueMs = next; } return { count, lastDueMs }; } function removeTasks( runtime: AgentActorContext, ids: readonly string[], ): readonly string[] { const removed = ids.filter((id) => runtime.getState().has(id)); if (removed.length > 0) void runtime.dispatch(new CronDelete({ ids: removed })); return removed; } function deliverFire( runtime: AgentActorContext, task: CronTask, context: { readonly coalescedCount: number; readonly firedAt: number }, ): Promise { const origin: CronJobOrigin = { kind: 'cron_job', jobId: task.id, cron: task.cron, recurring: task.recurring !== false, coalescedCount: context.coalescedCount, stale: isStaleAt(runtime, task, context.firedAt), }; const buffered = runtime.get(IAgentLoopService).snapshot().state === 'running'; try { runtime.get(IAgentLoopService).submit( { message: { role: 'user', content: [{ type: 'text', text: renderCronFireXml(origin, task.prompt) }] }, meta: { origin }, }, { steerIfActive: true }, ); } catch (error) { debugLog(runtime, `steer threw for task ${task.id}: ${error instanceof Error ? error.message : String(error)}`); return Promise.resolve(false); } void runtime.dispatch(new CronFired({ origin, prompt: task.prompt })); telemetryOf(runtime).track2(CRON_FIRED, { recurring: task.recurring !== false, coalesced_count: context.coalescedCount, stale: origin.stale, buffered, }); return Promise.resolve(true); } async function processDue( runtime: AgentActorContext, state: CronEffectState, task: CronTask, now: number, ): Promise { if (state.inFlight.has(task.id)) return; let parsed: ParsedCronExpression; try { parsed = parsedCron(state, task.cron); } catch (error) { debugLog(runtime, `tick failed to parse cron for task ${task.id}: ${error instanceof Error ? error.message : String(error)}`); return; } if ( !state.seededFromStore.has(task.id) && task.lastFiredAt !== undefined && Number.isFinite(task.lastFiredAt) && task.lastFiredAt <= now && !state.lastSeenAt.has(task.id) ) { state.lastSeenAt.set(task.id, task.lastFiredAt); } state.seededFromStore.add(task.id); const seen = state.lastSeenAt.get(task.id); const baseFromMs = seen !== undefined && seen > task.createdAt ? seen : task.createdAt; const nextFireAt = computeJitteredNext(runtime, task, parsed, baseFromMs); if (nextFireAt === null || now < nextFireAt) return; const ideal = computeNextCronRun(parsed, baseFromMs); let coalescedCount = 1; let lastDueMs: number | null = null; if (task.recurring !== false && ideal !== null) { const result = countCoalesced(runtime, task, parsed, ideal, now); coalescedCount = Math.max(1, result.count); lastDueMs = result.lastDueMs; } state.inFlight.add(task.id); const firedAt = state.clocks.wallNow(); let delivered = false; try { delivered = await deliverFire(runtime, task, { coalescedCount, firedAt }); } catch (error) { debugLog(runtime, `deliverDue threw for task ${task.id}: ${error instanceof Error ? error.message : String(error)}`); } finally { state.inFlight.delete(task.id); } if (!delivered) return; if (task.recurring === false || isStaleAt(runtime, task, firedAt)) { const removed = removeTasks(runtime, [task.id]); state.lastSeenAt.delete(task.id); state.seededFromStore.delete(task.id); if (task.recurring !== false && removed.length > 0) { const properties: CronDeletedEvent = { task_id: task.id, agent_id: undefined }; telemetryOf(runtime).track2(CRON_DELETED, properties); } return; } const advancedTo = lastDueMs ?? now; state.lastSeenAt.set(task.id, advancedTo); if (runtime.getState().has(task.id)) { void runtime.dispatch(new CronCursor({ id: task.id, lastFiredAt: advancedTo })); } } async function tickCron( runtime: AgentActorContext, state: CronEffectState, ): Promise { await configOf(runtime).ready; if (cronConfigOf(runtime).disabled || runtime.getState().size === 0) return; if (runtime.get(IAgentLoopService).snapshot().state === 'running') return; const now = state.clocks.wallNow(); await Promise.all([...runtime.getState().values()].map((task) => processDue(runtime, state, task, now))); } const cronEffects = fromCallback(({ input, receive, sendBack, }: { input: { readonly runtime: AgentActorContext; readonly restore: AgentActorRestoreEvent; }; receive: (listener: (event: CronTickEvent) => void) => void; sendBack: (event: CronActorEvent) => void; }) => { if (input.runtime.agent.agentId !== MAIN_AGENT_ID) return; const timer = new IntervalTimer({ unref: true }); const state: CronEffectState = { clocks: SYSTEM_CLOCKS, parsedCache: new Map(), lastSeenAt: new Map(), seededFromStore: new Set(), inFlight: new Set(), }; let disposed = false; let signalHandler: NodeJS.SignalsListener | undefined; receive((event) => { void tickCron(input.runtime, state).then( () => { event.resolve?.(); }, (error: unknown) => { event.reject?.(error); }, ); }); input.restore.waitUntil(configOf(input.runtime).ready.then(() => { if (disposed) return; const config = cronConfigOf(input.runtime); state.clocks = resolveClockSources(config.clock, config.debug) ?? SYSTEM_CLOCKS; const poll = config.manualTick ? null : config.pollIntervalMs; const interval = poll === undefined ? DEFAULT_POLL_INTERVAL_MS : poll; if (interval !== null && interval !== 0) { timer.cancelAndSet(() => { sendBack({ type: 'cron.tick' }); }, interval); } if (process.platform !== 'win32' && config.manualTick) { signalHandler = () => { sendBack({ type: 'cron.tick' }); }; process.on('SIGUSR1', signalHandler); } })); return () => { disposed = true; timer.dispose(); if (signalHandler !== undefined) process.off('SIGUSR1', signalHandler); state.inFlight.clear(); state.lastSeenAt.clear(); state.seededFromStore.clear(); state.parsedCache.clear(); }; }); function nextFireFor( runtime: AgentActorContext, task: CronTask, ): number | null { try { const clocks = clocksOf(runtime); const parsed = parseCronExpression(task.cron); const persistedCursor = task.lastFiredAt !== undefined && Number.isFinite(task.lastFiredAt) && task.lastFiredAt <= clocks.wallNow() ? task.lastFiredAt : undefined; const baseFromMs = persistedCursor !== undefined && persistedCursor > task.createdAt ? persistedCursor : task.createdAt; return computeJitteredNext(runtime, task, parsed, baseFromMs); } catch (error) { debugLog(runtime, `nextFireFor skipping task ${task.id}: ${error instanceof Error ? error.message : String(error)}`); return null; } } const cronActorLogic = setup({ types: {} as { context: CronActorContext; input: AgentActorContext; events: CronActorEvent; }, actors: { cronEffects }, }).createMachine({ context: ({ input }) => ({ tasks: new Map(), runtime: input }), initial: 'beforeRestore', states: { beforeRestore: { on: { 'runtime.restore': 'active', 'cron.tick': { actions: ({ event }) => { event.reject?.(new Error('Cron runtime is not restored')); }, }, }, }, active: { invoke: { id: 'cronEffects', src: 'cronEffects', input: ({ context, event }) => ({ runtime: context.runtime, restore: event as AgentActorRestoreEvent, }), }, on: { 'cron.tick': { actions: sendTo('cronEffects', ({ event }) => event) }, }, }, }, on: { 'cron.commit': { actions: assign({ tasks: ({ event }) => event.tasks }), }, }, }); export interface IAgentCronService { readonly _serviceBrand: undefined; now(): number; isDisabled(): boolean; addTask(init: CronTaskInit): CronTask; removeTasks(ids: readonly string[]): readonly string[]; getTask(id: string): CronTask | undefined; list(): readonly CronTask[]; isStale(task: CronTask): boolean; getNextFireTime(): number | null; getNextFireForTask(taskId: string): number | null; computeDisplayNextFire( task: CronTask, parsed: ParsedCronExpression, idealMs: number, ): number | null; handleMissed( tasks: readonly CronTask[], renderMissedNotification: (tasks: readonly CronTask[]) => readonly ContentPart[], ): Turn | undefined; emitScheduled(task: CronTask, agentId?: string): void; emitDeleted(taskId: string, agentId?: string): void; tick(): Promise; } export const IAgentCronService = createDecorator('agentCronService'); export class AgentCronService extends AgentActorService implements IAgentCronService { declare readonly _serviceBrand: undefined; private readonly actor: AgentActorContext; constructor( @IEventDispatcher dispatcher: IEventDispatcher, @IAgentScopeContext scopeContext: IAgentScopeContext, @IInstantiationService instantiation: IInstantiationService, ) { super(dispatcher, scopeContext, instantiation); this.actor = this.attachActor(cronActorLogic, { id: 'cron', durable: { events: [CronAdd, CronDelete, CronCursor], undoable: false, transition: (state, event) => { if (event instanceof CronAdd) { state.set(event.task.id, event.task); return; } if (event instanceof CronDelete) { for (const id of event.ids) state.delete(id); return; } if (event instanceof CronCursor) { const task = state.get(event.id); if (task !== undefined) state.set(event.id, { ...task, lastFiredAt: event.lastFiredAt }); } }, read: (snapshot) => (snapshot as CronActorSnapshot).context.tasks, commit: (actor, tasks) => { actor.send({ type: 'cron.commit', tasks }); }, }, }); } now(): number { return clocksOf(this.actor).wallNow(); } isDisabled(): boolean { return cronConfigOf(this.actor).disabled; } addTask(init: CronTaskInit): CronTask { const tasks = this.actor.getState(); let id: string | undefined; for (let attempt = 0; attempt < MAX_ID_ATTEMPTS; attempt += 1) { const candidate = ulid(); if (CRON_ID_REGEX.test(candidate) && !tasks.has(candidate)) { id = candidate; break; } } if (id === undefined) { throw new BugIndicatingError(`SessionCronService: failed to generate a unique ULID after ${MAX_ID_ATTEMPTS} attempts`); } const task: CronTask = { ...init, id, createdAt: this.now() }; void this.actor.dispatch(new CronAdd({ task })); return task; } removeTasks(ids: readonly string[]): readonly string[] { return removeTasks(this.actor, ids); } getTask(id: string): CronTask | undefined { return this.actor.getState().get(id); } list(): readonly CronTask[] { return [...this.actor.getState().values()]; } isStale(task: CronTask): boolean { return isStaleAt(this.actor, task, this.now()); } getNextFireTime(): number | null { let min: number | null = null; for (const task of this.actor.getState().values()) { const next = nextFireFor(this.actor, task); if (next !== null && (min === null || next < min)) min = next; } return min; } getNextFireForTask(taskId: string): number | null { const task = this.actor.getState().get(taskId); return task === undefined ? null : nextFireFor(this.actor, task); } computeDisplayNextFire( task: CronTask, parsed: ParsedCronExpression, idealMs: number, ): number | null { const noJitter = cronConfigOf(this.actor).noJitter; if (task.recurring === false) { return oneShotJitteredNextCronRunMs(task, idealMs, undefined, noJitter); } return jitteredNextCronRunMs(task, parsed, idealMs, undefined, noJitter); } handleMissed( tasks: readonly CronTask[], renderMissedNotification: (tasks: readonly CronTask[]) => readonly ContentPart[], ): Turn | undefined { if (tasks.length === 0) return undefined; const origin: CronMissedOrigin = { kind: 'cron_missed', count: tasks.length }; this.actor.get(IAgentLoopService).submit( { message: { role: 'user', content: [...renderMissedNotification(tasks)] }, meta: { origin }, }, { steerIfActive: true }, ); telemetryOf(this.actor).track2(CRON_MISSED, { count: tasks.length }); return undefined; } emitScheduled(task: CronTask, agentId?: string): void { const properties: CronScheduledEvent = { recurring: task.recurring !== false, agent_id: agentId }; telemetryOf(this.actor).track2(CRON_SCHEDULED, properties); } emitDeleted(taskId: string, agentId?: string): void { const properties: CronDeletedEvent = { task_id: taskId, agent_id: agentId }; telemetryOf(this.actor).track2(CRON_DELETED, properties); } tick(): Promise { return new Promise((resolve, reject) => { this.actor.send({ type: 'cron.tick', resolve, reject }); }); } }