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

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-code": patch
+---
+
+Cron tasks from the source session no longer fire inside a forked session.

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>

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';

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<string, CronTask>;
+export interface CronModelState {
+  readonly tasks: Map<string, CronTask>;
+  readonly forkNotice: { reminderPending: boolean };
+}
 
 const cronTaskSchema = z.object({
   id: z.string(),

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<CronModelState>;
 }
 
 interface CronCommitEvent {
   readonly type: 'cron.commit';
-  readonly tasks: CronModelState;
+  readonly model: CronModelState;
 }
 
 interface CronTickEvent {
@@ -154,7 +172,7 @@ function removeTasks(
   runtime: AgentActorContext<CronModelState>,
   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<void> {
   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<CronModelState> 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<CronModelState> 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<CronModelState> 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<CronModelState> 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<CronModelState> 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);
   }
 

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<z.infer<typeof goalForkedSchema>> {
-  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;

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<GoalRuntimeState> 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<GoalRuntimeState> 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;

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';

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<z.infer<typeof forkedSchema>> {
+  static override readonly type = 'forked';
+  static override readonly durable = true;
+  static override readonly schema = forkedSchema;
+}
+export interface Forked {
+  readonly agentId: string;
+}

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<TestAgen
   return ctx;
 }
 
+async function restoreCronRecords(records: readonly WireRecord[]): Promise<TestAgentContext> {
+  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: '<system-reminder>\nfork cleared\n</system-reminder>' }],
+          toolCalls: [],
+          origin: { kind: 'injection', variant: 'cron_fork_cleared' },
+        },
+      },
+    ]);
+    try {
+      expect(delivered.get(IAgentContextMemoryService).get()).toHaveLength(1);
+    } finally {
+      await delivered.dispose();
     }
   });
 

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<SessionWire>('/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' },
     ]);
   });