fix(agent-core-v2): stop cron ticks after shutdown

Author: elkaix

.changeset/stop-cron-after-close.md

@@ -0,0 +1,5 @@
+---
+"@pymodel/pythinker-code": patch
+---
+
+Prevent cron ticks from continuing after an agent shuts down.

packages/agent-core-v2/src/features/cron/cronAgentRuntime.ts

@@ -68,8 +68,12 @@ function configOf(runtime: AgentRuntimeContext<CronModelState>): IConfigService
   return runtime.get(IConfigService);
 }
 
+function readCronConfig(config: IConfigService): CronConfig {
+  return config.get<CronConfig>(CRON_SECTION) ?? DEFAULT_CRON_CONFIG;
+}
+
 function cronConfigOf(runtime: AgentRuntimeContext<CronModelState>): CronConfig {
-  return configOf(runtime).get<CronConfig>(CRON_SECTION) ?? DEFAULT_CRON_CONFIG;
+  return readCronConfig(configOf(runtime));
 }
 
 function clocksOf(runtime: AgentRuntimeContext<CronModelState>): ClockSources {
@@ -265,9 +269,12 @@ async function processDue(
 async function tickCron(
   runtime: AgentRuntimeContext<CronModelState>,
   state: CronEffectState,
+  config: IConfigService,
+  isDisposed: () => boolean,
 ): Promise<void> {
-  await configOf(runtime).ready;
-  if (cronConfigOf(runtime).disabled || runtime.getState().size === 0) return;
+  await config.ready;
+  if (isDisposed()) return;
+  if (readCronConfig(config).disabled || runtime.getState().size === 0) return;
   if (runtime.get(IAgentLoopService).status().state === 'running') return;
   const now = state.clocks.wallNow();
   await Promise.all([...runtime.getState().values()].map((task) => processDue(runtime, state, task, now)));
@@ -286,6 +293,7 @@ const cronEffects = fromCallback(({
   sendBack: (event: CronActorEvent) => void;
 }) => {
   if (input.runtime.agent.agentId !== MAIN_AGENT_ID) return;
+  const config = configOf(input.runtime);
   const timer = new IntervalTimer({ unref: true });
   const state: CronEffectState = {
     clocks: SYSTEM_CLOCKS,
@@ -297,18 +305,22 @@ const cronEffects = fromCallback(({
   let disposed = false;
   let signalHandler: NodeJS.SignalsListener | undefined;
   receive((event) => {
-    void tickCron(input.runtime, state).then(event.resolve, event.reject);
+    if (disposed) {
+      event.resolve?.();
+      return;
+    }
+    void tickCron(input.runtime, state, config, () => disposed).then(event.resolve, event.reject);
   });
-  input.restore.waitUntil(configOf(input.runtime).ready.then(() => {
+  input.restore.waitUntil(config.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 current = readCronConfig(config);
+    state.clocks = resolveClockSources(current.clock, current.debug) ?? SYSTEM_CLOCKS;
+    const poll = current.manualTick ? null : current.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) {
+    if (process.platform !== 'win32' && current.manualTick) {
       signalHandler = () => { sendBack({ type: 'cron.tick' }); };
       process.on('SIGUSR1', signalHandler);
     }

packages/agent-core-v2/test/features/cron/sessionCron.test.ts

@@ -1,7 +1,13 @@
 import { describe, expect, it } from 'vitest';
 
+import { DisposableStore } from '#/_base/di/lifecycle';
+import { createServices } from '#/_base/di/test';
+import type { AgentContext } from '#/agent/agentContext/agentContext';
+import { AgentRuntimeSet } from '#/agent/runtime/agentRuntimeSet';
 import { IAgentToolRegistryService } from '#/agent/toolRegistry/toolRegistry';
-import { AgentCron } from '#/features/cron/cronAgentRuntime';
+import { IConfigService } from '#/app/config/config';
+import { DEFAULT_CRON_CONFIG } from '#/features/cron/configSection';
+import { AgentCron, cronAgentRuntimeProvider } from '#/features/cron/cronAgentRuntime';
 import { CronCursor } from '#/features/cron/cronOps';
 
 import {
@@ -21,6 +27,51 @@ async function bootCronContext(options: TestAgentOptions = {}): Promise<TestAgen
 }
 
 describe('session cron wire persistence', () => {
+  it('settles an in-flight tick without reading services after close', async () => {
+    let releaseReady!: () => void;
+    const ready = new Promise<void>((resolve) => { releaseReady = resolve; });
+    let markTickStarted!: () => void;
+    const tickStarted = new Promise<void>((resolve) => { markTickStarted = resolve; });
+    let readyReads = 0;
+    let closed = false;
+    const disposables = new DisposableStore();
+    const services = createServices(disposables, {
+      additionalServices: (reg) => {
+        reg.definePartialInstance(IConfigService, {
+          get ready() {
+            readyReads += 1;
+            if (readyReads === 2) markTickStarted();
+            return ready;
+          },
+          get: <T>() => {
+            if (closed) throw new Error('config read after close');
+            return DEFAULT_CRON_CONFIG as T;
+          },
+        });
+      },
+    });
+    const agent = { agentId: 'main', generation: 1, space: {} } as AgentContext;
+    const runtimes = new AgentRuntimeSet(agent, services);
+    runtimes.apply({
+      definition: AgentCron,
+      provider: cronAgentRuntimeProvider,
+      generation: 1,
+      active: true,
+    });
+    runtimes.attachDurable({ attach: () => ({ dispose: () => {} }) });
+
+    const restoring = runtimes.restore();
+    const ticking = runtimes.resolve(AgentCron).tick();
+    await tickStarted;
+    await runtimes.close();
+    closed = true;
+    disposables.dispose();
+    releaseReady();
+
+    await expect(ticking).resolves.toBeUndefined();
+    await expect(restoring).resolves.toBeUndefined();
+  });
+
   it('writes cron ops as durable wire records and rebuilds the task table on replay', async () => {
     const persistence = new InMemoryWireRecordPersistence();
     const first = await bootCronContext({ persistence });

packages/agent-core-v2/test/features/tower/store.test.ts

@@ -754,7 +754,7 @@ describe('merge gate', () => {
 
     const after = await store.merge(third!.branch);
     expect(after.conflictsWith.map((c) => c.branch)).not.toContain(second!.branch);
-  });
+  }, 15_000);
 });
 
 describe('updateMission', () => {