feat(core): Schedule agent tasks via the durable scheduler behind a flag (no-changelog) (#37849)

Co-authored-by: Claude Fable 5.1 <[email protected]>
Co-authored-by: Robin Braumann <[email protected]>
Co-authored-by: Cursor <[email protected]>

.gitignore

@@ -83,7 +83,7 @@ packages/cli/src/commands/export/outputs
 .claude/plans/
 .claude/worktrees/
 .claude/specs/
-.cursor/plans/
+.cursor/
 .superset
 .conductor
 .n8n

packages/@n8n/config/src/configs/scheduler.config.ts

@@ -317,6 +317,27 @@ export class SchedulerConfig {
 	@Env('N8N_SCHEDULER_SYSTEM_TASKS_ENABLED')
 	enabledForSystemTasks: boolean = false;
 
+	/**
+	 * Whether the durable scheduler runs agent scheduled tasks instead of the
+	 * in-process timer of n8n. Off by default. Requires {@link enabled} to also
+	 * be on.
+	 *
+	 * In a multi-instance setup, turn this on only when every instance runs a
+	 * version of n8n that supports it. While older and newer versions run side by
+	 * side (for example during a rolling deploy), the older leader still runs
+	 * these tasks on its own timer. Then the same task can run twice.
+	 *
+	 * Give every main the same value. A main that starts with this flag off
+	 * removes the durable agent-task jobs, also the ones that a main with the
+	 * flag on has created.
+	 *
+	 * The two schedulers differ on a DST spring-forward: a time inside the
+	 * skipped hour runs at the moment the clock jumps on the in-process timer,
+	 * and shifts forward by the skipped hour on the durable scheduler.
+	 */
+	@Env('N8N_SCHEDULER_AGENT_TASKS_ENABLED')
+	enabledForAgentTasks: boolean = false;
+
 	/**
 	 * Temporary escape hatch for the durable-scheduler rollout (preview to GA).
 	 * Off by default; intended to be removed once the durable scheduler is GA.

packages/@n8n/config/test/config.test.ts

@@ -499,6 +499,7 @@ describe('GlobalConfig', () => {
 			misfireGraceSeconds: 60,
 			durableCursorsEnabled: false,
 			enabledForSystemTasks: false,
+			enabledForAgentTasks: false,
 			ownerReconciliationEnabled: true,
 			ownerReconciliationIntervalSeconds: 900,
 			ownerReconciliationTimeoutSeconds: 300,

packages/@n8n/db/src/repositories/scheduled-job.repository.ts

@@ -134,6 +134,21 @@ export class ScheduledJobRepository extends Repository<ScheduledJob> {
 		return await manager.findBy(ScheduledJob, { id: In(ids) });
 	}
 
+	/**
+	 * The member ids under which an owner holds jobs of one task type, each once.
+	 * A caller uses them to tell which of its parts still provision a job.
+	 */
+	async findOwnerMemberIds(owner: ScheduledJobOwnerRef, taskType: string): Promise<string[]> {
+		const rows = await this.find({
+			where: { ...ownerRefCriteria(owner), taskType, ownerMemberId: Not(IsNull()) },
+			select: { ownerMemberId: true },
+		});
+		const memberIds = rows.flatMap((row) =>
+			row.ownerMemberId === null ? [] : [row.ownerMemberId],
+		);
+		return [...new Set(memberIds)];
+	}
+
 	async countByOwner(owner: ScheduledJobOwner): Promise<number> {
 		return await this.count({ where: ownerCriteria(owner) });
 	}

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

@@ -22,6 +22,7 @@ import type {
 import type { AgentTaskSnapshotRepository } from '../repositories/agent-task-snapshot.repository';
 import type { AgentTaskRepository } from '../repositories/agent-task.repository';
 import type { AgentRepository } from '../repositories/agent.repository';
+import type { AgentTaskJobRegistrar } from '../scheduling/agent-task-job-registrar';
 
 const AGENT_ID = 'agent-1';
 const PROJECT_ID = 'project-1';
@@ -74,6 +75,11 @@ function makePublishedAgent(
 	} as Partial<Agent>);
 }
 
+/** The common fixture: a published agent whose only task is `task-1`, enabled. */
+function publishedAgentWithTask(): Agent {
+	return makePublishedAgent([{ id: 'task-1', enabled: true }]);
+}
+
 function makeSnapshot(overrides: Partial<AgentTaskSnapshot> = {}): AgentTaskSnapshot {
 	return {
 		versionId: 'ver-1',
@@ -106,27 +112,6 @@ async function* throwingStream(): AsyncGenerator<never> {
 	throw new Error('execution failed');
 }
 
-async function runTaskOf(
-	service: AgentTaskService,
-	agentId: string,
-	taskId: string,
-): Promise<void> {
-	await (service as unknown as { runTask(agentId: string, taskId: string): Promise<void> }).runTask(
-		agentId,
-		taskId,
-	);
-}
-
-async function runScheduledTaskOf(
-	service: AgentTaskService,
-	agentId: string,
-	taskId: string,
-): Promise<void> {
-	await (
-		service as unknown as { runScheduledTask(agentId: string, taskId: string): Promise<void> }
-	).runScheduledTask(agentId, taskId);
-}
-
 async function flushAsyncWork(): Promise<void> {
 	await new Promise((resolve) => setImmediate(resolve));
 }
@@ -145,6 +130,7 @@ describe('AgentTaskService', () => {
 	let agentTaskScheduler: ReturnType<typeof mock<ScheduledTaskManager>>;
 	let publisher: ReturnType<typeof mock<Publisher>>;
 	let modificationTelemetry: ReturnType<typeof mock<AgentModificationTelemetryService>>;
+	let durableJobRegistrar: ReturnType<typeof mock<AgentTaskJobRegistrar>>;
 	let txManager: { save: Mock; remove: Mock };
 	let service: AgentTaskService;
 
@@ -166,6 +152,7 @@ describe('AgentTaskService', () => {
 			agentTaskScheduler,
 			publisher,
 			modificationTelemetry,
+			durableJobRegistrar,
 		);
 	}
 
@@ -203,6 +190,9 @@ describe('AgentTaskService', () => {
 		publisher = mock<Publisher>();
 		publisher.publishCommand.mockResolvedValue(undefined);
 		modificationTelemetry = mock<AgentModificationTelemetryService>();
+		durableJobRegistrar = mock<AgentTaskJobRegistrar>();
+		// Legacy scheduling is the default under test. Durable-path cases flip this.
+		durableJobRegistrar.isEnabled.mockReturnValue(false);
 		// Default to the leader so existing registration assertions hold.
 		service = buildService(true);
 	});
@@ -625,9 +615,7 @@ describe('AgentTaskService', () => {
 
 	describe('registerEnabledForAgent', () => {
 		it('registers a cron job for each enabled published task snapshot', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(
-				makePublishedAgent([{ id: 'task-1', enabled: true }]),
-			);
+			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask());
 			(taskSnapshotRepository.findEnabledByVersionId as Mock).mockResolvedValue([makeSnapshot()]);
 
 			await service.registerEnabledForAgent(AGENT_ID);
@@ -644,9 +632,7 @@ describe('AgentTaskService', () => {
 		});
 
 		it("registers the cron in the snapshot's own timezone", async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(
-				makePublishedAgent([{ id: 'task-1', enabled: true }]),
-			);
+			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask());
 			(taskSnapshotRepository.findEnabledByVersionId as Mock).mockResolvedValue([
 				makeSnapshot({ cronExpression: '0 8 * * 5', timezone: 'Europe/London' }),
 			]);
@@ -665,9 +651,7 @@ describe('AgentTaskService', () => {
 		});
 
 		it('falls back to the instance timezone when the stored timezone is unknown', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(
-				makePublishedAgent([{ id: 'task-1', enabled: true }]),
-			);
+			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask());
 			(taskSnapshotRepository.findEnabledByVersionId as Mock).mockResolvedValue([
 				makeSnapshot({ timezone: 'Mars/Olympus_Mons' }),
 			]);
@@ -705,9 +689,7 @@ describe('AgentTaskService', () => {
 
 		it('does not register cron jobs on a follower (leader owns the cron)', async () => {
 			const follower = buildService(false);
-			(agentRepository.findOne as Mock).mockResolvedValue(
-				makePublishedAgent([{ id: 'task-1', enabled: true }]),
-			);
+			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask());
 			(taskSnapshotRepository.findEnabledByVersionId as Mock).mockResolvedValue([makeSnapshot()]);
 
 			await follower.registerEnabledForAgent(AGENT_ID);
@@ -716,9 +698,7 @@ describe('AgentTaskService', () => {
 		});
 
 		it('deregisters tasks missing from the enabled published snapshots', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(
-				makePublishedAgent([{ id: 'task-1', enabled: true }]),
-			);
+			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask());
 			(taskSnapshotRepository.findEnabledByVersionId as Mock).mockResolvedValue([makeSnapshot()]);
 			agentTaskScheduler.getTargetIds.mockReturnValue(['task-1', 'stale-task']);
 			agentTaskScheduler.hasTarget.mockImplementation((_group, taskId) => taskId === 'stale-task');
@@ -743,9 +723,7 @@ describe('AgentTaskService', () => {
 					finishRun = resolve;
 				});
 			}
-			(agentRepository.findOne as Mock).mockResolvedValue(
-				makePublishedAgent([{ id: 'task-1', enabled: true }]),
-			);
+			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask());
 			(taskSnapshotRepository.findEnabledByVersionId as Mock).mockResolvedValue([makeSnapshot()]);
 			(taskSnapshotRepository.findByVersionAndTaskId as Mock).mockResolvedValue(makeSnapshot());
 			taskRunLockRepository.acquire
@@ -789,9 +767,7 @@ describe('AgentTaskService', () => {
 			}
 			const previousLeader = service;
 			const nextLeader = buildService(true);
-			(agentRepository.findOne as Mock).mockResolvedValue(
-				makePublishedAgent([{ id: 'task-1', enabled: true }]),
-			);
+			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask());
 			(taskSnapshotRepository.findEnabledByVersionId as Mock).mockResolvedValue([makeSnapshot()]);
 			(taskSnapshotRepository.findByVersionAndTaskId as Mock).mockResolvedValue(makeSnapshot());
 			taskRunLockRepository.acquire
@@ -842,13 +818,23 @@ describe('AgentTaskService', () => {
 
 			expect(publisher.publishCommand).not.toHaveBeenCalled();
 		});
+
+		it('routes to the durable registrar when durable scheduling is on, skipping crons and pubsub', async () => {
+			setMultiMain(true);
+			durableJobRegistrar.isEnabled.mockReturnValue(true);
+
+			await service.requestReconcile(AGENT_ID);
+
+			expect(durableJobRegistrar.reconcile).toHaveBeenCalledWith(AGENT_ID);
+			expect(agentTaskScheduler.register).not.toHaveBeenCalled();
+			expect(publisher.publishCommand).not.toHaveBeenCalled();
+			expect(agentRepository.findOne).not.toHaveBeenCalled();
+		});
 	});
 
 	describe('handleTasksChanged', () => {
 		it('reconciles the agent so the leader registers its enabled tasks', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(
-				makePublishedAgent([{ id: 'task-1', enabled: true }]),
-			);
+			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask());
 			(taskSnapshotRepository.findEnabledByVersionId as Mock).mockResolvedValue([makeSnapshot()]);
 
 			await service.handleTasksChanged({ agentId: AGENT_ID });
@@ -863,6 +849,15 @@ describe('AgentTaskService', () => {
 				expect.any(Function),
 			);
 		});
+
+		it('registers no crons when durable scheduling is on, even for a broadcast from a legacy peer', async () => {
+			durableJobRegistrar.isEnabled.mockReturnValue(true);
+
+			await service.handleTasksChanged({ agentId: AGENT_ID });
+
+			expect(agentTaskScheduler.register).not.toHaveBeenCalled();
+			expect(agentRepository.findOne).not.toHaveBeenCalled();
+		});
 	});
 
 	describe('deregisterAgentTasks', () => {
@@ -895,18 +890,21 @@ describe('AgentTaskService', () => {
 		});
 	});
 
-	describe('runTask', () => {
-		const publishedAgentWithTask = (enabled: boolean) =>
-			makePublishedAgent([{ id: 'task-1', enabled }]);
+	describe('startScheduledRun', () => {
+		/** The fire-time reads: a published agent whose `task-1` snapshot is enabled. */
+		const arrangePublishedTask = (snapshot = makeSnapshot()) => {
+			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask());
+			(taskSnapshotRepository.findByVersionAndTaskId as Mock).mockResolvedValue(snapshot);
+		};
 
 		it('runs the published agent with the objective', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask(true));
-			(taskSnapshotRepository.findByVersionAndTaskId as Mock).mockResolvedValue(makeSnapshot());
+			arrangePublishedTask();
 			(agentExecutionOrchestratorService.executeForTaskPublished as Mock).mockReturnValue(
 				emptyStream(),
 			);
 
-			await runTaskOf(service, AGENT_ID, 'task-1');
+			await expect(service.startScheduledRun(AGENT_ID, 'task-1')).resolves.toBe('started');
+			await flushAsyncWork();
 
 			expect(agentExecutionOrchestratorService.executeForTaskPublished).toHaveBeenCalledWith(
 				expect.objectContaining({
@@ -922,17 +920,13 @@ describe('AgentTaskService', () => {
 		});
 
 		it('uses the published snapshot body, not the live draft row', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(
-				makePublishedAgent([{ id: 'task-1', enabled: true }]),
-			);
-			(taskSnapshotRepository.findByVersionAndTaskId as Mock).mockResolvedValue(
-				makeSnapshot({ objective: 'Published objective' }),
-			);
+			arrangePublishedTask(makeSnapshot({ objective: 'Published objective' }));
 			(agentExecutionOrchestratorService.executeForTaskPublished as Mock).mockReturnValue(
 				emptyStream(),
 			);
 
-			await runTaskOf(service, AGENT_ID, 'task-1');
+			await service.startScheduledRun(AGENT_ID, 'task-1');
+			await flushAsyncWork();
 
 			expect(agentExecutionOrchestratorService.executeForTaskPublished).toHaveBeenCalledWith(
 				expect.objectContaining({ message: expect.stringContaining('Published objective') }),
@@ -942,17 +936,13 @@ describe('AgentTaskService', () => {
 		});
 
 		it('names the schedule timezone without moving the timestamp off the instance zone', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(
-				makePublishedAgent([{ id: 'task-1', enabled: true }]),
-			);
-			(taskSnapshotRepository.findByVersionAndTaskId as Mock).mockResolvedValue(
-				makeSnapshot({ timezone: 'Asia/Tokyo' }),
-			);
+			arrangePublishedTask(makeSnapshot({ timezone: 'Asia/Tokyo' }));
 			(agentExecutionOrchestratorService.executeForTaskPublished as Mock).mockReturnValue(
 				emptyStream(),
 			);
 
-			await runTaskOf(service, AGENT_ID, 'task-1');
+			await service.startScheduledRun(AGENT_ID, 'task-1');
+			await flushAsyncWork();
 
 			// The timestamp has to agree with the `get_environment` tool, which reports
 			// the instance zone, so the schedule's zone is named instead of substituted.
@@ -962,33 +952,37 @@ describe('AgentTaskService', () => {
 			expect(message).toContain('This task is scheduled in Asia/Tokyo.');
 		});
 
-		it('skips when the agent is unpublished', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(makeAgent({ activeVersionId: null }));
+		it.each([
+			['the agent is gone', () => (agentRepository.findOne as Mock).mockResolvedValue(null)],
+			[
+				'the agent is unpublished',
+				() =>
+					(agentRepository.findOne as Mock).mockResolvedValue(makeAgent({ activeVersionId: null })),
+			],
+			[
+				'the published task snapshot is not enabled',
+				() => arrangePublishedTask(makeSnapshot({ enabled: false })),
+			],
+		])("returns 'stale' without taking the lock when %s", async (_, arrange) => {
+			arrange();
+			agentTaskScheduler.hasTarget.mockReturnValue(true);
 
-			await runTaskOf(service, AGENT_ID, 'task-1');
-
-			expect(agentExecutionOrchestratorService.executeForTaskPublished).not.toHaveBeenCalled();
-		});
-
-		it('skips when the published task snapshot is not enabled', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask(true));
-			(taskSnapshotRepository.findByVersionAndTaskId as Mock).mockResolvedValue(
-				makeSnapshot({ enabled: false }),
-			);
-
-			await runTaskOf(service, AGENT_ID, 'task-1');
+			await expect(service.startScheduledRun(AGENT_ID, 'task-1')).resolves.toBe('stale');
 
+			expect(taskRunLockRepository.acquire).not.toHaveBeenCalled();
 			expect(agentExecutionOrchestratorService.executeForTaskPublished).not.toHaveBeenCalled();
+			// The in-memory cron of a stale task goes with it.
+			expect(agentTaskScheduler.deregisterTarget).toHaveBeenCalledWith(agentTaskGroup(), 'task-1');
 		});
 
 		it('logs when execution throws', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask(true));
-			(taskSnapshotRepository.findByVersionAndTaskId as Mock).mockResolvedValue(makeSnapshot());
+			arrangePublishedTask();
 			(agentExecutionOrchestratorService.executeForTaskPublished as Mock).mockReturnValue(
 				throwingStream(),
 			);
 
-			await runTaskOf(service, AGENT_ID, 'task-1');
+			await service.startScheduledRun(AGENT_ID, 'task-1');
+			await flushAsyncWork();
 
 			expect(taskRepository.update).not.toHaveBeenCalled();
 			expect(logger.error).toHaveBeenCalledWith(
@@ -998,14 +992,15 @@ describe('AgentTaskService', () => {
 		});
 
 		it('allows a later scheduled run after a failed execution completes', async () => {
-			(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask(true));
-			(taskSnapshotRepository.findByVersionAndTaskId as Mock).mockResolvedValue(makeSnapshot());
+			arrangePublishedTask();
 			(agentExecutionOrchestratorService.executeForTaskPublished as Mock)
 				.mockReturnValueOnce(throwingStream())
 				.mockReturnValueOnce(emptyStream());
 
-			await runScheduledTaskOf(service, AGENT_ID, 'task-1');
-			await runScheduledTaskOf(service, AGENT_ID, 'task-1');
+			await service.startScheduledRun(AGENT_ID, 'task-1');
+			await flushAsyncWork();
+			await service.startScheduledRun(AGENT_ID, 'task-1');
+			await flushAsyncWork();
 
 			expect(agentExecutionOrchestratorService.executeForTaskPublished).toHaveBeenCalledTimes(2);
 			expect(taskRunLockRepository.release).toHaveBeenCalledTimes(2);
@@ -1021,16 +1016,12 @@ describe('AgentTaskService', () => {
 						finishRun = resolve;
 					});
 				}
-				(agentRepository.findOne as Mock).mockResolvedValue(publishedAgentWithTask(true));
-				(taskSnapshotRepository.findByVersionAndTaskId as Mock).mockResolvedValue(makeSnapshot());
+				arrangePublishedTask();
 				(agentExecutionOrchestratorService.executeForTaskPublished as Mock).mockReturnValue(
 					blockingStream(),
 				);
 
-				const run = runScheduledTaskOf(service, AGENT_ID, 'task-1');
-				await Promise.resolve();
-				await Promise.resolve();
-
+				await service.startScheduledRun(AGENT_ID, 'task-1');
 				await vi.advanceTimersByTimeAsync(60_000);
 
 				expect(taskRunLockRepository.renew).toHaveBeenCalledWith(
@@ -1039,11 +1030,75 @@ describe('AgentTaskService', () => {
 				);
 
 				finishRun();
-				await run;
 			} finally {
 				vi.useRealTimers();
 			}
 		});
+
+		it('propagates a lock-acquisition error without starting a run', async () => {
+			arrangePublishedTask();
+			taskRunLockRepository.acquire.mockRejectedValue(new Error('db down'));
+
+			await expect(service.startScheduledRun(AGENT_ID, 'task-1')).rejects.toThrow('db down');
+
+			expect(agentExecutionOrchestratorService.executeForTaskPublished).not.toHaveBeenCalled();
+			expect(taskRunLockRepository.release).not.toHaveBeenCalled();
+		});
+
+		it("returns 'skipped-active' without running when the lock is held", async () => {
+			arrangePublishedTask();
+			taskRunLockRepository.acquire.mockResolvedValue(null);
+
+			await expect(service.startScheduledRun(AGENT_ID, 'task-1')).resolves.toBe('skipped-active');
+
+			expect(agentExecutionOrchestratorService.executeForTaskPublished).not.toHaveBeenCalled();
+			expect(taskRunLockRepository.release).not.toHaveBeenCalled();
+		});
+
+		it("returns 'started' before the run resolves, then releases the lock after it", async () => {
+			let finishRun!: () => void;
+			// eslint-disable-next-line require-yield
+			async function* blockingStream(): AsyncGenerator<never> {
+				await new Promise<void>((resolve) => {
+					finishRun = resolve;
+				});
+			}
+			arrangePublishedTask();
+			(agentExecutionOrchestratorService.executeForTaskPublished as Mock).mockReturnValue(
+				blockingStream(),
+			);
+
+			await expect(service.startScheduledRun(AGENT_ID, 'task-1')).resolves.toBe('started');
+			await flushAsyncWork();
+			expect(taskRunLockRepository.release).not.toHaveBeenCalled();
+
+			finishRun();
+			await flushAsyncWork();
+			expect(taskRunLockRepository.release).toHaveBeenCalledTimes(1);
+		});
+
+		it('releases the lock when the run fails', async () => {
+			arrangePublishedTask();
+			(agentExecutionOrchestratorService.executeForTaskPublished as Mock).mockReturnValue(
+				throwingStream(),
+			);
+
+			await expect(service.startScheduledRun(AGENT_ID, 'task-1')).resolves.toBe('started');
+			await flushAsyncWork();
+
+			expect(taskRunLockRepository.release).toHaveBeenCalledTimes(1);
+		});
+	});
+
+	describe('reconnectAll', () => {
+		it('rebuilds no crons on leader takeover when durable scheduling is on', async () => {
+			durableJobRegistrar.isEnabled.mockReturnValue(true);
+
+			await service.reconnectAll();
+
+			expect(agentRepository.find).not.toHaveBeenCalled();
+			expect(agentTaskScheduler.register).not.toHaveBeenCalled();
+		});
 	});
 
 	describe('stopAll', () => {

packages/cli/src/modules/agents/agent-task.service.ts

@@ -21,6 +21,8 @@ import {
 	diffAgentConfigParts,
 } from './agent-modification-telemetry.service';
 import { AgentExecutionOrchestratorService } from './agent-execution-orchestrator.service';
+import { AgentTaskJobRegistrar } from './scheduling/agent-task-job-registrar';
+import { knownTaskTimezone } from './scheduling/task-timezone';
 import { Agent } from './entities/agent.entity';
 import { AgentTask } from './entities/agent-task.entity';
 import type { AgentTaskSnapshot } from './entities/agent-task-snapshot.entity';
@@ -72,6 +74,7 @@ export class AgentTaskService {
 		private readonly scheduledTaskManager: ScheduledTaskManager,
 		private readonly publisher: Publisher,
 		private readonly modificationTelemetry: AgentModificationTelemetryService,
+		private readonly durableJobRegistrar: AgentTaskJobRegistrar,
 	) {}
 
 	// ── CRUD ──────────────────────────────────────────────────────────────
@@ -327,6 +330,13 @@ export class AgentTaskService {
 	 * harmless.
 	 */
 	async requestReconcile(agentId: string): Promise<void> {
+		// Durable scheduling is database state. The main that handles the request
+		// writes it directly. There is no cron to register and no peer to broadcast to.
+		if (this.durableJobRegistrar.isEnabled()) {
+			await this.durableJobRegistrar.reconcile(agentId);
+			return;
+		}
+
 		await this.registerEnabledForAgent(agentId);
 		this.broadcastTasksChanged(agentId);
 	}
@@ -338,6 +348,12 @@ export class AgentTaskService {
 	 * Used by the local lifecycle path and the pubsub reconcile handler.
 	 */
 	async registerEnabledForAgent(agentId: string): Promise<void> {
+		// With durable scheduling on, no main can hold in-memory crons. If the
+		// leader registers them next to the durable jobs, every task runs twice
+		// per tick. The guard is here so that it also covers the pubsub reconcile
+		// path, for example a broadcast from a peer on the legacy path.
+		if (this.durableJobRegistrar.isEnabled()) return;
+
 		const agent = await this.agentRepository.findOne({
 			where: { id: agentId },
 			relations: { activeVersion: true },
@@ -384,6 +400,10 @@ export class AgentTaskService {
 
 	@OnLeaderTakeover()
 	async reconnectAll(): Promise<void> {
+		// Durable jobs live in the database and survive leader changes. There is
+		// nothing to rebuild on takeover.
+		if (this.durableJobRegistrar.isEnabled()) return;
+
 		const agents = await this.agentRepository.find({
 			where: { activeVersionId: Not(IsNull()) },
 			relations: { activeVersion: true },
@@ -457,7 +477,13 @@ export class AgentTaskService {
 				timezone,
 			},
 			() => {
-				void this.runScheduledTask(agentId, taskId);
+				void this.startScheduledRun(agentId, taskId).catch((error: unknown) => {
+					this.logger.error('[AgentTaskService] Scheduled task lock failed', {
+						taskId,
+						agentId,
+						error: error instanceof Error ? error.message : String(error),
+					});
+				});
 			},
 		);
 		if (!registered) return;
@@ -480,34 +506,62 @@ export class AgentTaskService {
 
 	// ── Run ───────────────────────────────────────────────────────────────
 
-	private async runScheduledTask(agentId: string, taskId: string): Promise<void> {
-		const holderId = randomUUID();
-		let lock: AgentTaskRunLockHandle | null = null;
-		let renewInterval: ReturnType<typeof setInterval> | undefined;
-		try {
-			lock = await this.taskRunLockRepository.acquire(agentId, taskId, {
-				holderId,
-				ttlMs: TASK_RUN_LOCK_TTL_MS,
-			});
-			if (!lock) {
-				this.logger.info('[AgentTaskService] Skipping task because previous run is still active', {
-					taskId,
-					agentId,
-				});
-				return;
-			}
-
-			renewInterval = this.startTaskRunLockRenewal(lock);
-			await this.runTask(agentId, taskId);
-		} catch (error) {
-			this.logger.error('[AgentTaskService] Scheduled task lock failed', {
+	/**
+	 * Checks that the task is still published and enabled, takes the run lock,
+	 * and starts the task run in the background. Resolves as soon as the lock
+	 * decision is made. Both schedulers enter here: the in-memory cron and the
+	 * durable handler. The handler must report its dispatch decision within the
+	 * lease of the occurrence, so it cannot wait for a run that takes minutes.
+	 *
+	 * A `stale` task drops its in-memory cron here; the durable handler removes
+	 * its own job. A lock-acquisition error propagates to the caller. The
+	 * background continuation logs run errors. It also renews the lock while the
+	 * run lasts, and releases the lock after the run.
+	 */
+	async startScheduledRun(
+		agentId: string,
+		taskId: string,
+	): Promise<'started' | 'skipped-active' | 'stale'> {
+		// Body comes from the PUBLISHED snapshot row, so name/objective/cron
+		// reflect publish time rather than live draft edits.
+		const agent = await this.agentRepository.findOne({ where: { id: agentId } });
+		const snapshot = agent?.activeVersionId
+			? await this.taskSnapshotRepository.findByVersionAndTaskId(agent.activeVersionId, taskId)
+			: null;
+		if (!agent?.activeVersionId || !snapshot?.enabled) {
+			this.logger.warn('[AgentTaskService] Task fired but is no longer published and enabled', {
 				taskId,
 				agentId,
-				error: error instanceof Error ? error.message : String(error),
 			});
-		} finally {
-			if (renewInterval) clearInterval(renewInterval);
-			if (lock) {
+			this.deregister(agentId, taskId);
+			return 'stale';
+		}
+
+		const holderId = randomUUID();
+		const lock = await this.taskRunLockRepository.acquire(agentId, taskId, {
+			holderId,
+			ttlMs: TASK_RUN_LOCK_TTL_MS,
+		});
+		if (!lock) {
+			this.logger.info('[AgentTaskService] Skipping task because previous run is still active', {
+				taskId,
+				agentId,
+			});
+			return 'skipped-active';
+		}
+
+		const renewInterval = this.startTaskRunLockRenewal(lock);
+		void (async () => {
+			try {
+				await this.runTask(agent, snapshot);
+			} catch (error) {
+				this.logger.error('[AgentTaskService] Scheduled task run failed', {
+					taskId,
+					agentId,
+					error: error instanceof Error ? error.message : String(error),
+				});
+			} finally {
+				clearInterval(renewInterval);
 				await this.taskRunLockRepository.release(lock).catch((error) => {
 					this.logger.warn('[AgentTaskService] Failed to release task run lock', {
 						taskId,
@@ -516,7 +570,8 @@ export class AgentTaskService {
 					});
 				});
 			}
-		}
+		})();
+		return 'started';
 	}
 
 	private startTaskRunLockRenewal(lock: AgentTaskRunLockHandle): ReturnType<typeof setInterval> {
@@ -541,77 +596,35 @@ export class AgentTaskService {
 		}, TASK_RUN_LOCK_RENEW_MS);
 	}
 
-	private async runTask(agentId: string, taskId: string): Promise<void> {
-		let projectId: string | undefined;
+	private async runTask(agent: Agent, snapshot: AgentTaskSnapshot): Promise<void> {
+		const { id: agentId, projectId } = agent;
+		const { taskId } = snapshot;
+		const { message, threadId } = this.buildTaskRunMessage(
+			taskId,
+			snapshot.objective,
+			snapshot.timezone,
+		);
 
-		try {
-			const agent = await this.agentRepository.findOne({
-				where: { id: agentId },
-				relations: { activeVersion: true },
-			});
-			if (!agent) {
-				this.deregister(agentId, taskId);
-				return;
-			}
-			projectId = agent.projectId;
+		this.logger.info('[AgentTaskService] Task fired', {
+			taskId,
+			agentId,
+			projectId,
+			cronExpression: snapshot.cronExpression,
+			timezone: snapshot.timezone,
+		});
 
-			if (!agent.activeVersionId) {
-				this.logger.warn('[AgentTaskService] Task fired for unpublished agent', {
-					taskId,
-					agentId,
-				});
-				this.deregister(agentId, taskId);
-				return;
-			}
-			// Body comes from the PUBLISHED snapshot row, so name/objective/cron
-			// reflect publish time rather than live draft edits.
-			const snapshot = await this.taskSnapshotRepository.findByVersionAndTaskId(
-				agent.activeVersionId,
-				taskId,
-			);
-			if (!snapshot?.enabled) {
-				this.logger.warn('[AgentTaskService] Task fired but has no enabled published snapshot', {
-					taskId,
-					agentId,
-				});
-				this.deregister(agentId, taskId);
-				return;
-			}
-
-			const { message, threadId } = this.buildTaskRunMessage(
-				taskId,
-				snapshot.objective,
-				snapshot.timezone,
-			);
-
-			this.logger.info('[AgentTaskService] Task fired', {
-				taskId,
+		await this.consumeTaskRun(
+			'Task run',
+			{ taskId, agentId, projectId },
+			this.agentExecutionOrchestratorService.executeForTaskPublished({
 				agentId,
 				projectId,
-				cronExpression: snapshot.cronExpression,
-				timezone: snapshot.timezone,
-			});
-
-			await this.consumeTaskRun(
-				'Task run',
-				{ taskId, agentId, projectId },
-				this.agentExecutionOrchestratorService.executeForTaskPublished({
-					agentId: agent.id,
-					projectId: agent.projectId,
-					message,
-					memory: { threadId, resourceId: taskRunMemoryResourceId(taskId) },
-					taskId,
-					taskVersionId: agent.activeVersionId,
-				}),
-			);
-		} catch (error) {
-			this.logger.error('[AgentTaskService] Task run failed', {
+				message,
+				memory: { threadId, resourceId: taskRunMemoryResourceId(taskId) },
 				taskId,
-				agentId,
-				projectId,
-				error: error instanceof Error ? error.message : String(error),
-			});
-		}
+				taskVersionId: snapshot.versionId,
+			}),
+		);
 	}
 
 	/**
@@ -728,22 +741,10 @@ export class AgentTaskService {
 		}
 	}
 
-	/**
-	 * Timezone a task's cron is evaluated in. Null means "instance timezone" —
-	 * the only option before tasks carried their own zone. An unresolvable zone
-	 * falls back to the instance timezone rather than dropping the task, since
-	 * `CronTime` would throw and take the agent's whole reconcile with it.
-	 */
 	private resolveTaskTimezone(taskTimezone: string | null | undefined, taskId: string): string {
-		if (!taskTimezone) return this.globalConfig.generic.timezone;
-		if (!isValidTimeZone(taskTimezone)) {
-			this.logger.warn('[AgentTaskService] Task has unknown timezone, using instance timezone', {
-				taskId,
-				timezone: taskTimezone,
-			});
-			return this.globalConfig.generic.timezone;
-		}
-		return taskTimezone;
+		return (
+			knownTaskTimezone(taskTimezone, taskId, this.logger) ?? this.globalConfig.generic.timezone
+		);
 	}
 
 	private toDto(task: AgentTask): AgentTaskDto {

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

@@ -147,8 +147,7 @@ export class AgentsModule implements ModuleInterface {
 			channelReconciler.init();
 		}
 
-		// Tasks are leader-only: only the leader should run the cron and reconnect tasks on startup.
-		// TODO: migrate to the durable scheduler
+		// In-memory task crons are leader-only. Only the leader reconnects them on startup.
 		if (instanceSettings.isLeader) {
 			void taskService.reconnectAll().catch((error) => {
 				logger.error('[Agents] Failed to reconnect tasks on startup', {
@@ -158,6 +157,37 @@ export class AgentsModule implements ModuleInterface {
 		} else {
 			logger.debug('[Agents] Skipping task reconnect on startup — not leader');
 		}
+
+		// Durable scheduling for agent tasks runs on mains only. The handler is
+		// registered before DurableScheduler starts, because module init precedes
+		// its start() in the start command. A main with the flag off registers no
+		// handler, so the executor never claims a leftover agent-task row. The
+		// reconcile runs with the flag in either state: it backfills jobs when the
+		// flag is on and removes them when it is off. It is awaited, so that the
+		// executor cannot claim an occurrence of a job it is about to redefine or
+		// remove.
+		if (instanceSettings.instanceType === 'main') {
+			const { AgentTaskJobRegistrar } = await import('./scheduling/agent-task-job-registrar.js');
+			const registrar = Container.get(AgentTaskJobRegistrar);
+
+			if (registrar.isEnabled()) {
+				const { AgentTaskTaskHandler } = await import('./scheduling/agent-task-task-handler.js');
+				const { DurableScheduler } = await import('@/scheduling/durable-scheduler.js');
+				const agentTaskHandler = Container.get(AgentTaskTaskHandler);
+				Container.get(DurableScheduler).registerTaskHandler(
+					agentTaskHandler.taskType,
+					agentTaskHandler,
+				);
+			}
+
+			try {
+				await registrar.reconcileAll();
+			} catch (error) {
+				logger.error('[Agents] Failed to reconcile durable agent-task jobs on startup', {
+					error: error instanceof Error ? error.message : String(error),
+				});
+			}
+		}
 	}
 
 	async systemTasks() {

packages/cli/src/modules/agents/integrations/__tests__/cron-validation.test.ts

@@ -0,0 +1,19 @@
+import { isValidCronExpression } from '../cron-validation';
+
+describe('isValidCronExpression', () => {
+	it.each(['0 9 * * *', '30 8 * * 1', '*/5 * * * *', '0 9 * * 1-5'])(
+		'accepts %s, which both schedulers can run',
+		(expression) => {
+			expect(isValidCronExpression(expression)).toBe(true);
+		},
+	);
+
+	it.each([
+		['a nickname the durable scheduler cannot plan', '@daily'],
+		['an impossible date', '0 9 30 2 *'],
+		['a malformed expression', 'not a cron'],
+		['an empty string', ''],
+	])('rejects %s', (_name, expression) => {
+		expect(isValidCronExpression(expression)).toBe(false);
+	});
+});

packages/cli/src/modules/agents/integrations/cron-validation.ts

@@ -1,11 +1,26 @@
+import { computeFirstRunAt } from '@n8n/scheduler';
 import { validateCronExpression } from 'cron';
+import type { CronExpression } from 'n8n-workflow';
 
 /**
- * Whether the given expression parses as a valid 5-field cron.
+ * Whether the given expression is a 5-field cron that both task schedulers can
+ * run. The in-memory scheduler parses with `cron`, the durable scheduler plans
+ * with `cron-parser`, and the two disagree on nicknames such as `@daily` and on
+ * impossible dates such as `0 9 30 2 *`. A task passes only when both accept it.
  *
  * Shared between the schedule REST endpoint and the JSON-config Zod schema so
  * both surfaces reject malformed crons with the same rule.
  */
-export function isValidCronExpression(expression: string): boolean {
-	return validateCronExpression(expression).valid;
+export function isValidCronExpression(expression: string): expression is CronExpression {
+	if (!validateCronExpression(expression).valid) return false;
+
+	try {
+		computeFirstRunAt(
+			{ kind: 'cron', cronExpression: expression as CronExpression, timezone: 'UTC' },
+			new Date(),
+		);
+		return true;
+	} catch {
+		return false;
+	}
 }

packages/cli/src/modules/agents/repositories/agent.repository.ts

@@ -305,6 +305,28 @@ export class AgentRepository extends Repository<Agent> {
 			.getMany();
 	}
 
+	/** The ids of all agents with a published version. Loads no version rows. */
+	async findPublishedAgentIds(): Promise<string[]> {
+		const rows = await this.find({
+			where: { activeVersionId: Not(IsNull()) },
+			select: ['id'],
+		});
+		return rows.map((row) => row.id);
+	}
+
+	/**
+	 * The published version id of an agent, or `null` when the agent is missing
+	 * or unpublished. Loads no version row, so callers that only need the id do
+	 * not pay for the version's JSON columns.
+	 */
+	async findActiveVersionId(agentId: string): Promise<string | null> {
+		const row = await this.findOne({
+			where: { id: agentId },
+			select: ['id', 'activeVersionId'],
+		});
+		return row?.activeVersionId ?? null;
+	}
+
 	/** The ids, from the given list, that belong to an agent with a published version. */
 	async findPublishedIds(agentIds: string[]): Promise<Set<string>> {
 		if (agentIds.length === 0) return new Set();

packages/cli/src/modules/agents/scheduling/__tests__/agent-task-job-registrar.test.ts

@@ -0,0 +1,286 @@
+import type { Logger } from '@n8n/backend-common';
+import type { GlobalConfig } from '@n8n/config';
+import { ScheduledJobMisfirePolicy } from '@n8n/constants';
+import type { ScheduledJobRepository } from '@n8n/db';
+import { mock } from 'vitest-mock-extended';
+
+import { AgentScheduledJobOwner } from '@/scheduling/agent-scheduled-job-owner';
+import type { DurableJobProvisioner } from '@/scheduling/durable-job-provisioner';
+
+import type { AgentTaskSnapshot } from '../../entities/agent-task-snapshot.entity';
+import type { AgentTaskSnapshotRepository } from '../../repositories/agent-task-snapshot.repository';
+import type { AgentRepository } from '../../repositories/agent.repository';
+import { AGENT_TASK_TASK_TYPE } from '../agent-task-job';
+import { AgentTaskJobRegistrar } from '../agent-task-job-registrar';
+
+const AGENT_ID = 'agent-1';
+const owner = (taskId: string) => ({
+	ownerType: 'agent',
+	ownerId: AGENT_ID,
+	ownerMemberId: taskId,
+});
+const OWNER_REF = { ownerType: 'agent', ownerId: AGENT_ID };
+
+describe('AgentTaskJobRegistrar', () => {
+	const logger = mock<Logger>();
+	logger.scoped.mockReturnValue(logger);
+	const agentRepository = mock<AgentRepository>();
+	const taskSnapshotRepository = mock<AgentTaskSnapshotRepository>();
+	const scheduledJobRepository = mock<ScheduledJobRepository>();
+	const provisioner = mock<DurableJobProvisioner>();
+	// The real owner, because the assertions below are about the owner shape that it produces.
+	const agentOwner = new AgentScheduledJobOwner(mock());
+
+	const makeRegistrar = (scheduler: Partial<GlobalConfig['scheduler']> = {}) =>
+		new AgentTaskJobRegistrar(
+			logger,
+			mock<GlobalConfig>({
+				scheduler: {
+					enabled: true,
+					enabledForAgentTasks: true,
+					ownerReconciliationBatchSize: 500,
+					...scheduler,
+				},
+				generic: { timezone: 'UTC' },
+			}),
+			agentRepository,
+			taskSnapshotRepository,
+			scheduledJobRepository,
+			provisioner,
+			agentOwner,
+		);
+
+	const snapshot = (overrides: Partial<AgentTaskSnapshot> = {}): AgentTaskSnapshot =>
+		mock<AgentTaskSnapshot>({
+			versionId: 'version-1',
+			taskId: 'task-1',
+			enabled: true,
+			cronExpression: '0 9 * * *',
+			timezone: null,
+			...overrides,
+		});
+
+	beforeEach(() => {
+		vi.clearAllMocks();
+		logger.scoped.mockReturnValue(logger);
+		agentRepository.findActiveVersionId.mockResolvedValue('version-1');
+		agentRepository.findPublishedAgentIds.mockResolvedValue([AGENT_ID]);
+		taskSnapshotRepository.findEnabledByVersionId.mockResolvedValue([snapshot()]);
+		scheduledJobRepository.findOwnerMemberIds.mockResolvedValue([]);
+		scheduledJobRepository.findOwnerIds.mockResolvedValue([]);
+		provisioner.provision.mockResolvedValue({
+			inserted: [],
+			redefined: [],
+			unchanged: [],
+			removed: [],
+		});
+		provisioner.deprovisionOwnerMember.mockResolvedValue({ removed: 0 });
+		provisioner.deprovisionOwnerTaskType.mockResolvedValue({ removed: 0 });
+	});
+
+	describe('reconcile', () => {
+		it('provisions each enabled snapshot as a skip-policy cron job owned by the agent through the task', async () => {
+			taskSnapshotRepository.findEnabledByVersionId.mockResolvedValue([
+				snapshot(),
+				snapshot({ taskId: 'task-2', cronExpression: '30 8 * * 1', timezone: 'Europe/Berlin' }),
+			]);
+
+			await makeRegistrar().reconcile(AGENT_ID);
+
+			expect(taskSnapshotRepository.findEnabledByVersionId).toHaveBeenCalledWith('version-1');
+			expect(provisioner.provision).toHaveBeenCalledTimes(2);
+			expect(provisioner.provision).toHaveBeenNthCalledWith(1, {
+				owner: owner('task-1'),
+				taskType: AGENT_TASK_TASK_TYPE,
+				payload: { agentId: AGENT_ID, taskId: 'task-1' },
+				misfirePolicy: ScheduledJobMisfirePolicy.Skip,
+				desired: [
+					{
+						name: `agent-task:${AGENT_ID}:task-1`,
+						schedule: { kind: 'cron', cronExpression: '0 9 * * *', timezone: null },
+						firstRunAt: expect.any(Date),
+					},
+				],
+			});
+			expect(provisioner.provision).toHaveBeenNthCalledWith(2, {
+				owner: owner('task-2'),
+				taskType: AGENT_TASK_TASK_TYPE,
+				payload: { agentId: AGENT_ID, taskId: 'task-2' },
+				misfirePolicy: ScheduledJobMisfirePolicy.Skip,
+				desired: [
+					{
+						name: `agent-task:${AGENT_ID}:task-2`,
+						schedule: { kind: 'cron', cronExpression: '30 8 * * 1', timezone: 'Europe/Berlin' },
+						firstRunAt: expect.any(Date),
+					},
+				],
+			});
+		});
+
+		it('removes the jobs of tasks that left the published config, keeping the ones still enabled', async () => {
+			scheduledJobRepository.findOwnerMemberIds.mockResolvedValue(['task-1', 'task-gone']);
+
+			await makeRegistrar().reconcile(AGENT_ID);
+
+			expect(scheduledJobRepository.findOwnerMemberIds).toHaveBeenCalledWith(
+				OWNER_REF,
+				AGENT_TASK_TASK_TYPE,
+			);
+			expect(provisioner.deprovisionOwnerMember).toHaveBeenCalledTimes(1);
+			expect(provisioner.deprovisionOwnerMember).toHaveBeenCalledWith(owner('task-gone'));
+		});
+
+		it.each([
+			[
+				'the published version has no enabled task',
+				() => taskSnapshotRepository.findEnabledByVersionId.mockResolvedValue([]),
+			],
+			[
+				'the agent is deleted or unpublished',
+				() => agentRepository.findActiveVersionId.mockResolvedValue(null),
+			],
+		])('removes every remaining agent-task job when %s', async (_, arrange) => {
+			arrange();
+			scheduledJobRepository.findOwnerMemberIds.mockResolvedValue(['task-1']);
+
+			await makeRegistrar().reconcile(AGENT_ID);
+
+			expect(provisioner.provision).not.toHaveBeenCalled();
+			expect(provisioner.deprovisionOwnerMember).toHaveBeenCalledWith(owner('task-1'));
+		});
+
+		it('provisions no job for a snapshot with an invalid cron, warning instead of failing the reconcile', async () => {
+			taskSnapshotRepository.findEnabledByVersionId.mockResolvedValue([
+				snapshot({ taskId: 'bad-task', cronExpression: 'not a cron' }),
+				snapshot(),
+			]);
+
+			await makeRegistrar().reconcile(AGENT_ID);
+
+			expect(provisioner.provision).toHaveBeenCalledTimes(2);
+			expect(provisioner.provision.mock.calls[0][0]).toMatchObject({
+				owner: owner('bad-task'),
+				desired: [],
+			});
+			expect(provisioner.provision.mock.calls[1][0].desired).toHaveLength(1);
+			expect(logger.warn).toHaveBeenCalledWith(
+				'Skipping task with invalid cron',
+				expect.objectContaining({ taskId: 'bad-task' }),
+			);
+		});
+
+		it('falls back to the instance timezone when a snapshot carries an unknown one', async () => {
+			taskSnapshotRepository.findEnabledByVersionId.mockResolvedValue([
+				snapshot({ timezone: 'Neverland/Nowhere' }),
+			]);
+
+			await makeRegistrar().reconcile(AGENT_ID);
+
+			expect(provisioner.provision.mock.calls[0][0].desired[0].schedule).toEqual({
+				kind: 'cron',
+				cronExpression: '0 9 * * *',
+				timezone: null,
+			});
+			expect(logger.warn).toHaveBeenCalledWith(
+				'Task has unknown timezone, using instance timezone',
+				expect.objectContaining({ timezone: 'Neverland/Nowhere' }),
+			);
+		});
+
+		it('applies the version published while it ran, so the newest config wins', async () => {
+			// version-1 has task-1, version-2 has task-2. The publish of version-2 lands
+			// between the first read and the re-read.
+			agentRepository.findActiveVersionId
+				.mockResolvedValueOnce('version-1')
+				.mockResolvedValueOnce('version-2')
+				.mockResolvedValue('version-2');
+			taskSnapshotRepository.findEnabledByVersionId.mockImplementation(async (versionId) =>
+				versionId === 'version-1'
+					? [snapshot()]
+					: [snapshot({ versionId: 'version-2', taskId: 'task-2' })],
+			);
+			scheduledJobRepository.findOwnerMemberIds.mockResolvedValue(['task-1', 'task-2']);
+
+			await makeRegistrar().reconcile(AGENT_ID);
+
+			expect(provisioner.provision.mock.calls.at(-1)?.[0]).toMatchObject({
+				owner: owner('task-2'),
+			});
+			expect(provisioner.deprovisionOwnerMember.mock.calls.at(-1)?.[0]).toEqual(owner('task-1'));
+		});
+
+		it('stops after a bounded number of passes when the active version never settles', async () => {
+			let publishes = 0;
+			agentRepository.findActiveVersionId.mockImplementation(async () => `version-${publishes++}`);
+
+			await makeRegistrar().reconcile(AGENT_ID);
+
+			expect(provisioner.provision).toHaveBeenCalledTimes(5);
+			expect(logger.warn).toHaveBeenCalledWith(
+				expect.stringContaining('Reconcile gave up'),
+				expect.objectContaining({ agentId: AGENT_ID }),
+			);
+		});
+	});
+
+	describe('reconcileAll', () => {
+		it('does nothing while the durable scheduler is off', async () => {
+			await makeRegistrar({ enabled: false }).reconcileAll();
+
+			expect(scheduledJobRepository.findOwnerIds).not.toHaveBeenCalled();
+			expect(provisioner.provision).not.toHaveBeenCalled();
+			expect(agentRepository.findPublishedAgentIds).not.toHaveBeenCalled();
+		});
+
+		it('deletes every agent’s jobs when the agent-tasks flag is off, so no occurrences pile up', async () => {
+			scheduledJobRepository.findOwnerIds
+				.mockResolvedValueOnce(['agent-a', 'agent-b'])
+				.mockResolvedValueOnce([]);
+			provisioner.deprovisionOwnerTaskType.mockResolvedValue({ removed: 2 });
+
+			await makeRegistrar({ enabledForAgentTasks: false }).reconcileAll();
+
+			expect(scheduledJobRepository.findOwnerIds).toHaveBeenNthCalledWith(
+				1,
+				'agent',
+				expect.any(Date),
+				500,
+				undefined,
+			);
+			expect(scheduledJobRepository.findOwnerIds).toHaveBeenNthCalledWith(
+				2,
+				'agent',
+				expect.any(Date),
+				500,
+				'agent-b',
+			);
+			expect(provisioner.deprovisionOwnerTaskType).toHaveBeenCalledWith(
+				{ ownerType: 'agent', ownerId: 'agent-a' },
+				AGENT_TASK_TASK_TYPE,
+			);
+			expect(provisioner.deprovisionOwnerTaskType).toHaveBeenCalledWith(
+				{ ownerType: 'agent', ownerId: 'agent-b' },
+				AGENT_TASK_TASK_TYPE,
+			);
+			expect(provisioner.provision).not.toHaveBeenCalled();
+		});
+
+		it('reconciles every published agent, continuing past one that fails', async () => {
+			agentRepository.findPublishedAgentIds.mockResolvedValue(['agent-broken', AGENT_ID]);
+			const registrar = makeRegistrar();
+			const reconcile = vi
+				.spyOn(registrar, 'reconcile')
+				.mockRejectedValueOnce(new Error('boom'))
+				.mockResolvedValueOnce(undefined);
+
+			await registrar.reconcileAll();
+
+			expect(reconcile).toHaveBeenCalledTimes(2);
+			expect(reconcile).toHaveBeenNthCalledWith(2, AGENT_ID);
+			expect(logger.error).toHaveBeenCalledWith(
+				'Failed to reconcile an agent’s durable jobs',
+				expect.objectContaining({ agentId: 'agent-broken', error: 'boom' }),
+			);
+		});
+	});
+});

packages/cli/src/modules/agents/scheduling/__tests__/agent-task-task-handler.test.ts

@@ -0,0 +1,109 @@
+import type { Logger } from '@n8n/backend-common';
+import { createDispatchReporter, type ClaimedTask } from '@n8n/scheduler';
+import { mock } from 'vitest-mock-extended';
+
+import type { AgentTaskService } from '../../agent-task.service';
+import { AGENT_TASK_TASK_TYPE } from '../agent-task-job';
+import type { AgentTaskJobRegistrar } from '../agent-task-job-registrar';
+import { AgentTaskTaskHandler } from '../agent-task-task-handler';
+
+const AGENT_ID = 'agent-1';
+const TASK_ID = 'task-1';
+
+describe('AgentTaskTaskHandler', () => {
+	const logger = mock<Logger>();
+	logger.scoped.mockReturnValue(logger);
+	const agentTaskService = mock<AgentTaskService>();
+	const registrar = mock<AgentTaskJobRegistrar>();
+
+	const handler = new AgentTaskTaskHandler(logger, agentTaskService, registrar);
+
+	// The dispatch-marker callback of the executor. vi.clearAllMocks() clears it in each test.
+	const onDispatch = vi.fn();
+	const report = createDispatchReporter(onDispatch);
+
+	const scheduledFor = new Date('2026-08-04T09:00:00.000Z');
+
+	const buildTask = (overrides: Partial<ClaimedTask> = {}): ClaimedTask => ({
+		id: 'occurrence-1',
+		jobId: 7,
+		taskType: AGENT_TASK_TASK_TYPE,
+		payload: { agentId: AGENT_ID, taskId: TASK_ID },
+		scheduledFor,
+		runAt: scheduledFor,
+		status: 'running',
+		attempts: 0,
+		maxAttempts: 1,
+		leaseEpoch: 1,
+		...overrides,
+	});
+
+	const flushAsyncWork = async () => await new Promise(setImmediate);
+
+	beforeEach(() => {
+		vi.clearAllMocks();
+		logger.scoped.mockReturnValue(logger);
+		agentTaskService.startScheduledRun.mockResolvedValue('started');
+		registrar.reconcile.mockResolvedValue(undefined);
+	});
+
+	it.each([
+		['a missing taskId', { agentId: AGENT_ID }],
+		['an empty agentId', { agentId: '', taskId: TASK_ID }],
+	])('throws on a payload with %s without reporting a dispatch', async (_, payload) => {
+		const task = buildTask({ payload });
+
+		await expect(handler.execute(task, report)).rejects.toThrow(
+			'Agent-task payload is missing agentId or taskId',
+		);
+		expect(onDispatch).not.toHaveBeenCalled();
+		expect(agentTaskService.startScheduledRun).not.toHaveBeenCalled();
+	});
+
+	describe('stale occurrences', () => {
+		beforeEach(() => {
+			agentTaskService.startScheduledRun.mockResolvedValue('stale');
+		});
+
+		it('completes without effect and self-heals', async () => {
+			const decision = await handler.execute(buildTask(), report);
+			await flushAsyncWork();
+
+			expect(decision).toBe(report.notDispatched());
+			expect(onDispatch).not.toHaveBeenCalled();
+			expect(registrar.reconcile).toHaveBeenCalledWith(AGENT_ID);
+		});
+
+		it('completes the occurrence and logs when the self-heal reconcile fails', async () => {
+			registrar.reconcile.mockRejectedValue(new Error('db down'));
+
+			const decision = await handler.execute(buildTask(), report);
+			await flushAsyncWork();
+
+			expect(decision).toBe(report.notDispatched());
+			expect(onDispatch).not.toHaveBeenCalled();
+			expect(logger.warn).toHaveBeenCalledWith(
+				'Failed to reconcile an agent after a stale occurrence',
+				expect.objectContaining({ agentId: AGENT_ID, error: 'db down' }),
+			);
+		});
+	});
+
+	it('skips the tick without dispatching when the previous run still holds the lock', async () => {
+		agentTaskService.startScheduledRun.mockResolvedValue('skipped-active');
+
+		const decision = await handler.execute(buildTask(), report);
+
+		expect(decision).toBe(report.notDispatched());
+		expect(onDispatch).not.toHaveBeenCalled();
+		expect(registrar.reconcile).not.toHaveBeenCalled();
+	});
+
+	it('reports dispatched after handing the run off', async () => {
+		const decision = await handler.execute(buildTask(), report);
+
+		expect(agentTaskService.startScheduledRun).toHaveBeenCalledWith(AGENT_ID, TASK_ID);
+		expect(onDispatch).toHaveBeenCalledTimes(1);
+		expect(decision).toBe(report.dispatched());
+	});
+});

packages/cli/src/modules/agents/scheduling/agent-task-job-registrar.ts

@@ -0,0 +1,208 @@
+import { Logger } from '@n8n/backend-common';
+import { GlobalConfig } from '@n8n/config';
+import { ScheduledJobMisfirePolicy } from '@n8n/constants';
+import { ScheduledJobRepository } from '@n8n/db';
+import { Service } from '@n8n/di';
+import type { DesiredJob } from '@n8n/scheduler';
+import { computeFirstRunAt } from '@n8n/scheduler';
+
+import { AgentScheduledJobOwner } from '@/scheduling/agent-scheduled-job-owner';
+import { DurableJobProvisioner } from '@/scheduling/durable-job-provisioner';
+
+import { AGENT_TASK_TASK_TYPE, agentTaskJobName, type AgentTaskJobPayload } from './agent-task-job';
+import { knownTaskTimezone } from './task-timezone';
+import type { AgentTaskSnapshot } from '../entities/agent-task-snapshot.entity';
+import { isValidCronExpression } from '../integrations/cron-validation';
+import { AgentTaskSnapshotRepository } from '../repositories/agent-task-snapshot.repository';
+import { AgentRepository } from '../repositories/agent.repository';
+
+/** Passes a reconcile makes before it stops chasing an active version that keeps changing. */
+const MAX_RECONCILE_PASSES = 5;
+
+/**
+ * The write side of durable agent-task scheduling. It keeps the
+ * `scheduled_job` rows of an agent in step with its published, enabled task
+ * snapshots. Each task is one owner member, so the job of a task carries its
+ * own payload. A task can be added or removed without a change to its siblings.
+ *
+ * Reconciles are plain database writes, so any main can apply one directly,
+ * with no leader and no pubsub. The scheduler drops missed occurrences
+ * (`skip`), the same as the in-memory scheduler that this replaces.
+ */
+@Service()
+export class AgentTaskJobRegistrar {
+	constructor(
+		private logger: Logger,
+		private readonly globalConfig: GlobalConfig,
+		private readonly agentRepository: AgentRepository,
+		private readonly taskSnapshotRepository: AgentTaskSnapshotRepository,
+		private readonly scheduledJobRepository: ScheduledJobRepository,
+		private readonly provisioner: DurableJobProvisioner,
+		private readonly owner: AgentScheduledJobOwner,
+	) {
+		this.logger = this.logger.scoped('scheduler');
+	}
+
+	/** Whether the durable scheduler runs agent tasks on this instance. */
+	isEnabled(): boolean {
+		const { enabled, enabledForAgentTasks } = this.globalConfig.scheduler;
+		return enabled && enabledForAgentTasks;
+	}
+
+	/**
+	 * Makes the durable jobs of an agent match its published config. Enabled
+	 * snapshots of the active version become jobs. All other jobs are removed.
+	 * An unpublished or deleted agent keeps no jobs. Callers gate on
+	 * `isEnabled()`.
+	 */
+	async reconcile(agentId: string): Promise<void> {
+		// Reconciles take no lock, so two can interleave: two publishes, or a
+		// publish and a delete. The one that acted on a stale read can undo the
+		// newer one. Each repeats until the version it applied is still the active
+		// one, so the last writer always applies the current version. The bound
+		// keeps a version that never settles from pinning startup, which awaits
+		// this; the publish that moved it reconciles again on its own.
+		let versionId = await this.agentRepository.findActiveVersionId(agentId);
+		for (let pass = 1; pass <= MAX_RECONCILE_PASSES; pass++) {
+			await this.applyVersion(agentId, versionId);
+			const current = await this.agentRepository.findActiveVersionId(agentId);
+			if (current === versionId) return;
+			versionId = current;
+		}
+		this.logger.warn('Reconcile gave up, the active version changed on every pass', {
+			agentId,
+			passes: MAX_RECONCILE_PASSES,
+		});
+	}
+
+	private async applyVersion(agentId: string, versionId: string | null): Promise<void> {
+		// No active version (unpublished, deleted) means no desired jobs.
+		const snapshots = versionId
+			? await this.taskSnapshotRepository.findEnabledByVersionId(versionId)
+			: [];
+		const keptTaskIds = new Set<string>();
+		for (const snapshot of snapshots) {
+			const desired = this.desiredJobFor(agentId, snapshot);
+			const payload: AgentTaskJobPayload = { agentId, taskId: snapshot.taskId };
+			await this.provisioner.provision({
+				owner: this.owner.member(agentId, snapshot.taskId),
+				taskType: AGENT_TASK_TASK_TYPE,
+				payload: { ...payload },
+				desired: desired ? [desired] : [],
+				misfirePolicy: ScheduledJobMisfirePolicy.Skip,
+			});
+			keptTaskIds.add(snapshot.taskId);
+		}
+
+		// Tasks that left the published config (removed, disabled, or the whole
+		// version gone) can still hold jobs from an earlier reconcile. The loop
+		// above did not touch their members.
+		const memberIds = await this.scheduledJobRepository.findOwnerMemberIds(
+			this.owner.ref(agentId),
+			AGENT_TASK_TASK_TYPE,
+		);
+		for (const taskId of memberIds) {
+			if (keptTaskIds.has(taskId)) continue;
+			await this.provisioner.deprovisionOwnerMember(this.owner.member(agentId, taskId));
+		}
+	}
+
+	/**
+	 * Brings stored state in line with the flag at startup. Flag on: backfills
+	 * jobs for every published agent, which covers the first flip and rows
+	 * missed while this instance was down. Flag off while the scheduler runs:
+	 * removes all agent jobs. The executor never claims an unregistered task
+	 * type, so leftover rows collect pending occurrences that fire as a storm on
+	 * a later flip. The removal is also the rollback path.
+	 *
+	 * Runs on every main. Provisioning reconciles in place, so concurrent
+	 * startups converge.
+	 */
+	async reconcileAll(): Promise<void> {
+		if (!this.globalConfig.scheduler.enabled) return;
+
+		if (!this.isEnabled()) {
+			await this.deprovisionAllAgents();
+			return;
+		}
+
+		const agentIds = await this.agentRepository.findPublishedAgentIds();
+		this.logger.debug('Reconciling durable jobs of published agents', { count: agentIds.length });
+		for (const agentId of agentIds) {
+			try {
+				await this.reconcile(agentId);
+			} catch (error) {
+				this.logger.error('Failed to reconcile an agent’s durable jobs', {
+					agentId,
+					error: error instanceof Error ? error.message : String(error),
+				});
+			}
+		}
+	}
+
+	/**
+	 * Walks every agent that owns jobs and removes its agent-task jobs, one page
+	 * of owner ids at a time. Jobs of other task types that an agent may own are
+	 * outside this flag and stay.
+	 */
+	private async deprovisionAllAgents(): Promise<void> {
+		const pageSize = this.globalConfig.scheduler.ownerReconciliationBatchSize;
+		let after: string | undefined;
+		let removed = 0;
+		for (;;) {
+			const agentIds = await this.scheduledJobRepository.findOwnerIds(
+				this.owner.ownerType,
+				new Date(),
+				pageSize,
+				after,
+			);
+			if (agentIds.length === 0) break;
+
+			for (const agentId of agentIds) {
+				const result = await this.provisioner.deprovisionOwnerTaskType(
+					this.owner.ref(agentId),
+					AGENT_TASK_TASK_TYPE,
+				);
+				removed += result.removed;
+			}
+			after = agentIds[agentIds.length - 1];
+		}
+
+		if (removed > 0) {
+			this.logger.info('Removed durable agent-task jobs because the feature is off', { removed });
+		}
+	}
+
+	/**
+	 * Maps one enabled snapshot to a desired job. A snapshot whose cron or
+	 * timezone cannot be planned gives no job and a warning. The reconcile of
+	 * the agent does not fail, the same as in the in-memory registration path.
+	 */
+	private desiredJobFor(agentId: string, snapshot: AgentTaskSnapshot): DesiredJob | null {
+		const { taskId, cronExpression } = snapshot;
+
+		// Also the type guard that narrows the stored string to a `CronExpression`.
+		if (!isValidCronExpression(cronExpression)) {
+			this.logger.warn('Skipping task with invalid cron', { taskId, agentId });
+			return null;
+		}
+
+		const timezone = knownTaskTimezone(snapshot.timezone, taskId, this.logger);
+		const schedule = { kind: 'cron' as const, cronExpression, timezone };
+
+		try {
+			const firstRunAt = computeFirstRunAt(
+				{ ...schedule, timezone: timezone ?? this.globalConfig.generic.timezone },
+				new Date(),
+			);
+			return { name: agentTaskJobName(agentId, taskId), schedule, firstRunAt };
+		} catch (error) {
+			this.logger.warn('Skipping task whose schedule cannot be planned', {
+				taskId,
+				agentId,
+				error: error instanceof Error ? error.message : String(error),
+			});
+			return null;
+		}
+	}
+}

packages/cli/src/modules/agents/scheduling/agent-task-job.ts

@@ -0,0 +1,30 @@
+/**
+ * Task type that routes durable-scheduler occurrences to the agent-task handler.
+ * The agent (`ownerId`) owns a job of this type through one of its scheduled
+ * tasks (`ownerMemberId`).
+ */
+export const AGENT_TASK_TASK_TYPE = 'agent:scheduled-task';
+
+/**
+ * Stamped on every agent-task job and copied onto each occurrence. It carries
+ * no version id. The handler reads the `activeVersionId` of the agent again at
+ * fire time, so an occurrence materialized before a republish runs the newest
+ * published snapshot.
+ */
+export interface AgentTaskJobPayload {
+	agentId: string;
+	taskId: string;
+}
+
+const isNonEmptyString = (value: unknown): value is string =>
+	typeof value === 'string' && value !== '';
+
+export const isAgentTaskJobPayload = (
+	payload: Record<string, unknown>,
+): payload is Record<string, unknown> & AgentTaskJobPayload =>
+	isNonEmptyString(payload.agentId) && isNonEmptyString(payload.taskId);
+
+/** Unique job name for one agent task. Task ids are unique within an agent. */
+export function agentTaskJobName(agentId: string, taskId: string): string {
+	return `agent-task:${agentId}:${taskId}`;
+}

packages/cli/src/modules/agents/scheduling/agent-task-task-handler.ts

@@ -0,0 +1,83 @@
+import { Logger } from '@n8n/backend-common';
+import { Service } from '@n8n/di';
+import type { ClaimedTask, DispatchDecision, DispatchReporter, TaskHandler } from '@n8n/scheduler';
+import { UnexpectedError } from 'n8n-workflow';
+
+import { AgentTaskService } from '../agent-task.service';
+import { AGENT_TASK_TASK_TYPE, isAgentTaskJobPayload } from './agent-task-job';
+import { AgentTaskJobRegistrar } from './agent-task-job-registrar';
+
+/**
+ * Runs a due agent-task occurrence.
+ *
+ * The handoff is the start of the run. The handler takes the run lock, starts
+ * the agent run in the background, and reports dispatched. The run lasts
+ * minutes, far more than the lease of the occurrence, and nothing renews the
+ * lease during the run. So the handler must not wait for the run. The
+ * `agent_task_run_lock` that the run holds bounds overlap and redelivery.
+ *
+ * The handoff also ends the crash protection of the scheduler. A main that
+ * dies after the lock and before the run records its `agent_execution` row
+ * leaves no trace, and the lock blocks the next ticks until its TTL ends. The
+ * in-memory scheduler has the same gap. Once `@n8n/scheduler` can renew the
+ * lease during a handler, the handler can await the run instead, and the
+ * reaper then covers a crash at any point of the run.
+ */
+@Service()
+export class AgentTaskTaskHandler implements TaskHandler {
+	readonly taskType = AGENT_TASK_TASK_TYPE;
+
+	constructor(
+		private logger: Logger,
+		private readonly agentTaskService: AgentTaskService,
+		private readonly registrar: AgentTaskJobRegistrar,
+	) {
+		this.logger = this.logger.scoped('scheduler');
+	}
+
+	async execute(task: ClaimedTask, report: DispatchReporter): Promise<DispatchDecision> {
+		if (!isAgentTaskJobPayload(task.payload)) {
+			throw new UnexpectedError('Agent-task payload is missing agentId or taskId', {
+				extra: { taskId: task.id, jobId: task.jobId },
+			});
+		}
+		const { agentId, taskId } = task.payload;
+
+		const outcome = await this.agentTaskService.startScheduledRun(agentId, taskId);
+		if (outcome === 'stale') {
+			// The task no longer qualifies under the published config. The handler
+			// completes the occurrence without effect and reconciles again, so
+			// that the job that produced it goes away.
+			this.logger.warn('Dropping occurrence of an agent task that is no longer scheduled', {
+				agentId,
+				taskId,
+				jobId: task.jobId,
+			});
+			this.selfHeal(agentId);
+			return report.notDispatched();
+		}
+		if (outcome === 'skipped-active') {
+			// The previous run still holds the lock. The handler skips this tick,
+			// the same as the in-memory scheduler. The occurrence completes, and
+			// the next tick is a fresh one.
+			return report.notDispatched();
+		}
+
+		this.logger.debug('Handed off agent-task occurrence to a run', {
+			agentId,
+			taskId,
+			jobId: task.jobId,
+		});
+		return report.dispatched();
+	}
+
+	/** Removes the stale job behind this occurrence. Best-effort, off the hot path. */
+	private selfHeal(agentId: string): void {
+		void this.registrar.reconcile(agentId).catch((error) => {
+			this.logger.warn('Failed to reconcile an agent after a stale occurrence', {
+				agentId,
+				error: error instanceof Error ? error.message : String(error),
+			});
+		});
+	}
+}

packages/cli/src/modules/agents/scheduling/task-timezone.ts

@@ -0,0 +1,19 @@
+import { isValidTimeZone } from '@n8n/api-types';
+import type { Logger } from '@n8n/backend-common';
+
+/**
+ * The IANA zone a task's cron is evaluated in, or `null` for the instance
+ * timezone. Null is the shape tasks had before they carried a zone. An unknown
+ * zone also resolves to null, with a warning, rather than failing the task:
+ * the cron planner would throw and take the agent's whole reconcile with it.
+ */
+export function knownTaskTimezone(
+	timezone: string | null | undefined,
+	taskId: string,
+	logger: Logger,
+): string | null {
+	if (!timezone) return null;
+	if (isValidTimeZone(timezone)) return timezone;
+	logger.warn('Task has unknown timezone, using instance timezone', { taskId, timezone });
+	return null;
+}

packages/cli/test/integration/scheduling/agent-task-multi-main.test.ts

@@ -0,0 +1,199 @@
+import { createTeamProject, testDb, testModules } from '@n8n/backend-test-utils';
+import { GlobalConfig } from '@n8n/config';
+import { ScheduledJobOwnerType } from '@n8n/constants';
+import { DataSource, ScheduledJobRepository, ScheduledTaskRepository } from '@n8n/db';
+import { Container } from '@n8n/di';
+import { createScheduler } from '@n8n/scheduler';
+import type { Scheduler, SchedulerPasses } from '@n8n/scheduler';
+import { v4 as uuid } from 'uuid';
+import { mock } from 'vitest-mock-extended';
+
+import { AgentTaskService } from '@/modules/agents/agent-task.service';
+import type { Agent } from '@/modules/agents/entities/agent.entity';
+import { AgentHistoryRepository } from '@/modules/agents/repositories/agent-history.repository';
+import { AgentTaskSnapshotRepository } from '@/modules/agents/repositories/agent-task-snapshot.repository';
+import { AgentRepository } from '@/modules/agents/repositories/agent.repository';
+import { AGENT_TASK_TASK_TYPE } from '@/modules/agents/scheduling/agent-task-job';
+import { AgentTaskJobRegistrar } from '@/modules/agents/scheduling/agent-task-job-registrar';
+import { AgentTaskTaskHandler } from '@/modules/agents/scheduling/agent-task-task-handler';
+import { buildMaterializerTransaction } from '@/scheduling/durable-scheduler';
+
+import { retryUntil } from '../shared/retry-until';
+import { createDueJobFactory } from './shared/job-factory';
+
+/**
+ * Agent tasks on the durable scheduler, over one database and two scheduler
+ * instances, the way two mains run in production. The registrar, provisioner
+ * and handler are the production classes; only the run start is stubbed. The
+ * generic multi-main suite proves the claim guarantees for any task type. This
+ * proves the agent wiring on top of them: a publish on any main is a database
+ * write that every main fires from, a due occurrence starts exactly one run,
+ * a task unpublished after its occurrence was recorded does not run, and a
+ * main with the flag off removes the agent-task jobs and nothing else.
+ */
+describe('agent tasks across two mains over one database', () => {
+	let jobRepo: ScheduledJobRepository;
+	let taskRepo: ScheduledTaskRepository;
+	let agentRepo: AgentRepository;
+	let historyRepo: AgentHistoryRepository;
+	let snapshotRepo: AgentTaskSnapshotRepository;
+	let registrar: AgentTaskJobRegistrar;
+	let schedulerConfig: GlobalConfig['scheduler'];
+	let projectId: string;
+
+	const startScheduledRun = vi.fn<AgentTaskService['startScheduledRun']>();
+
+	let mainA: Scheduler & SchedulerPasses;
+	let mainB: Scheduler & SchedulerPasses;
+
+	const buildMain = (hostId: string) => {
+		const scheduler = createScheduler({
+			hostId,
+			materializerTransaction: buildMaterializerTransaction(
+				Container.get(DataSource),
+				jobRepo,
+				taskRepo,
+			),
+			taskStore: taskRepo,
+			executor: { leaseSeconds: 30, lookaheadSeconds: 5, batchSize: 5 },
+		});
+		const handler = Container.get(AgentTaskTaskHandler);
+		scheduler.registerTaskHandler(handler.taskType, handler);
+		return scheduler;
+	};
+
+	/** A published agent with one task on the given cron. */
+	const publishAgentWithTask = async (cronExpression = '* * * * *') => {
+		const versionId = uuid();
+		// Task ids are capped at 32 characters.
+		const taskId = uuid().replaceAll('-', '');
+		const agent = await agentRepo.save(
+			agentRepo.create({
+				id: uuid(),
+				name: 'Scheduled agent',
+				projectId,
+				schema: { name: 'Scheduled agent', model: 'm', instructions: 'i' },
+				integrations: [],
+				tools: {},
+				skills: {},
+				versionId,
+				activeVersionId: null,
+			} as Partial<Agent>),
+		);
+		await historyRepo.insert({
+			versionId,
+			agentId: agent.id,
+			author: 'test',
+			schema: null,
+			tools: null,
+			skills: null,
+		});
+		await snapshotRepo.saveForVersion([
+			{
+				versionId,
+				taskId,
+				enabled: true,
+				name: 'Every minute',
+				objective: 'Report',
+				cronExpression,
+				timezone: null,
+			},
+		]);
+		await agentRepo.update({ id: agent.id }, { activeVersionId: versionId });
+		return { agentId: agent.id, taskId };
+	};
+
+	/** Publishes the task's job and pulls its first occurrence into the past. */
+	const provisionDueOccurrence = async (agentId: string) => {
+		await registrar.reconcile(agentId);
+		const job = await jobRepo.findOneByOrFail({ ownerId: agentId, taskType: AGENT_TASK_TASK_TYPE });
+		await taskRepo.update({ jobId: job.id }, { runAt: new Date(Date.now() - 1000) });
+		return job;
+	};
+
+	beforeAll(async () => {
+		await testModules.loadModules(['agents']);
+		await testDb.init();
+		Container.set(AgentTaskService, mock<AgentTaskService>({ startScheduledRun }));
+		schedulerConfig = Container.get(GlobalConfig).scheduler;
+		schedulerConfig.enabled = true;
+		schedulerConfig.enabledForAgentTasks = true;
+
+		jobRepo = Container.get(ScheduledJobRepository);
+		taskRepo = Container.get(ScheduledTaskRepository);
+		agentRepo = Container.get(AgentRepository);
+		historyRepo = Container.get(AgentHistoryRepository);
+		snapshotRepo = Container.get(AgentTaskSnapshotRepository);
+		registrar = Container.get(AgentTaskJobRegistrar);
+		mainA = buildMain('main-a');
+		mainB = buildMain('main-b');
+	});
+
+	beforeEach(async () => {
+		startScheduledRun.mockReset();
+		startScheduledRun.mockResolvedValue('started');
+		schedulerConfig.enabledForAgentTasks = true;
+		await testDb.truncate(['ScheduledTask', 'ScheduledJob']);
+		await snapshotRepo.delete({});
+		await historyRepo.delete({});
+		await agentRepo.delete({});
+		projectId = (await createTeamProject()).id;
+	});
+
+	afterAll(async () => {
+		await mainA.stop();
+		await mainB.stop();
+		await testDb.terminate();
+	});
+
+	it('starts exactly one run for a due occurrence when both mains claim at once', async () => {
+		const { agentId, taskId } = await publishAgentWithTask();
+		const job = await provisionDueOccurrence(agentId);
+
+		const [claimedA, claimedB] = await Promise.all([mainA.execute(), mainB.execute()]);
+
+		expect(claimedA.length + claimedB.length).toBe(1);
+		await retryUntil(async () => {
+			const [occurrence] = await taskRepo.findBy({ jobId: job.id });
+			expect(occurrence.status).toBe('succeeded');
+		});
+		expect(startScheduledRun).toHaveBeenCalledTimes(1);
+		expect(startScheduledRun).toHaveBeenCalledWith(agentId, taskId);
+	}, 15_000);
+
+	it('removes the job of a task unpublished on another main once its occurrence fires stale', async () => {
+		const { agentId } = await publishAgentWithTask();
+		await provisionDueOccurrence(agentId);
+		// The unpublish is a plain database write on some main; no pubsub reaches the
+		// main that fires. The run path reports the task stale at fire time.
+		await agentRepo.update({ id: agentId }, { activeVersionId: null });
+		startScheduledRun.mockResolvedValue('stale');
+
+		await Promise.all([mainA.execute(), mainB.execute()]);
+
+		await retryUntil(async () =>
+			expect(await jobRepo.countBy({ ownerId: agentId, taskType: AGENT_TASK_TASK_TYPE })).toBe(0),
+		);
+		expect(await taskRepo.countBy({ status: 'pending' })).toBe(0);
+	}, 15_000);
+
+	it('removes the agent-task jobs of every agent when the flag is off, and nothing else', async () => {
+		const { agentId } = await publishAgentWithTask();
+		const other = await publishAgentWithTask();
+		const agentTaskJob = await provisionDueOccurrence(agentId);
+		await provisionDueOccurrence(other.agentId);
+		// A job the agent owns through another feature, outside this flag.
+		const timerJob = await createDueJobFactory(
+			jobRepo,
+			'agent:test-timer',
+			'timer',
+		)({ ownerType: ScheduledJobOwnerType.Agent, ownerId: agentId, ownerMemberId: null });
+
+		schedulerConfig.enabledForAgentTasks = false;
+		await registrar.reconcileAll();
+
+		expect(await jobRepo.countBy({ taskType: AGENT_TASK_TASK_TYPE })).toBe(0);
+		expect(await taskRepo.countBy({ jobId: agentTaskJob.id })).toBe(0);
+		expect(await jobRepo.findOneBy({ id: timerJob.id })).not.toBeNull();
+	});
+});

packages/cli/test/integration/scheduling/scheduled-repositories.test.ts

@@ -616,6 +616,25 @@ describe('scheduled repositories', () => {
 		});
 	});
 
+	describe('ScheduledJobRepository.findOwnerMemberIds', () => {
+		it('lists each member once, for this owner and task type only, skipping member-less jobs', async () => {
+			const agent = { ownerType: 'agent', ownerId: 'agent-1' };
+			const taskType = 'agent:scheduled-task';
+			// Two jobs under one member must yield that member once.
+			await createJob({ ...agent, ownerMemberId: 'task-1', taskType });
+			await createJob({ ...agent, ownerMemberId: 'task-1', taskType });
+			await createJob({ ...agent, ownerMemberId: 'task-2', taskType });
+			// Not this task type, not a member, not this owner.
+			await createJob({ ...agent, ownerMemberId: 'timer-1', taskType: 'agent:wakeup' });
+			await createJob({ ...agent, ownerMemberId: null, taskType });
+			await createJob({ ...agent, ownerId: 'agent-2', ownerMemberId: 'task-9', taskType });
+
+			const memberIds = await jobRepository.findOwnerMemberIds(agent, taskType);
+
+			expect(memberIds.sort()).toEqual(['task-1', 'task-2']);
+		});
+	});
+
 	describe('ScheduledJobRepository.findQuarantinedByOwnerIds', () => {
 		it('returns at most `limit` jobs', async () => {
 			const owner = workflowOwned('wf-quarantined', 'node');