From f409caa21e71ce7beb158d29ffca1fed76216a64 Mon Sep 17 00:00:00 2001 From: Haozhe Date: Tue, 29 Sep 2026 19:10:48 +0800 Subject: [PATCH] fix(agent-core-v2): clear inherited cron tasks at the fork record (#4083) Fork copies the wire journal, so cron tasks persisted as cron.add records were inherited by the fork and fired from both sessions. Fold the forked marker record into the cron actor: the task table is cleared at the fork point and a one-time cron_fork_cleared reminder is delivered (tracked by a durable forkNotice flag confirmed via ContextAppendMessage), mirroring the goal feature's fork handling. The forked event class moves from features/goal to session/agentLifecycle so both features share it. --- .changeset/fork-cron-clear.md | 5 + apps/vis/server/src/lib/agent-record-types.ts | 4 +- .../agent-core-v2/docs/wire-manifest.d.ts | 4 +- .../src/features/cron/cronOps.ts | 5 +- .../src/features/cron/cronService.ts | 78 +++++++++++---- .../src/features/goal/goalOps.ts | 11 --- .../src/features/goal/goalService.ts | 8 +- packages/agent-core-v2/src/index.ts | 1 + .../src/session/agentLifecycle/forked.ts | 15 +++ .../test/features/cron/sessionCron.test.ts | 94 ++++++++++++++----- packages/kap-server/test/sessions.test.ts | 5 +- 11 files changed, 162 insertions(+), 68 deletions(-) create mode 100644 .changeset/fork-cron-clear.md create mode 100644 packages/agent-core-v2/src/session/agentLifecycle/forked.ts 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' }, ]); });