diff --git a/.gitignore b/.gitignore index 9ad13b95f43..edcd2941646 100644 --- a/.gitignore +++ b/.gitignore @@ -83,7 +83,7 @@ packages/cli/src/commands/export/outputs .claude/plans/ .claude/worktrees/ .claude/specs/ -.cursor/plans/ +.cursor/ .superset .conductor .n8n diff --git a/packages/@n8n/config/src/configs/scheduler.config.ts b/packages/@n8n/config/src/configs/scheduler.config.ts index e60c3092008..9128908f99b 100644 --- a/packages/@n8n/config/src/configs/scheduler.config.ts +++ b/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. diff --git a/packages/@n8n/config/test/config.test.ts b/packages/@n8n/config/test/config.test.ts index 2899a22ffb2..23669e8f484 100644 --- a/packages/@n8n/config/test/config.test.ts +++ b/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, diff --git a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts index fbfd2418381..ab0fbc1dab8 100644 --- a/packages/@n8n/db/src/repositories/scheduled-job.repository.ts +++ b/packages/@n8n/db/src/repositories/scheduled-job.repository.ts @@ -134,6 +134,21 @@ export class ScheduledJobRepository extends Repository { 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 { + 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 { return await this.count({ where: ownerCriteria(owner) }); } diff --git a/packages/cli/src/modules/agents/__tests__/agent-task.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-task.service.test.ts index dbeb9b72b9e..dd4c872b389 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-task.service.test.ts +++ b/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); } +/** 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 { return { versionId: 'ver-1', @@ -106,27 +112,6 @@ async function* throwingStream(): AsyncGenerator { throw new Error('execution failed'); } -async function runTaskOf( - service: AgentTaskService, - agentId: string, - taskId: string, -): Promise { - await (service as unknown as { runTask(agentId: string, taskId: string): Promise }).runTask( - agentId, - taskId, - ); -} - -async function runScheduledTaskOf( - service: AgentTaskService, - agentId: string, - taskId: string, -): Promise { - await ( - service as unknown as { runScheduledTask(agentId: string, taskId: string): Promise } - ).runScheduledTask(agentId, taskId); -} - async function flushAsyncWork(): Promise { await new Promise((resolve) => setImmediate(resolve)); } @@ -145,6 +130,7 @@ describe('AgentTaskService', () => { let agentTaskScheduler: ReturnType>; let publisher: ReturnType>; let modificationTelemetry: ReturnType>; + let durableJobRegistrar: ReturnType>; 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.publishCommand.mockResolvedValue(undefined); modificationTelemetry = mock(); + durableJobRegistrar = mock(); + // 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 { + await new Promise((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', () => { diff --git a/packages/cli/src/modules/agents/agent-task.service.ts b/packages/cli/src/modules/agents/agent-task.service.ts index 7c1e817ef8e..608ccf1cde6 100644 --- a/packages/cli/src/modules/agents/agent-task.service.ts +++ b/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 { + // 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 { + // 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 { + // 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 { - const holderId = randomUUID(); - let lock: AgentTaskRunLockHandle | null = null; - let renewInterval: ReturnType | 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 { @@ -541,77 +596,35 @@ export class AgentTaskService { }, TASK_RUN_LOCK_RENEW_MS); } - private async runTask(agentId: string, taskId: string): Promise { - let projectId: string | undefined; + private async runTask(agent: Agent, snapshot: AgentTaskSnapshot): Promise { + 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 { diff --git a/packages/cli/src/modules/agents/agents.module.ts b/packages/cli/src/modules/agents/agents.module.ts index 48823a726a6..4f4a7417898 100644 --- a/packages/cli/src/modules/agents/agents.module.ts +++ b/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() { diff --git a/packages/cli/src/modules/agents/integrations/__tests__/cron-validation.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/cron-validation.test.ts new file mode 100644 index 00000000000..a9208f0e41a --- /dev/null +++ b/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); + }); +}); diff --git a/packages/cli/src/modules/agents/integrations/cron-validation.ts b/packages/cli/src/modules/agents/integrations/cron-validation.ts index 4bc6e82c7a9..8e30def3f84 100644 --- a/packages/cli/src/modules/agents/integrations/cron-validation.ts +++ b/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; + } } diff --git a/packages/cli/src/modules/agents/repositories/agent.repository.ts b/packages/cli/src/modules/agents/repositories/agent.repository.ts index 0ddcaa65484..047ec8cba81 100644 --- a/packages/cli/src/modules/agents/repositories/agent.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent.repository.ts @@ -305,6 +305,28 @@ export class AgentRepository extends Repository { .getMany(); } + /** The ids of all agents with a published version. Loads no version rows. */ + async findPublishedAgentIds(): Promise { + 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 { + 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> { if (agentIds.length === 0) return new Set(); diff --git a/packages/cli/src/modules/agents/scheduling/__tests__/agent-task-job-registrar.test.ts b/packages/cli/src/modules/agents/scheduling/__tests__/agent-task-job-registrar.test.ts new file mode 100644 index 00000000000..2de95083e0b --- /dev/null +++ b/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.scoped.mockReturnValue(logger); + const agentRepository = mock(); + const taskSnapshotRepository = mock(); + const scheduledJobRepository = mock(); + const provisioner = mock(); + // The real owner, because the assertions below are about the owner shape that it produces. + const agentOwner = new AgentScheduledJobOwner(mock()); + + const makeRegistrar = (scheduler: Partial = {}) => + new AgentTaskJobRegistrar( + logger, + mock({ + scheduler: { + enabled: true, + enabledForAgentTasks: true, + ownerReconciliationBatchSize: 500, + ...scheduler, + }, + generic: { timezone: 'UTC' }, + }), + agentRepository, + taskSnapshotRepository, + scheduledJobRepository, + provisioner, + agentOwner, + ); + + const snapshot = (overrides: Partial = {}): AgentTaskSnapshot => + mock({ + 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' }), + ); + }); + }); +}); diff --git a/packages/cli/src/modules/agents/scheduling/__tests__/agent-task-task-handler.test.ts b/packages/cli/src/modules/agents/scheduling/__tests__/agent-task-task-handler.test.ts new file mode 100644 index 00000000000..7315a9c7971 --- /dev/null +++ b/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.scoped.mockReturnValue(logger); + const agentTaskService = mock(); + const registrar = mock(); + + 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 => ({ + 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()); + }); +}); diff --git a/packages/cli/src/modules/agents/scheduling/agent-task-job-registrar.ts b/packages/cli/src/modules/agents/scheduling/agent-task-job-registrar.ts new file mode 100644 index 00000000000..ab1e728c747 --- /dev/null +++ b/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 { + // 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 { + // No active version (unpublished, deleted) means no desired jobs. + const snapshots = versionId + ? await this.taskSnapshotRepository.findEnabledByVersionId(versionId) + : []; + const keptTaskIds = new Set(); + 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 { + 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 { + 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; + } + } +} diff --git a/packages/cli/src/modules/agents/scheduling/agent-task-job.ts b/packages/cli/src/modules/agents/scheduling/agent-task-job.ts new file mode 100644 index 00000000000..5d24dafd4c0 --- /dev/null +++ b/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, +): payload is Record & 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}`; +} diff --git a/packages/cli/src/modules/agents/scheduling/agent-task-task-handler.ts b/packages/cli/src/modules/agents/scheduling/agent-task-task-handler.ts new file mode 100644 index 00000000000..a8d20b7e569 --- /dev/null +++ b/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 { + 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), + }); + }); + } +} diff --git a/packages/cli/src/modules/agents/scheduling/task-timezone.ts b/packages/cli/src/modules/agents/scheduling/task-timezone.ts new file mode 100644 index 00000000000..6d9ca072dd3 --- /dev/null +++ b/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; +} diff --git a/packages/cli/test/integration/scheduling/agent-task-multi-main.test.ts b/packages/cli/test/integration/scheduling/agent-task-multi-main.test.ts new file mode 100644 index 00000000000..da94904d108 --- /dev/null +++ b/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(); + + 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), + ); + 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({ 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(); + }); +}); diff --git a/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts b/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts index 26fe22b305f..4dc537ac4e4 100644 --- a/packages/cli/test/integration/scheduling/scheduled-repositories.test.ts +++ b/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');