diff --git a/.changeset/fork-cron-clear.md b/.changeset/fork-cron-clear.md new file mode 100644 index 000000000..9337401ce --- /dev/null +++ b/.changeset/fork-cron-clear.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Cron tasks from the source session no longer fire inside a forked session. diff --git a/apps/vis/server/src/lib/agent-record-types.ts b/apps/vis/server/src/lib/agent-record-types.ts index ee3384038..d591e263b 100644 --- a/apps/vis/server/src/lib/agent-record-types.ts +++ b/apps/vis/server/src/lib/agent-record-types.ts @@ -33,12 +33,12 @@ import type { ExportSessionManifest, FileHistoryCheckpointed, FileHistoryTracked, + Forked, FullCompactionBegin, FullCompactionCancel, FullCompactionComplete, GoalClear, GoalCreate, - GoalForked, GoalUpdate, InteractionRequestEvent, InteractionResolvedEvent, @@ -174,7 +174,7 @@ export type AgentRecord = | WireRecordOf<'cron.delete', CronDeletePayload> | WireRecordOf<'file_history.checkpoint', FileHistoryCheckpointed> | WireRecordOf<'file_history.tracked', FileHistoryTracked> - | WireRecordOf<'forked', GoalForked> + | WireRecordOf<'forked', Forked> | WireRecordOf<'full_compaction.begin', FullCompactionBegin> | WireRecordOf<'full_compaction.cancel', FullCompactionCancel> | WireRecordOf<'full_compaction.complete', FullCompactionComplete> diff --git a/packages/agent-core-v2/docs/wire-manifest.d.ts b/packages/agent-core-v2/docs/wire-manifest.d.ts index 6603ba1cb..5845454c2 100644 --- a/packages/agent-core-v2/docs/wire-manifest.d.ts +++ b/packages/agent-core-v2/docs/wire-manifest.d.ts @@ -36,7 +36,7 @@ // cron.delete (none) src/features/cron/cronOps.ts // file_history.checkpoint fileHistory src/features/fileHistory/fileHistoryOps.ts // file_history.tracked fileHistory src/features/fileHistory/fileHistoryOps.ts -// forked (none) src/features/goal/goalOps.ts +// forked (none) src/session/agentLifecycle/forked.ts // full_compaction.begin fullCompaction src/agent/fullCompaction/compactionOps.ts // full_compaction.cancel fullCompaction src/agent/fullCompaction/compactionOps.ts // full_compaction.complete fullCompaction src/agent/fullCompaction/compactionOps.ts @@ -248,7 +248,7 @@ interface FileHistoryTrackedPayload { /** * states: (none) - * owner: src/features/goal/goalOps.ts + * owner: src/session/agentLifecycle/forked.ts */ interface ForkedPayload { _name: 'forked'; diff --git a/packages/agent-core-v2/src/features/cron/cronOps.ts b/packages/agent-core-v2/src/features/cron/cronOps.ts index 544aa817c..6c282e523 100644 --- a/packages/agent-core-v2/src/features/cron/cronOps.ts +++ b/packages/agent-core-v2/src/features/cron/cronOps.ts @@ -5,7 +5,10 @@ import type { CronJobOrigin } from '#/agent/contextMemory/types'; import type { CronTask } from '#/features/cron/cronTask'; import { Event2 } from '#/app/event/event2'; -export type CronModelState = Map; +export interface CronModelState { + readonly tasks: Map; + readonly forkNotice: { reminderPending: boolean }; +} const cronTaskSchema = z.object({ id: z.string(), diff --git a/packages/agent-core-v2/src/features/cron/cronService.ts b/packages/agent-core-v2/src/features/cron/cronService.ts index 1d8d0d7e8..30a305a21 100644 --- a/packages/agent-core-v2/src/features/cron/cronService.ts +++ b/packages/agent-core-v2/src/features/cron/cronService.ts @@ -23,7 +23,11 @@ import type { CronDeletedEvent, CronScheduledEvent } from '#/app/telemetry/event import { ITelemetryService } from '#/app/telemetry/telemetry'; import { BugIndicatingError } from '#/errors'; import type { ContentPart } from '#human/llm/message'; +import { ContextAppendMessage } from '#/agent/contextMemory/contextEvents'; +import type { ContextMessage } from '#/agent/contextMemory/types'; +import { IAgentReminderService } from '#/features/reminder/reminderService'; import { MAIN_AGENT_ID } from '#/session/agentLifecycle/agentLifecycle'; +import { Forked } from '#/session/agentLifecycle/forked'; import { IEventDispatcher } from '#/state/eventDispatcher'; import { CronAdd, CronCursor, CronDelete, CronFired, type CronModelState } from './cronOps'; @@ -31,6 +35,7 @@ import { CronAdd, CronCursor, CronDelete, CronFired, type CronModelState } from registerEvent2Class(CronAdd); registerEvent2Class(CronDelete); registerEvent2Class(CronCursor); +registerEvent2Class(Forked); const STALE_THRESHOLD_MS = 7 * 24 * 60 * 60 * 1000; const DEFAULT_POLL_INTERVAL_MS = 1_000; @@ -43,14 +48,27 @@ export const CRON_FIRED = 'cron_fired' as const; export const CRON_MISSED = 'cron_missed' as const; export const CRON_DELETED = 'cron_deleted' as const; +const CRON_FORK_CLEARED_REMINDER = [ + 'This fork does not have any scheduled cron tasks.', + 'Tasks from the source session continue to run in the source session.', + 'Create new tasks here if needed.', +].join(' '); + +const CRON_FORK_CLEARED_REMINDER_NAME = 'cron_fork_cleared'; + +function isCronForkClearedReminder(message: ContextMessage): boolean { + const origin = message.origin; + return origin?.kind === 'injection' && origin.variant === CRON_FORK_CLEARED_REMINDER_NAME; +} + interface CronActorContext { - readonly tasks: CronModelState; + readonly model: CronModelState; readonly runtime: AgentActorContext; } interface CronCommitEvent { readonly type: 'cron.commit'; - readonly tasks: CronModelState; + readonly model: CronModelState; } interface CronTickEvent { @@ -154,7 +172,7 @@ function removeTasks( runtime: AgentActorContext, ids: readonly string[], ): readonly string[] { - const removed = ids.filter((id) => runtime.getState().has(id)); + const removed = ids.filter((id) => runtime.getState().tasks.has(id)); if (removed.length > 0) void runtime.dispatch(new CronDelete({ ids: removed })); return removed; } @@ -254,7 +272,7 @@ async function processDue( } const advancedTo = lastDueMs ?? now; state.lastSeenAt.set(task.id, advancedTo); - if (runtime.getState().has(task.id)) { + if (runtime.getState().tasks.has(task.id)) { void runtime.dispatch(new CronCursor({ id: task.id, lastFiredAt: advancedTo })); } } @@ -264,10 +282,10 @@ async function tickCron( state: CronEffectState, ): Promise { await configOf(runtime).ready; - if (cronConfigOf(runtime).disabled || runtime.getState().size === 0) return; + if (cronConfigOf(runtime).disabled || runtime.getState().tasks.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))); + await Promise.all([...runtime.getState().tasks.values()].map((task) => processDue(runtime, state, task, now))); } const cronEffects = fromCallback(({ @@ -283,6 +301,11 @@ const cronEffects = fromCallback(({ sendBack: (event: CronActorEvent) => void; }) => { if (input.runtime.agent.agentId !== MAIN_AGENT_ID) return; + if (input.runtime.getState().forkNotice.reminderPending) { + input.runtime.get(IAgentReminderService).notify(CRON_FORK_CLEARED_REMINDER, { + variant: CRON_FORK_CLEARED_REMINDER_NAME, + }); + } const timer = new IntervalTimer({ unref: true }); const state: CronEffectState = { clocks: SYSTEM_CLOCKS, @@ -356,7 +379,10 @@ const cronActorLogic = setup({ }, actors: { cronEffects }, }).createMachine({ - context: ({ input }) => ({ tasks: new Map(), runtime: input }), + context: ({ input }) => ({ + model: { tasks: new Map(), forkNotice: { reminderPending: false } }, + runtime: input, + }), initial: 'beforeRestore', states: { beforeRestore: { @@ -383,7 +409,7 @@ const cronActorLogic = setup({ }, on: { 'cron.commit': { - actions: assign({ tasks: ({ event }) => event.tasks }), + actions: assign({ model: ({ event }) => event.model }), }, }, }); @@ -429,24 +455,36 @@ export class AgentCronService extends AgentActorService implemen this.actor = this.attachActor(cronActorLogic, { id: 'cron', durable: { - events: [CronAdd, CronDelete, CronCursor], + events: [CronAdd, CronDelete, CronCursor, Forked, ContextAppendMessage], undoable: false, transition: (state, event) => { if (event instanceof CronAdd) { - state.set(event.task.id, event.task); + state.tasks.set(event.task.id, event.task); return; } if (event instanceof CronDelete) { - for (const id of event.ids) state.delete(id); + for (const id of event.ids) state.tasks.delete(id); return; } if (event instanceof CronCursor) { - const task = state.get(event.id); - if (task !== undefined) state.set(event.id, { ...task, lastFiredAt: event.lastFiredAt }); + const task = state.tasks.get(event.id); + if (task !== undefined) state.tasks.set(event.id, { ...task, lastFiredAt: event.lastFiredAt }); + return; + } + if (event instanceof Forked) { + state.forkNotice.reminderPending = + state.tasks.size > 0 || state.forkNotice.reminderPending; + state.tasks.clear(); + return; + } + if (event instanceof ContextAppendMessage) { + if (state.forkNotice.reminderPending && isCronForkClearedReminder(event.message)) { + state.forkNotice.reminderPending = false; + } } }, - read: (snapshot) => (snapshot as CronActorSnapshot).context.tasks, - commit: (actor, tasks) => { actor.send({ type: 'cron.commit', tasks }); }, + read: (snapshot) => (snapshot as CronActorSnapshot).context.model, + commit: (actor, model) => { actor.send({ type: 'cron.commit', model }); }, }, }); } @@ -460,7 +498,7 @@ export class AgentCronService extends AgentActorService implemen } addTask(init: CronTaskInit): CronTask { - const tasks = this.actor.getState(); + const tasks = this.actor.getState().tasks; let id: string | undefined; for (let attempt = 0; attempt < MAX_ID_ATTEMPTS; attempt += 1) { const candidate = ulid(); @@ -482,11 +520,11 @@ export class AgentCronService extends AgentActorService implemen } getTask(id: string): CronTask | undefined { - return this.actor.getState().get(id); + return this.actor.getState().tasks.get(id); } list(): readonly CronTask[] { - return [...this.actor.getState().values()]; + return [...this.actor.getState().tasks.values()]; } isStale(task: CronTask): boolean { @@ -495,7 +533,7 @@ export class AgentCronService extends AgentActorService implemen getNextFireTime(): number | null { let min: number | null = null; - for (const task of this.actor.getState().values()) { + for (const task of this.actor.getState().tasks.values()) { const next = nextFireFor(this.actor, task); if (next !== null && (min === null || next < min)) min = next; } @@ -503,7 +541,7 @@ export class AgentCronService extends AgentActorService implemen } getNextFireForTask(taskId: string): number | null { - const task = this.actor.getState().get(taskId); + const task = this.actor.getState().tasks.get(taskId); return task === undefined ? null : nextFireFor(this.actor, task); } diff --git a/packages/agent-core-v2/src/features/goal/goalOps.ts b/packages/agent-core-v2/src/features/goal/goalOps.ts index 6e2169b06..ffe5f6823 100644 --- a/packages/agent-core-v2/src/features/goal/goalOps.ts +++ b/packages/agent-core-v2/src/features/goal/goalOps.ts @@ -111,17 +111,6 @@ export interface GoalClear { readonly agentId: string; } -const goalForkedSchema = z.object({ agentId: z.string() }); - -export class GoalForked extends AgentEvent2> { - static override readonly type = 'forked'; - static override readonly durable = true; - static override readonly schema = goalForkedSchema; -} -export interface GoalForked { - readonly agentId: string; -} - export interface GoalUpdatedPayload { readonly agentId: string; snapshot: GoalSnapshot | null; diff --git a/packages/agent-core-v2/src/features/goal/goalService.ts b/packages/agent-core-v2/src/features/goal/goalService.ts index f914554de..994d5ac5d 100644 --- a/packages/agent-core-v2/src/features/goal/goalService.ts +++ b/packages/agent-core-v2/src/features/goal/goalService.ts @@ -49,6 +49,7 @@ import { type KimiErrorPayload, } from '#/errors'; import { IAgentLifecycleService, MAIN_AGENT_ID } from '#/session/agentLifecycle/agentLifecycle'; +import { Forked } from '#/session/agentLifecycle/forked'; import { ISessionUsageService } from '#/session/usage/sessionUsage'; import { IEventDispatcher } from '#/state/eventDispatcher'; import type { ExecutableToolResult } from '#/tool/toolContract'; @@ -58,7 +59,6 @@ import { IGoalDeadlineScheduler } from './goalDeadlineScheduler'; import { GoalClear, GoalCreate, - GoalForked, GoalUpdate, GoalUpdated, type GoalModelState, @@ -79,7 +79,7 @@ import type { registerEvent2Class(GoalCreate); registerEvent2Class(GoalUpdate); registerEvent2Class(GoalClear); -registerEvent2Class(GoalForked); +registerEvent2Class(Forked); const MAX_GOAL_OBJECTIVE_LENGTH = 4000; @@ -1322,7 +1322,7 @@ export class AgentGoalService extends AgentActorService implem this.actor = this.attachActor(goalActorLogic, { id: 'goal', durable: { - events: [GoalCreate, GoalUpdate, GoalClear, GoalForked, ContextAppendMessage], + events: [GoalCreate, GoalUpdate, GoalClear, Forked, ContextAppendMessage], undoable: false, transition: (state, event) => { if (event instanceof GoalCreate) { @@ -1375,7 +1375,7 @@ export class AgentGoalService extends AgentActorService implem state.forkNotice.goalPresent = false; return; } - if (event instanceof GoalForked) { + if (event instanceof Forked) { state.goal = null; state.forkNotice.reminderPending = state.forkNotice.goalPresent || state.forkNotice.reminderPending; diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index 75351347c..825780dec 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -463,6 +463,7 @@ export * from '#/features/cron/tools/cron-delete/cron-delete'; import '#/session/agentLifecycle/profile/profiles'; export * from '#/session/agentLifecycle/agentLifecycle'; export * from '#/session/agentLifecycle/agentLifecycleService'; +export * from '#/session/agentLifecycle/forked'; export * from '#/session/agentLifecycle/mainAgent'; export * from '#/session/mcp/sessionMcpHandle'; import '#/app/mcpConfig/configSection'; diff --git a/packages/agent-core-v2/src/session/agentLifecycle/forked.ts b/packages/agent-core-v2/src/session/agentLifecycle/forked.ts new file mode 100644 index 000000000..30a910023 --- /dev/null +++ b/packages/agent-core-v2/src/session/agentLifecycle/forked.ts @@ -0,0 +1,15 @@ +/* oxlint-disable typescript-eslint/no-unsafe-declaration-merging, eslint-plugin-import/namespace -- Event2 class+payload-interface declaration merging is the sanctioned event-declaration idiom. */ +import { z } from 'zod'; + +import { AgentEvent2 } from '#/app/event/event2'; + +const forkedSchema = z.object({ agentId: z.string() }); + +export class Forked extends AgentEvent2> { + static override readonly type = 'forked'; + static override readonly durable = true; + static override readonly schema = forkedSchema; +} +export interface Forked { + readonly agentId: string; +} diff --git a/packages/agent-core-v2/test/features/cron/sessionCron.test.ts b/packages/agent-core-v2/test/features/cron/sessionCron.test.ts index 073df4135..6c4bab273 100644 --- a/packages/agent-core-v2/test/features/cron/sessionCron.test.ts +++ b/packages/agent-core-v2/test/features/cron/sessionCron.test.ts @@ -1,8 +1,10 @@ import { describe, expect, it } from 'vitest'; +import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; import { IAgentToolRegistryService } from '#/agent/toolRegistry/toolRegistry'; import { IAgentCronService } from '#/features/cron/cronService'; import { CronCursor } from '#/features/cron/cronOps'; +import type { WireRecord } from '#/wire/record'; import { createTestAgent, @@ -20,6 +22,13 @@ async function bootCronContext(options: TestAgentOptions = {}): Promise { + const ctx = await bootCronContext(); + ctx.get(IAgentCronService).list(); + await ctx.restore(records); + return ctx; +} + describe('session cron wire persistence', () => { it('writes cron ops as durable wire records and rebuilds the task table on replay', async () => { const persistence = new InMemoryWireRecordPersistence(); @@ -28,12 +37,15 @@ describe('session cron wire persistence', () => { await first.restorePersisted(); const cron = first.get(IAgentCronService); - const task = cron.addTask({ cron: '0 9 * * *', prompt: 'wire me', recurring: true }); - await first.dispatcher.dispatch(new CronCursor({ id: task.id, lastFiredAt: 1234 })); + const kept = cron.addTask({ cron: '0 9 * * *', prompt: 'keep', recurring: true }); + const dropped = cron.addTask({ cron: '0 10 * * *', prompt: 'drop', recurring: true }); + cron.removeTasks([dropped.id]); + await first.dispatcher.dispatch(new CronCursor({ id: kept.id, lastFiredAt: 1234 })); await first.dispatcher.flush(); const types = persistence.records.map((record) => record.type); expect(types).toContain('cron.add'); + expect(types).toContain('cron.delete'); expect(types).toContain('cron.cursor'); } finally { await first.dispose(); @@ -50,7 +62,7 @@ describe('session cron wire persistence', () => { expect(rebuilt).toHaveLength(1); expect(rebuilt[0]).toMatchObject({ cron: '0 9 * * *', - prompt: 'wire me', + prompt: 'keep', recurring: true, lastFiredAt: 1234, }); @@ -59,35 +71,65 @@ describe('session cron wire persistence', () => { } }); - it('drops deleted tasks on replay', async () => { - const persistence = new InMemoryWireRecordPersistence(); - const first = await bootCronContext({ persistence }); + it('clears inherited tasks at the fork record and delivers the fork-cleared reminder exactly once', async () => { + const addRecord = (id: string, prompt: string): WireRecord => ({ + type: 'cron.add', + task: { id, cron: '0 9 * * *', prompt, createdAt: 1000, recurring: true }, + }); + + const withInherited = await restoreCronRecords([ + addRecord('seed-inherited', 'inherited'), + { type: 'forked' }, + ]); try { - await first.restorePersisted(); - - const cron = first.get(IAgentCronService); - const kept = cron.addTask({ cron: '0 9 * * *', prompt: 'keep', recurring: true }); - const dropped = cron.addTask({ cron: '0 10 * * *', prompt: 'drop', recurring: true }); - cron.removeTasks([dropped.id]); - await first.dispatcher.flush(); - - const types = persistence.records.map((record) => record.type); - expect(types).toContain('cron.delete'); - expect(kept.id).not.toBe(dropped.id); + expect(withInherited.get(IAgentCronService).list()).toEqual([]); + const reminder = withInherited.get(IAgentContextMemoryService).get().at(-1); + expect(reminder?.origin).toEqual({ kind: 'injection', variant: 'cron_fork_cleared' }); + const text = JSON.stringify(reminder?.content); + expect(text).toContain('This fork does not have any scheduled cron tasks.'); + expect(text).toContain('Tasks from the source session continue to run in the source session.'); } finally { - await first.dispose(); + await withInherited.dispose(); } - const second = await bootCronContext({ - persistence: new InMemoryWireRecordPersistence(persistence.records), - }); + const noTasks = await restoreCronRecords([{ type: 'forked' }]); try { - await second.restorePersisted(); - - const resumed = second.get(IAgentCronService); - expect(resumed.list().map((task) => task.prompt)).toEqual(['keep']); + expect(noTasks.get(IAgentCronService).list()).toEqual([]); + expect(noTasks.get(IAgentContextMemoryService).get()).toEqual([]); } finally { - await second.dispose(); + await noTasks.dispose(); + } + + const recreated = await restoreCronRecords([ + addRecord('seed-inherited', 'inherited'), + { type: 'forked' }, + addRecord('seed-recreated', 'recreated'), + ]); + try { + expect(recreated.get(IAgentCronService).list().map((task) => task.prompt)).toEqual([ + 'recreated', + ]); + } finally { + await recreated.dispose(); + } + + const delivered = await restoreCronRecords([ + addRecord('seed-inherited', 'inherited'), + { type: 'forked' }, + { + type: 'context.append_message', + message: { + role: 'user', + content: [{ type: 'text', text: '\nfork cleared\n' }], + toolCalls: [], + origin: { kind: 'injection', variant: 'cron_fork_cleared' }, + }, + }, + ]); + try { + expect(delivered.get(IAgentContextMemoryService).get()).toHaveLength(1); + } finally { + await delivered.dispose(); } }); diff --git a/packages/kap-server/test/sessions.test.ts b/packages/kap-server/test/sessions.test.ts index 9b558a1f0..66dc43285 100644 --- a/packages/kap-server/test/sessions.test.ts +++ b/packages/kap-server/test/sessions.test.ts @@ -1206,7 +1206,7 @@ describe('server-v2 /api/v1/sessions', () => { expect(children.body.data.items.some((s) => s.id === forked.body.data.id)).toBe(false); }); - it('fork inherits cron tasks through the copied wire', async () => { + it('fork clears inherited cron tasks while the source keeps them', async () => { const cwd = home as string; const parent = await postJson('/api/v1/sessions', { metadata: { cwd } }); const parentId = parent.body.data.id; @@ -1225,7 +1225,8 @@ describe('server-v2 /api/v1/sessions', () => { const resumed = await resumeSessionById((server as RunningServer).core.accessor, forkedId); expect(resumed).toBeDefined(); const forkedCron = resumed!.accessor.get(IAgentLifecycleService).handleOf(MAIN_AGENT_ID)!.accessor.get(IAgentCronService); - expect(forkedCron.list().map((t) => ({ id: t.id, prompt: t.prompt }))).toEqual([ + expect(forkedCron.list()).toEqual([]); + expect(cron.list().map((t) => ({ id: t.id, prompt: t.prompt }))).toEqual([ { id: task.id, prompt: 'fork me' }, ]); });