refactor(core): Run AI checkpoint pruning as system tasks (no-changelog) (#37544)

Co-authored-by: Veljko Terzic <veljko.terzic@n8n.io>

packages/@n8n/decorators/src/system-task/index.ts

@@ -6,3 +6,4 @@ export type {
 	SystemTaskSchedule,
 } from './system-task';
 export { SystemTaskMetadata } from './system-task-metadata';
+export { wholeSeconds } from './whole-seconds';

packages/cli/eslint.config.mjs

@@ -417,8 +417,6 @@ export default defineConfig(
 		// @SystemTask() class. Entries are removed as each migrates on its own ticket.
 		files: [
 			'./src/license.ts',
-			'./src/modules/agents/integrations/n8n-checkpoint-storage.ts',
-			'./src/modules/instance-ai/instance-ai.service.ts',
 			'./src/modules/instance-reporting/instance-reporting-scheduler.service.ts',
 			'./src/services/pruning/executions-pruning.service.ts',
 			'./src/services/pruning/workflow-history-compaction.service.ts',

packages/cli/src/commands/start.ts

@@ -31,7 +31,6 @@ import { FeatureNotLicensedError } from '@/errors/feature-not-licensed.error';
 import { MessageEventBus } from '@/eventbus/message-event-bus/message-event-bus';
 import { EventService } from '@/events/event.service';
 import { LoadNodesAndCredentials } from '@/load-nodes-and-credentials';
-import { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage';
 import { MultiMainSetup } from '@/scaling/multi-main-setup.ee';
 import { Publisher } from '@/scaling/pubsub/publisher.service';
 import { PubSubRegistry } from '@/scaling/pubsub/pubsub.registry';
@@ -424,7 +423,6 @@ export class Start extends BaseCommand<z.infer<typeof flagsSchema>> {
 		Container.get(ExecutionsPruningService).init();
 		Container.get(WorkflowHistoryCompactionService).init();
 		Container.get(WorkflowStatisticsRollupService).init();
-		Container.get(N8NCheckpointStorage).init();
 		Container.get(SystemTaskRunner).init();
 		Container.get(DurableScheduler).start();
 

packages/cli/src/modules/agents/__tests__/agent-checkpoint-pruning.task.test.ts

@@ -0,0 +1,24 @@
+import { mock } from 'vitest-mock-extended';
+
+import { AgentCheckpointPruningTask } from '../agent-checkpoint-pruning.task';
+import type { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage';
+
+describe('AgentCheckpointPruningTask', () => {
+	const checkpointStorage = mock<N8NCheckpointStorage>();
+	const task = new AgentCheckpointPruningTask(checkpointStorage);
+
+	it('should declare an hourly prune cadence', () => {
+		expect(task.name).toBe('agent-checkpoint-pruning');
+		expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 3600 });
+		expect(task.effects).toBe('idempotent');
+		expect(task.durable).toBe(false);
+		expect(task.runOnTakeover).toBe(true);
+		expect(task.retryDelaySeconds).toBe(30);
+	});
+
+	it('should prune stale suspensions on run', async () => {
+		await task.run();
+
+		expect(checkpointStorage.pruneStaleSuspensions).toHaveBeenCalledTimes(1);
+	});
+});

packages/cli/src/modules/agents/agent-checkpoint-pruning.task.ts

@@ -0,0 +1,33 @@
+import { Time } from '@n8n/constants';
+import { SystemTask } from '@n8n/decorators';
+import type { SystemTaskEffects, SystemTaskSchedule } from '@n8n/decorators';
+
+import { N8NCheckpointStorage } from './integrations/n8n-checkpoint-storage';
+
+/**
+ * Expires agent checkpoints past their TTL, so a stale suspended run can no
+ * longer be resumed and the checkpoint table stays small.
+ */
+@SystemTask()
+export class AgentCheckpointPruningTask implements SystemTask {
+	readonly name = 'agent-checkpoint-pruning';
+
+	readonly schedule: SystemTaskSchedule = {
+		kind: 'interval',
+		intervalSeconds: Time.hours.toSeconds,
+	};
+
+	readonly effects: SystemTaskEffects = 'idempotent';
+
+	readonly durable = false;
+
+	readonly runOnTakeover = true;
+
+	readonly retryDelaySeconds = 30;
+
+	constructor(private readonly checkpointStorage: N8NCheckpointStorage) {}
+
+	async run(): Promise<void> {
+		await this.checkpointStorage.pruneStaleSuspensions();
+	}
+}

packages/cli/src/modules/agents/agents.module.ts

@@ -160,6 +160,11 @@ export class AgentsModule implements ModuleInterface {
 		}
 	}
 
+	async systemTasks() {
+		const { AgentCheckpointPruningTask } = await import('./agent-checkpoint-pruning.task.js');
+		return [AgentCheckpointPruningTask];
+	}
+
 	@OnShutdown()
 	async shutdown() {
 		if (this.interruptedExecutionSweepTimer) {

packages/cli/src/modules/agents/integrations/__tests__/n8n-checkpoint-storage.test.ts

@@ -1,8 +1,6 @@
 import type { SerializableAgentState } from '@n8n/agents';
-import type { ModuleRegistry } from '@n8n/backend-common';
 import { mockLogger } from '@n8n/backend-test-utils';
 import type { AgentsConfig } from '@n8n/config';
-import type { InstanceSettings } from 'n8n-core';
 import { mock } from 'vitest-mock-extended';
 
 import {
@@ -34,11 +32,9 @@ const principalHash = hashAgentSandboxPrincipal({ type: 'n8n-user', userId: 'use
 function makeService() {
 	const repository = mock<AgentCheckpointRepository>();
 	const service = new N8NCheckpointStorage(
-		mock<InstanceSettings>({ isLeader: false, instanceRole: 'follower' }),
 		repository,
 		mockLogger(),
 		mock<AgentsConfig>({ checkpointTtlSeconds: 60 }),
-		mock<ModuleRegistry>({ isActive: vi.fn().mockReturnValue(true) }),
 	);
 
 	return { service, repository };

packages/cli/src/modules/agents/integrations/n8n-checkpoint-storage.ts

@@ -3,14 +3,11 @@ import {
 	type CheckpointStore,
 	type SerializableAgentState,
 } from '@n8n/agents';
-import { Logger, ModuleRegistry } from '@n8n/backend-common';
+import { Logger } from '@n8n/backend-common';
 import { AgentsConfig } from '@n8n/config';
 import { Time } from '@n8n/constants';
-import { OnLeaderStepdown, OnLeaderTakeover, OnShutdown } from '@n8n/decorators';
 import { Service } from '@n8n/di';
-import { InstanceSettings } from 'n8n-core';
 import { jsonParse, UnexpectedError, UserError } from 'n8n-workflow';
-import { strict } from 'node:assert';
 
 import {
 	decodeAgentSandboxHostMetadata,
@@ -46,21 +43,12 @@ export const CHECKPOINT_RECONCILIATION_OVERFLOW = Symbol('checkpoint-reconciliat
 
 @Service()
 export class N8NCheckpointStorage {
-	private pruneTimeout: NodeJS.Timeout | undefined;
-
-	private isStopping = false;
-
-	private isInitialized = false;
-
 	constructor(
-		private readonly instanceSettings: InstanceSettings,
 		private readonly agentCheckpointRepository: AgentCheckpointRepository,
 		private readonly logger: Logger,
 		private readonly agentsConfig: AgentsConfig,
-		private readonly moduleRegistry: ModuleRegistry,
 	) {
 		this.logger = this.logger.scoped('agents');
-		this.isInitialized = this.moduleRegistry.isActive('agents');
 	}
 
 	getStorage(agentId: string): CheckpointStore {
@@ -100,12 +88,6 @@ export class N8NCheckpointStorage {
 		return runIds;
 	}
 
-	init() {
-		strict(this.instanceSettings.instanceRole !== 'unset', 'Instance role is not set');
-
-		if (this.instanceSettings.isLeader) this.startPruning();
-	}
-
 	async save(key: string, checkpointState: SerializableAgentState, agentId: string): Promise<void> {
 		const state = stripStateFileData(checkpointState);
 		const existing = await this.agentCheckpointRepository.findByRunId(key);
@@ -219,46 +201,16 @@ export class N8NCheckpointStorage {
 		await this.agentCheckpointRepository.expireByRunIdAndAgentId(key, agentId);
 	}
 
-	@OnLeaderTakeover()
-	startPruning() {
-		this.isStopping = false;
-		this.scheduleNextPrune(0);
-	}
-
-	@OnLeaderStepdown()
-	stopPruning() {
-		clearTimeout(this.pruneTimeout);
-		this.pruneTimeout = undefined;
-	}
-
-	@OnShutdown()
-	shutdown() {
-		this.isStopping = true;
-		this.stopPruning();
-	}
-
-	private scheduleNextPrune(delayMs = Time.hours.toMilliseconds) {
-		if (this.isStopping || !this.isInitialized) return;
-		this.pruneTimeout = setTimeout(async () => {
-			await this.pruneStaleSuspensions();
-		}, delayMs);
-	}
-
-	private async pruneStaleSuspensions() {
+	/** Marks checkpoints past their TTL as expired. A failure propagates to the caller. */
+	async pruneStaleSuspensions() {
 		const ttlMs = this.agentsConfig.checkpointTtlSeconds * Time.seconds.toMilliseconds;
 		const cutoffDate = new Date(Date.now() - ttlMs);
 
-		try {
-			const count = await this.agentCheckpointRepository.markExpired(cutoffDate);
-			if (count > 0) {
-				this.logger.info('Marked stale agent checkpoints as expired', { count });
-			} else {
-				this.logger.debug('No stale agent checkpoints to expire');
-			}
-			this.scheduleNextPrune();
-		} catch (error: unknown) {
-			this.logger.warn('Failed to expire stale agent checkpoints', { error });
-			this.scheduleNextPrune(Time.seconds.toMilliseconds * 30);
+		const count = await this.agentCheckpointRepository.markExpired(cutoffDate);
+		if (count > 0) {
+			this.logger.info('Marked stale agent checkpoints as expired', { count });
+		} else {
+			this.logger.debug('No stale agent checkpoints to expire');
 		}
 	}
 }

packages/cli/src/modules/insights/insights-compaction.task.ts

@@ -1,10 +1,9 @@
 import { Time } from '@n8n/constants';
-import { SystemTask } from '@n8n/decorators';
+import { SystemTask, wholeSeconds } from '@n8n/decorators';
 import type { SystemTaskEffects, SystemTaskSchedule } from '@n8n/decorators';
 
 import { InsightsCompactionService } from './insights-compaction.service';
 import { InsightsConfig } from './insights.config';
-import { wholeSeconds } from './whole-seconds';
 
 /**
  * Rolls raw insight events up into per-period summaries, so the insights

packages/cli/src/modules/insights/insights-pruning.task.ts

@@ -1,10 +1,9 @@
 import { Time } from '@n8n/constants';
-import { SystemTask } from '@n8n/decorators';
+import { SystemTask, wholeSeconds } from '@n8n/decorators';
 import type { SystemTaskEffects, SystemTaskSchedule } from '@n8n/decorators';
 
 import { InsightsPruningService } from './insights-pruning.service';
 import { InsightsConfig } from './insights.config';
-import { wholeSeconds } from './whole-seconds';
 
 /**
  * Deletes insights statistics older than the configured retention period, so

packages/cli/src/modules/instance-ai/__tests__/instance-ai-checkpoint-pruning.task.test.ts

@@ -0,0 +1,48 @@
+import type { InstanceAiConfig } from '@n8n/config';
+import type { IntervalSchedule } from '@n8n/scheduler';
+import { validateSchedule } from '@n8n/scheduler';
+import { mock } from 'vitest-mock-extended';
+
+import { InstanceAiCheckpointPruningTask } from '../instance-ai-checkpoint-pruning.task';
+import type { InstanceAiService } from '../instance-ai.service';
+
+describe('InstanceAiCheckpointPruningTask', () => {
+	const config = mock<InstanceAiConfig>({ pruneInterval: 60 * 60 * 1000 });
+	const instanceAiService = mock<InstanceAiService>();
+	const task = new InstanceAiCheckpointPruningTask(config, instanceAiService);
+
+	it('should declare the configured prune cadence', () => {
+		expect(task.name).toBe('instance-ai-checkpoint-pruning');
+		expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: 3600 });
+		expect(task.effects).toBe('idempotent');
+		expect(task.durable).toBe(false);
+		expect(task.runOnTakeover).toBe(true);
+		expect(task.retryDelaySeconds).toBe(30);
+	});
+
+	it.each([
+		{ pruneInterval: 1_500, expected: 2 },
+		{ pruneInterval: 500, expected: 1 },
+		{ pruneInterval: 1, expected: 1 },
+	])(
+		'should schedule a whole number of seconds for a prune interval of $pruneInterval ms',
+		({ pruneInterval, expected }) => {
+			const task = new InstanceAiCheckpointPruningTask(
+				mock<InstanceAiConfig>({ pruneInterval }),
+				instanceAiService,
+			);
+
+			expect(task.schedule).toEqual({ kind: 'interval', intervalSeconds: expected });
+			expect(() => validateSchedule(task.schedule as IntervalSchedule)).not.toThrow();
+		},
+	);
+
+	it('should prune expired data on run and pass the signal through', async () => {
+		const { signal } = new AbortController();
+
+		await task.run(signal);
+
+		expect(instanceAiService.pruneExpiredData).toHaveBeenCalledTimes(1);
+		expect(instanceAiService.pruneExpiredData).toHaveBeenCalledWith(expect.any(Number), signal);
+	});
+});

packages/cli/src/modules/instance-ai/__tests__/instance-ai-memory.service.test.ts

@@ -1280,6 +1280,33 @@ describe('InstanceAiMemoryService.cleanupExpiredThreads', () => {
 
 		dateNow.mockRestore();
 	});
+
+	it('stops before the next thread once the signal is aborted', async () => {
+		const dateNow = vi
+			.spyOn(Date, 'now')
+			.mockReturnValue(new Date('2026-05-15T00:00:00.000Z').getTime());
+		const first = makeThread('expired-1', '2026-05-01T00:00:00.000Z');
+		const second = makeThread('expired-2', '2026-05-02T00:00:00.000Z');
+		const controller = new AbortController();
+
+		mockListThreads.mockResolvedValue({
+			threads: [first, second],
+			total: 2,
+			page: 0,
+			hasMore: false,
+		});
+		mockDeleteThread.mockImplementation(async () => controller.abort());
+
+		const service = createService({ threadTtlDays: 7 });
+		const deletedCount = await service.cleanupExpiredThreads(undefined, controller.signal);
+
+		expect(deletedCount).toBe(1);
+		expect(mockDeleteThread).toHaveBeenCalledTimes(1);
+		expect(mockDeleteThread).toHaveBeenCalledWith(first.id);
+		expect(mockListThreads).toHaveBeenCalledTimes(1);
+
+		dateNow.mockRestore();
+	});
 });
 
 describe('bindAgentBuilderTarget', () => {

packages/cli/src/modules/instance-ai/__tests__/instance-ai.service.test.ts

@@ -294,22 +294,16 @@ function createStartRunService(): StartRunServiceInternals {
 }
 
 type CheckpointPruneServiceInternals = {
-	startCheckpointPruning: () => void;
-	stopCheckpointPruning: () => void;
-	runScheduledPrune: (now?: number) => Promise<void>;
+	pruneExpiredData: (now?: number, signal?: AbortSignal) => Promise<void>;
 	suspendedThreads: {
 		pruneStalePendingConfirmations: MockedFunction<(now: number) => Promise<void>>;
 	};
-	pruneExpiredThreads: MockedFunction<() => Promise<void>>;
-	scheduleCheckpointPrune: MockedFunction<(delayMs?: number) => void>;
+	pruneExpiredThreads: MockedFunction<(signal?: AbortSignal) => Promise<void>>;
 	checkpointStore: {
 		markExpiredOlderThan: MockedFunction<(olderThan: Date) => Promise<number>>;
 		hardDeleteExpiredOlderThan: MockedFunction<(olderThan: Date) => Promise<number>>;
 	};
-	checkpointPruneTimer?: NodeJS.Timeout;
-	checkpointPruningStopped: boolean;
 	instanceAiConfig: {
-		pruneInterval: number;
 		snapshotRetention: number;
 		checkpointGcRetention: number;
 	};
@@ -320,7 +314,6 @@ function createCheckpointPruneService(): CheckpointPruneServiceInternals {
 	const service = Object.create(
 		InstanceAiService.prototype,
 	) as unknown as CheckpointPruneServiceInternals;
-	service.scheduleCheckpointPrune = vi.fn();
 	service.suspendedThreads = {
 		pruneStalePendingConfirmations: vi.fn(async (_now: number) => undefined),
 	};
@@ -329,9 +322,7 @@ function createCheckpointPruneService(): CheckpointPruneServiceInternals {
 		markExpiredOlderThan: vi.fn(async (_olderThan: Date) => 0),
 		hardDeleteExpiredOlderThan: vi.fn(async (_olderThan: Date) => 0),
 	};
-	service.checkpointPruningStopped = true;
 	service.instanceAiConfig = {
-		pruneInterval: 60 * 60 * 1000,
 		snapshotRetention: 24 * 60 * 60 * 1000,
 		checkpointGcRetention: 7 * 24 * 60 * 60 * 1000,
 	};
@@ -398,7 +389,6 @@ function createInstanceAiErrorReporterMock() {
 
 type ShutdownServiceInternals = {
 	shutdown: () => Promise<void>;
-	stopCheckpointPruning: MockedFunction<() => void>;
 	liveness: { shutdown: MockedFunction<() => void> };
 	runState: {
 		shutdown: MockedFunction<
@@ -1077,7 +1067,6 @@ describe('InstanceAiService — shutdown', () => {
 		const service = Object.create(
 			InstanceAiService.prototype,
 		) as unknown as ShutdownServiceInternals;
-		service.stopCheckpointPruning = vi.fn();
 		service.liveness = { shutdown: vi.fn() };
 		service.runState = {
 			shutdown: vi.fn(() => ({ activeRuns: [], suspendedRuns: [] })),
@@ -1343,12 +1332,12 @@ describe('InstanceAiService — run start', () => {
 	});
 });
 
-describe('InstanceAiService — scheduled pruning', () => {
+describe('InstanceAiService — expired data pruning', () => {
 	it('marks checkpoints expired older than the retention window', async () => {
 		const service = createCheckpointPruneService();
 		const now = new Date('2026-05-13T12:00:00.000Z').getTime();
 
-		await service.runScheduledPrune(now);
+		await service.pruneExpiredData(now);
 
 		// snapshotRetention = 24h → tombstone anything untouched since 05-12
 		expect(service.checkpointStore.markExpiredOlderThan).toHaveBeenCalledWith(
@@ -1360,59 +1349,98 @@ describe('InstanceAiService — scheduled pruning', () => {
 		);
 		expect(service.suspendedThreads.pruneStalePendingConfirmations).toHaveBeenCalledWith(now);
 		expect(service.pruneExpiredThreads).toHaveBeenCalled();
-		expect(service.scheduleCheckpointPrune).toHaveBeenCalledWith();
+	});
+
+	it('passes the signal to the thread sweep', async () => {
+		const service = createCheckpointPruneService();
+		const { signal } = new AbortController();
+
+		await service.pruneExpiredData(new Date('2026-05-13T12:00:00.000Z').getTime(), signal);
+
+		expect(service.pruneExpiredThreads).toHaveBeenCalledWith(signal);
+	});
+
+	it('stops before the next step once the signal is aborted', async () => {
+		const service = createCheckpointPruneService();
+		const controller = new AbortController();
+		service.checkpointStore.markExpiredOlderThan.mockImplementation(async () => {
+			controller.abort();
+			return 0;
+		});
+
+		await service.pruneExpiredData(
+			new Date('2026-05-13T12:00:00.000Z').getTime(),
+			controller.signal,
+		);
+
+		expect(service.checkpointStore.hardDeleteExpiredOlderThan).not.toHaveBeenCalled();
+		expect(service.suspendedThreads.pruneStalePendingConfirmations).not.toHaveBeenCalled();
+		expect(service.pruneExpiredThreads).not.toHaveBeenCalled();
+		expect(service.logger.debug).toHaveBeenCalledWith(
+			'Stopped the Instance AI prune pass early because the run was aborted',
+		);
+	});
+
+	it('logs an early stop when the signal aborts during the thread sweep', async () => {
+		const service = createCheckpointPruneService();
+		const controller = new AbortController();
+		service.pruneExpiredThreads.mockImplementation(async () => controller.abort());
+
+		await service.pruneExpiredData(
+			new Date('2026-05-13T12:00:00.000Z').getTime(),
+			controller.signal,
+		);
+
+		expect(service.pruneExpiredThreads).toHaveBeenCalledTimes(1);
+		expect(service.logger.debug).toHaveBeenCalledWith(
+			'Stopped the Instance AI prune pass early because the run was aborted',
+		);
 	});
 
 	it('skips hard-deleting tombstones when the GC retention is disabled', async () => {
 		const service = createCheckpointPruneService();
 		service.instanceAiConfig.checkpointGcRetention = 0;
 
-		await service.runScheduledPrune(new Date('2026-05-13T12:00:00.000Z').getTime());
+		await service.pruneExpiredData(new Date('2026-05-13T12:00:00.000Z').getTime());
 
 		expect(service.checkpointStore.hardDeleteExpiredOlderThan).not.toHaveBeenCalled();
-		// The rest of the cycle still runs.
+		// The rest of the pass still runs.
 		expect(service.checkpointStore.markExpiredOlderThan).toHaveBeenCalled();
-		expect(service.scheduleCheckpointPrune).toHaveBeenCalledWith();
+	});
+
+	it('propagates a checkpoint expiry failure to the caller', async () => {
+		const service = createCheckpointPruneService();
+		service.checkpointStore.markExpiredOlderThan.mockRejectedValueOnce(new Error('db down'));
+
+		await expect(
+			service.pruneExpiredData(new Date('2026-05-13T12:00:00.000Z').getTime()),
+		).rejects.toThrow('db down');
+
+		expect(service.suspendedThreads.pruneStalePendingConfirmations).not.toHaveBeenCalled();
 	});
 
 	it('continues the prune cycle when hard-deleting tombstones fails', async () => {
 		const service = createCheckpointPruneService();
 		service.checkpointStore.hardDeleteExpiredOlderThan.mockRejectedValueOnce(new Error('db down'));
 
-		await service.runScheduledPrune(new Date('2026-05-13T12:00:00.000Z').getTime());
+		await service.pruneExpiredData(new Date('2026-05-13T12:00:00.000Z').getTime());
 
-		// A GC failure is swallowed and never forces the short retry cadence.
+		// A GC failure is swallowed and never interrupts the rest of the pass.
 		expect(service.suspendedThreads.pruneStalePendingConfirmations).toHaveBeenCalled();
 		expect(service.pruneExpiredThreads).toHaveBeenCalled();
-		expect(service.scheduleCheckpointPrune).toHaveBeenCalledWith();
 		expect(service.logger.warn).toHaveBeenCalled();
 	});
-
-	it('starts checkpoint pruning when configured', () => {
-		const service = createCheckpointPruneService();
-
-		service.startCheckpointPruning();
-
-		expect(service.checkpointPruningStopped).toBe(false);
-		expect(service.scheduleCheckpointPrune).toHaveBeenCalledWith(0);
-	});
-
-	it('does not start checkpoint pruning when disabled', () => {
-		const service = createCheckpointPruneService();
-		service.instanceAiConfig.pruneInterval = 0;
-
-		service.startCheckpointPruning();
-
-		expect(service.scheduleCheckpointPrune).not.toHaveBeenCalled();
-	});
 });
 
 type ExpiredThreadPruneServiceInternals = {
-	pruneExpiredThreads: () => Promise<void>;
+	pruneExpiredThreads: (signal?: AbortSignal) => Promise<void>;
 	clearThreadState: MockedFunction<(threadId: string) => Promise<void>>;
 	memoryService: {
 		cleanupExpiredThreads: MockedFunction<
-			(onThreadDeleted?: (threadId: string) => Promise<void>) => Promise<number>
+			(
+				onThreadDeleted?: (threadId: string) => Promise<void>,
+				signal?: AbortSignal,
+			) => Promise<number>
 		>;
 	};
 	logger: { warn: Mock };
@@ -1444,6 +1472,18 @@ describe('InstanceAiService — expired thread pruning', () => {
 		expect(service.clearThreadState).toHaveBeenCalledWith('thread-1');
 	});
 
+	it('passes the signal to the memory service', async () => {
+		const service = createExpiredThreadPruneService();
+		const { signal } = new AbortController();
+
+		await service.pruneExpiredThreads(signal);
+
+		expect(service.memoryService.cleanupExpiredThreads).toHaveBeenCalledWith(
+			expect.any(Function),
+			signal,
+		);
+	});
+
 	it('swallows errors so the recurring prune is not disrupted', async () => {
 		const service = createExpiredThreadPruneService();
 		service.memoryService.cleanupExpiredThreads.mockRejectedValueOnce(new Error('db down'));

packages/cli/src/modules/instance-ai/instance-ai-checkpoint-pruning.task.ts

@@ -0,0 +1,40 @@
+import { InstanceAiConfig } from '@n8n/config';
+import { Time } from '@n8n/constants';
+import { SystemTask, wholeSeconds } from '@n8n/decorators';
+import type { SystemTaskEffects, SystemTaskSchedule } from '@n8n/decorators';
+
+import { InstanceAiService } from './instance-ai.service';
+
+/**
+ * Expires stale Instance AI checkpoints, hard-deletes their tombstones past
+ * the GC horizon, and drops expired pending confirmations and conversation
+ * threads, so suspended-run state does not pile up forever.
+ */
+@SystemTask()
+export class InstanceAiCheckpointPruningTask implements SystemTask {
+	readonly name = 'instance-ai-checkpoint-pruning';
+
+	readonly schedule: SystemTaskSchedule = {
+		kind: 'interval',
+		intervalSeconds: wholeSeconds(
+			this.instanceAiConfig.pruneInterval * Time.milliseconds.toSeconds,
+		),
+	};
+
+	readonly effects: SystemTaskEffects = 'idempotent';
+
+	readonly durable = false;
+
+	readonly runOnTakeover = true;
+
+	readonly retryDelaySeconds = 30;
+
+	constructor(
+		private readonly instanceAiConfig: InstanceAiConfig,
+		private readonly instanceAiService: InstanceAiService,
+	) {}
+
+	async run(signal: AbortSignal): Promise<void> {
+		await this.instanceAiService.pruneExpiredData(Date.now(), signal);
+	}
+}

packages/cli/src/modules/instance-ai/instance-ai-memory.service.ts

@@ -633,10 +633,12 @@ export class InstanceAiMemoryService {
 	/**
 	 * Delete conversation threads older than the configured TTL. Invoked on a
 	 * recurring schedule by the leader instance's prune job. Idempotent and
-	 * safe to call repeatedly — no-op if threadTtlDays is 0 (disabled).
+	 * safe to call repeatedly — no-op if threadTtlDays is 0 (disabled). Stops
+	 * before the next thread once `signal` aborts.
 	 */
 	async cleanupExpiredThreads(
 		onThreadDeleted?: (threadId: string) => Promise<void>,
+		signal?: AbortSignal,
 	): Promise<number> {
 		const ttlDays = this.instanceAiConfig.threadTtlDays;
 		if (!ttlDays || ttlDays <= 0) return 0;
@@ -650,7 +652,7 @@ export class InstanceAiMemoryService {
 		const perPage = 100;
 		let hasMore = true;
 
-		while (hasMore) {
+		while (hasMore && !signal?.aborted) {
 			const result = await this.agentMemory.listThreads({
 				perPage,
 				page: 0,
@@ -658,6 +660,7 @@ export class InstanceAiMemoryService {
 			});
 			let deletedInPage = 0;
 			for (const thread of result.threads) {
+				if (signal?.aborted) break;
 				if (thread.updatedAt < cutoff) {
 					try {
 						await onThreadDeleted?.(thread.id);

packages/cli/src/modules/instance-ai/instance-ai.module.ts

@@ -53,6 +53,16 @@ export class InstanceAiModule implements ModuleInterface {
 		}
 	}
 
+	async systemTasks() {
+		const { InstanceAiConfig } = await import('@n8n/config');
+		if (Container.get(InstanceAiConfig).pruneInterval <= 0) return [];
+
+		const { InstanceAiCheckpointPruningTask } = await import(
+			'./instance-ai-checkpoint-pruning.task.js'
+		);
+		return [InstanceAiCheckpointPruningTask];
+	}
+
 	async settings() {
 		const { GlobalConfig } = await import('@n8n/config');
 		const { InstanceAiService } = await import('./instance-ai.service.js');

packages/cli/src/modules/instance-ai/instance-ai.service.ts

@@ -30,7 +30,7 @@ import { Logger, ModuleRegistry } from '@n8n/backend-common';
 import { SsrfProtectionService } from '@n8n/backend-network';
 import { GlobalConfig, SsrfProtectionConfig, type InstanceAiConfig } from '@n8n/config';
 import { UserRepository, type User } from '@n8n/db';
-import { OnLeaderStepdown, OnLeaderTakeover, OnPubSubEvent } from '@n8n/decorators';
+import { OnPubSubEvent } from '@n8n/decorators';
 import { Container, Service } from '@n8n/di';
 import {
 	MAX_STEPS,
@@ -385,7 +385,6 @@ function isTelemetryConfigurableAgent(
 	);
 }
 
-const INSTANCE_AI_CHECKPOINT_PRUNE_RETRY_MS = 30 * 1000;
 const WORKFLOW_SETUP_ROUTING_CLAIM_TTL_MS = 15 * 60 * 1000;
 
 const CONFIRMATION_EXPIRED_MESSAGE =
@@ -772,10 +771,6 @@ export class InstanceAiService {
 
 	private readonly taskProjector: WorkflowVerificationTaskProjector;
 
-	private checkpointPruneTimer: NodeJS.Timeout | undefined;
-
-	private checkpointPruningStopped = true;
-
 	/**
 	 * In-flight `executeRun` / `processResumedStream` promises. Tracked so
 	 * `shutdown()` can drain them before n8n closes the DB connection — the
@@ -940,7 +935,6 @@ export class InstanceAiService {
 		});
 
 		this.liveness.start();
-		if (this.instanceSettings.isLeader) this.startCheckpointPruning();
 	}
 
 	private async createProxyRunConfig(user: User): Promise<{
@@ -1891,7 +1885,6 @@ export class InstanceAiService {
 	}
 
 	async shutdown(): Promise<void> {
-		this.stopCheckpointPruning();
 		this.liveness.shutdown();
 
 		const { activeRuns, suspendedRuns, pendingThreadIds } = this.runState.shutdown();
@@ -1990,28 +1983,6 @@ export class InstanceAiService {
 		this.logger.debug('Instance AI service shut down');
 	}
 
-	@OnLeaderTakeover()
-	startCheckpointPruning(): void {
-		if (this.checkpointPruneTimer || this.instanceAiConfig.pruneInterval <= 0) return;
-		this.checkpointPruningStopped = false;
-		this.scheduleCheckpointPrune(0);
-	}
-
-	@OnLeaderStepdown()
-	stopCheckpointPruning(): void {
-		this.checkpointPruningStopped = true;
-		clearTimeout(this.checkpointPruneTimer);
-		this.checkpointPruneTimer = undefined;
-	}
-
-	private scheduleCheckpointPrune(delayMs = this.instanceAiConfig.pruneInterval): void {
-		if (this.checkpointPruningStopped) return;
-		this.checkpointPruneTimer = setTimeout(() => {
-			void this.runScheduledPrune();
-		}, delayMs);
-		this.checkpointPruneTimer.unref();
-	}
-
 	/**
 	 * Track a fire-and-forget run so `shutdown()` can wait for its cleanup
 	 * (finally block + SDK `cleanupRun`) to finish before the DB closes.
@@ -2074,32 +2045,26 @@ export class InstanceAiService {
 	}
 
 	/**
-	 * One tick of the recurring leader prune cycle: expire stale checkpoints,
-	 * hard-delete tombstones past the GC horizon, drop expired pending
-	 * confirmations, and delete expired conversation threads, then schedule the
-	 * next run. A checkpoint failure reschedules with a shorter retry delay; the
-	 * confirmation and thread steps swallow their own errors so they never
-	 * disrupt the cycle.
+	 * One prune pass: expire stale checkpoints, hard-delete tombstones past the
+	 * GC horizon, drop expired pending confirmations, and delete expired
+	 * conversation threads. A checkpoint failure propagates to the caller; the
+	 * GC, confirmation, and thread steps swallow their own errors. Stops before
+	 * the next step once `signal` aborts.
 	 */
-	private async runScheduledPrune(now = Date.now()): Promise<void> {
+	async pruneExpiredData(now = Date.now(), signal?: AbortSignal): Promise<void> {
 		const olderThan = new Date(now - this.instanceAiConfig.snapshotRetention);
 
-		try {
-			const count = await this.checkpointStore.markExpiredOlderThan(olderThan);
-			if (count > 0) {
-				this.logger.info('Expired stale Instance AI checkpoints', { count });
-			} else {
-				this.logger.debug('No stale Instance AI checkpoints to expire');
-			}
-			await this.hardDeleteExpiredCheckpoints(now);
-			await this.suspendedThreads.pruneStalePendingConfirmations(now);
-			await this.pruneExpiredThreads();
-			this.scheduleCheckpointPrune();
-		} catch (error: unknown) {
-			this.logger.warn('Failed to expire stale Instance AI checkpoints', {
-				error: getErrorMessage(error),
-			});
-			this.scheduleCheckpointPrune(INSTANCE_AI_CHECKPOINT_PRUNE_RETRY_MS);
+		const count = await this.checkpointStore.markExpiredOlderThan(olderThan);
+		if (count > 0) {
+			this.logger.info('Expired stale Instance AI checkpoints', { count });
+		} else {
+			this.logger.debug('No stale Instance AI checkpoints to expire');
+		}
+		if (!signal?.aborted) await this.hardDeleteExpiredCheckpoints(now);
+		if (!signal?.aborted) await this.suspendedThreads.pruneStalePendingConfirmations(now);
+		if (!signal?.aborted) await this.pruneExpiredThreads(signal);
+		if (signal?.aborted) {
+			this.logger.debug('Stopped the Instance AI prune pass early because the run was aborted');
 		}
 	}
 
@@ -2128,16 +2093,12 @@ export class InstanceAiService {
 		}
 	}
 
-	/**
-	 * Delete conversation threads older than the configured TTL as part of the
-	 * recurring leader prune. Has its own try/catch so a failure here never
-	 * disrupts checkpoint pruning or the next scheduled run. No-op when
-	 * `threadTtlDays` is 0 (handled inside `cleanupExpiredThreads`).
-	 */
-	private async pruneExpiredThreads(): Promise<void> {
+	/** Deletes conversation threads past their TTL and logs instead of throwing on failure. */
+	private async pruneExpiredThreads(signal?: AbortSignal): Promise<void> {
 		try {
 			await this.memoryService.cleanupExpiredThreads(
 				async (threadId) => await this.clearThreadState(threadId),
+				signal,
 			);
 		} catch (error: unknown) {
 			this.logger.warn('Failed to clean up expired Instance AI conversation threads', {

packages/cli/test/integration/commands/start.cmd.test.ts

@@ -26,7 +26,6 @@ import { ActiveWorkflowManager } from '@/active-workflow-manager';
 import { Start } from '@/commands/start';
 import { EnqueuedExecutionRecoveryService } from '@/executions/enqueued-execution-recovery.service';
 import { ExternalHooks } from '@/external-hooks';
-import { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage';
 import { Push } from '@/push';
 import { PubSubEventBus } from '@/scaling/pubsub/pubsub.eventbus';
 import { DurableScheduler } from '@/scheduling/durable-scheduler';
@@ -55,7 +54,6 @@ mockInstance(ExternalHooks);
 mockInstance(ExecutionsPruningService);
 mockInstance(WorkflowHistoryCompactionService);
 mockInstance(WorkflowStatisticsRollupService);
-mockInstance(N8NCheckpointStorage);
 mockInstance(DurableScheduler);
 // Also proves `run()`'s dynamic `.js` import resolves to the same module as this
 // static one - otherwise the mock would silently not apply and the real recovery