diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts index 8366d47c7eb..5e358b27304 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution-orchestrator.service.test.ts @@ -330,7 +330,7 @@ describe('AgentExecutionOrchestratorService', () => { it('waits for a foreign claimed row to end and claims again on conflict', async () => { const { service, executionService, turnCoordinator, permitFor } = makeService(); executionService.startClaimedExecutionRecording - .mockRejectedValueOnce(new AgentThreadClaimConflictError('thread-1')) + .mockRejectedValueOnce(new AgentThreadClaimConflictError()) .mockResolvedValueOnce({ executionId: 'execution-2', claimLost: new AbortController().signal, @@ -365,7 +365,7 @@ describe('AgentExecutionOrchestratorService', () => { const { service, executionService, turnCoordinator, permitFor } = makeService(); const permit = await permitFor('thread-1'); executionService.startClaimedExecutionRecording.mockRejectedValue( - new AgentThreadClaimConflictError('thread-1'), + new AgentThreadClaimConflictError(), ); turnCoordinator.executionRepository.existsRunningByThread.mockImplementationOnce(async () => { turnCoordinator.leases.at(-1)?.abort(); @@ -388,7 +388,7 @@ describe('AgentExecutionOrchestratorService', () => { permit, }), ), - ).rejects.toThrow('Agent thread lease was lost before the turn could claim its thread'); + ).rejects.toThrow('Agent thread lease was lost'); expect(executionService.startClaimedExecutionRecording).toHaveBeenCalledOnce(); expect(runtime.agent.stream).not.toHaveBeenCalled(); diff --git a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts index cdf1cdeece7..6c2dc2508a5 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-execution.service.test.ts @@ -73,7 +73,6 @@ describe('AgentExecutionService', () => { agentExecutionRepository = mock(); agentExecutionRepository.updateIfRunning.mockResolvedValue(true); - agentExecutionRepository.updateIfAbandoned.mockResolvedValue(true); agentExecutionRepository.updateTimelineIfRunning.mockResolvedValue(true); agentExecutionThreadRepository = mock(); n8nMemory = mock(); @@ -864,7 +863,7 @@ describe('AgentExecutionService', () => { executionUpdateBroadcaster, ); const partial = [{ type: 'text', content: 'Partial', timestamp: 1, endTime: 2 }] as const; - agentExecutionRepository.updateIfAbandoned.mockResolvedValue(true); + agentExecutionRepository.updateIfRunning.mockResolvedValue(true); agentExecutionThreadRepository.findOneBy.mockResolvedValue(makeThread()); await service.finalizeInterruptedExecution({ @@ -883,9 +882,8 @@ describe('AgentExecutionService', () => { executionId: 'execution-1', }), ); - expect(agentExecutionRepository.updateIfAbandoned).toHaveBeenCalledWith( + expect(agentExecutionRepository.updateIfRunning).toHaveBeenCalledWith( 'execution-1', - expect.any(Date), expect.objectContaining({ status: 'interrupted', timeline: partial, @@ -901,6 +899,7 @@ describe('AgentExecutionService', () => { }, }, }), + expect.any(Date), ); expect(agentExecutionLogStore.write).not.toHaveBeenCalled(); }); diff --git a/packages/cli/src/modules/agents/__tests__/agent-thread-turn-coordinator.test.ts b/packages/cli/src/modules/agents/__tests__/agent-thread-turn-coordinator.test.ts index dc2088dde46..60f07bd6d39 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-thread-turn-coordinator.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-thread-turn-coordinator.test.ts @@ -127,8 +127,8 @@ describe('AgentThreadTurnCoordinator', () => { const run = coordinator.run('t1', undefined, body); await flush(); - leases[0].abort(new Error('lock lost')); - await expect(run).rejects.toThrow('lock lost'); + leases[0].abort(); + await expect(run).rejects.toThrow('Agent thread lease was lost'); expect(body).not.toHaveBeenCalled(); executionRepository.existsRunningByThread.mockResolvedValue(false); @@ -157,7 +157,7 @@ describe('AgentThreadTurnCoordinator', () => { const stream = coordinator.stream('t1', undefined, async function* (permit) { expect(permit.threadId).toBe('t1'); yield 'one'; - leases[0].abort(new Error('lock lost')); + leases[0].abort(); expect(permit.leaseLost.aborted).toBe(true); yield 'two'; }); diff --git a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts index 6824f278645..a619263a45b 100644 --- a/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts +++ b/packages/cli/src/modules/agents/agent-execution-orchestrator.service.ts @@ -902,6 +902,9 @@ export class AgentExecutionOrchestratorService { const integrationType = isDraft ? N8N_CHAT_INTEGRATION_TYPE : identity.integrationType; return await this.turnCoordinator.run(memory.threadId, abortSignal, async (permit) => { + // A suspended thread belongs to its pending resume, which may have + // started while this wake waited. Execution rows keep `suspended` + // after a resume, so the checkpoint store decides. if ( (await this.agentExecutionService.hasSuspendedRun(memory.threadId)) && (await this.n8nCheckpointStorage.findSuspendedForThread(agentId, memory.threadId)) !== null @@ -1236,14 +1239,10 @@ export class AgentExecutionOrchestratorService { permit: AgentThreadTurnPermit, abortSignal?: AbortSignal, ): Promise<{ executionId: string; abortSignal: AbortSignal }> { - // The lock service aborts `leaseLost` without a reason; name the cause here. - const leaseLostError = () => - new OperationalError('Agent thread lease was lost before the turn could claim its thread'); const waitSignal = AbortSignal.any( [abortSignal, permit.leaseLost].filter((s) => s !== undefined), ); for (;;) { - if (permit.leaseLost.aborted) throw leaseLostError(); waitSignal.throwIfAborted(); try { const { executionId, claimLost } = @@ -1260,11 +1259,7 @@ export class AgentExecutionOrchestratorService { agentId: params.agentId, threadId: params.threadId, }); - try { - await this.turnCoordinator.waitUntilThreadIdle(params.threadId, waitSignal); - } catch (waitError) { - throw permit.leaseLost.aborted ? leaseLostError() : waitError; - } + await this.turnCoordinator.waitUntilThreadIdle(params.threadId, waitSignal); } } } diff --git a/packages/cli/src/modules/agents/agent-execution.service.ts b/packages/cli/src/modules/agents/agent-execution.service.ts index c1d70b46184..fa68580d41c 100644 --- a/packages/cli/src/modules/agents/agent-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-execution.service.ts @@ -265,10 +265,8 @@ export class AgentExecutionService { const duration = execution.startedAt ? Math.max(0, stoppedAt.getTime() - execution.startedAt.getTime()) : 0; - const staleBefore = new Date(Date.now() - AgentExecutionService.livenessGraceMs); - const finalized = await this.agentExecutionRepository.updateIfAbandoned( + const finalized = await this.agentExecutionRepository.updateIfRunning( execution.id, - staleBefore, { status: 'interrupted', stoppedAt, @@ -283,6 +281,7 @@ export class AgentExecutionService { stoppedAt: stoppedAt.getTime(), }), }, + new Date(Date.now() - AgentExecutionService.livenessGraceMs), ); if (finalized) void this.notifyInterruptedExecution(execution); return finalized; diff --git a/packages/cli/src/modules/agents/agent-thread-turn-coordinator.ts b/packages/cli/src/modules/agents/agent-thread-turn-coordinator.ts index 4261449144b..3e6ad45c107 100644 --- a/packages/cli/src/modules/agents/agent-thread-turn-coordinator.ts +++ b/packages/cli/src/modules/agents/agent-thread-turn-coordinator.ts @@ -3,7 +3,7 @@ import { Service } from '@n8n/di'; import { ensureError } from '@n8n/utils/errors/ensure-error'; import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise'; import { sleep } from '@n8n/utils/sleep'; -import { UserError } from 'n8n-workflow'; +import { OperationalError, UserError } from 'n8n-workflow'; import { AgentExecutionRepository } from './repositories/agent-execution.repository'; @@ -44,10 +44,6 @@ export interface AgentThreadTurnLease { release: () => Promise; } -interface Waiter { - resolve: () => void; -} - /** * Admits one agent turn per durable thread at a time. * @@ -61,7 +57,7 @@ interface Waiter { @Service() export class AgentThreadTurnCoordinator { /** Threads with an active turn in this process, with their waiting turns in arrival order. */ - private readonly waiters = new Map(); + private readonly waiters = new Map void>>(); constructor( private readonly lockService: LockService, @@ -111,8 +107,7 @@ export class AgentThreadTurnCoordinator { [signal, lease.leaseLost].filter((s) => s !== undefined), ); await this.waitUntilThreadIdle(threadId, idleWaitSignal); - signal?.throwIfAborted(); - lease.leaseLost.throwIfAborted(); + idleWaitSignal.throwIfAborted(); } catch (error) { await lease?.release(); this.releaseLocally(threadId); @@ -152,11 +147,9 @@ export class AgentThreadTurnCoordinator { if (index !== -1) queue.splice(index, 1); reject(ensureError(signal?.reason)); }; - const waiter: Waiter = { - resolve: () => { - signal?.removeEventListener('abort', onAbort); - resolve(); - }, + const waiter = () => { + signal?.removeEventListener('abort', onAbort); + resolve(); }; queue.push(waiter); signal?.addEventListener('abort', onAbort, { once: true }); @@ -166,28 +159,35 @@ export class AgentThreadTurnCoordinator { /** Hand the turn to the next local waiter, or forget the thread when idle. */ private releaseLocally(threadId: string): void { const next = this.waiters.get(threadId)?.shift(); - if (next) next.resolve(); + if (next) next(); else this.waiters.delete(threadId); } /** * Hold the distributed lease as a handle instead of a callback scope, so a - * generator can keep it until its consumer is done. + * generator can keep it until its consumer is done. The lock service aborts + * without a reason, so `leaseLost` carries one for every consumer. */ private async acquireLease( threadId: string, ): Promise<{ leaseLost: AbortSignal; release: () => Promise }> { - const acquired = createDeferredPromise(); + const acquired = createDeferredPromise(); const released = createDeferredPromise(); + const lost = new AbortController(); const held = this.lockService .withLease(LockNamespace.KNOWN_LOCKS, `agent-thread-turn:${threadId}`, async (signal) => { - acquired.resolve(signal); + signal.addEventListener( + 'abort', + () => lost.abort(new OperationalError('Agent thread lease was lost')), + { once: true }, + ); + acquired.resolve(); await released.promise; }) .catch((error: Error) => acquired.reject(error)); - const leaseLost = await acquired.promise; + await acquired.promise; return { - leaseLost, + leaseLost: lost.signal, release: async () => { released.resolve(); await held; diff --git a/packages/cli/src/modules/agents/background/__tests__/agent-wake.service.test.ts b/packages/cli/src/modules/agents/background/__tests__/agent-wake.service.test.ts index f1ce6ce5b86..df1a0b649d2 100644 --- a/packages/cli/src/modules/agents/background/__tests__/agent-wake.service.test.ts +++ b/packages/cli/src/modules/agents/background/__tests__/agent-wake.service.test.ts @@ -13,9 +13,7 @@ import { hashAgentSandboxPrincipal } from '../../agent-sandbox-principal'; import { AgentThreadQueueFullError } from '../../agent-thread-turn-coordinator'; import type { AgentBackgroundJob } from '../../entities/agent-background-job.entity'; import type { ChatIntegrationRegistry } from '../../integrations/agent-chat-integration'; -import type { N8NCheckpointStorage } from '../../integrations/n8n-checkpoint-storage'; import type { AgentBackgroundJobRepository } from '../../repositories/agent-background-job.repository'; -import type { AgentExecutionRepository } from '../../repositories/agent-execution.repository'; import type { AgentRepository } from '../../repositories/agent.repository'; import { AgentWakeService, @@ -60,10 +58,8 @@ function makeJob(overrides: Partial = {}): AgentBackgroundJo function setup(options: { worker?: boolean; enabled?: boolean } = {}) { const jobRepository = mock(); - const executionRepository = mock(); const agentRepository = mock(); const userRepository = mock(); - const checkpointStorage = mock(); const integrationRegistry = mock(); const orchestrator = mock(); const lockService = mock(); @@ -74,9 +70,6 @@ function setup(options: { worker?: boolean; enabled?: boolean } = {}) { logger.scoped.mockReturnValue(logger); jobRepository.findWakeableUnconsumedSettled.mockResolvedValue([makeJob()]); - executionRepository.existsRunningByThread.mockResolvedValue(false); - executionRepository.hasSuspendedRun.mockResolvedValue(false); - checkpointStorage.findSuspendedForThread.mockResolvedValue(null); agentRepository.findById.mockResolvedValue({ id: 'agent-1', projectId: 'project-1' } as never); userRepository.findByIdWithRole.mockResolvedValue(user as never); integrationRegistry.get.mockReturnValue({} as never); @@ -95,10 +88,8 @@ function setup(options: { worker?: boolean; enabled?: boolean } = {}) { const service = new AgentWakeService( jobRepository, - executionRepository, agentRepository, userRepository, - checkpointStorage, integrationRegistry, orchestrator, lockService, @@ -111,10 +102,8 @@ function setup(options: { worker?: boolean; enabled?: boolean } = {}) { return { service, jobRepository, - executionRepository, agentRepository, userRepository, - checkpointStorage, orchestrator, lockService, publisher, @@ -357,21 +346,6 @@ describe('AgentWakeService', () => { } }); - it('wakes behind a running parent turn but not a suspended one', async () => { - // The orchestrator queues the wake behind the running turn; the service no longer skips it. - const running = setup(); - running.executionRepository.existsRunningByThread.mockResolvedValue(true); - await running.service.attemptWake('thread-1'); - expect(running.orchestrator.executeForWake).toHaveBeenCalledTimes(1); - expect(running.jobRepository.markMailConsumed).toHaveBeenCalledWith('thread-1', ['job-1']); - - const suspended = setup(); - suspended.executionRepository.hasSuspendedRun.mockResolvedValue(true); - suspended.checkpointStorage.findSuspendedForThread.mockResolvedValue({} as never); - await suspended.service.attemptWake('thread-1'); - expect(suspended.orchestrator.executeForWake).not.toHaveBeenCalled(); - }); - it('leaves results pending without recording a failure when the admitted wake is skipped', async () => { const { service, orchestrator, jobRepository, logger } = setup(); orchestrator.executeForWake.mockResolvedValue('skipped'); @@ -396,23 +370,6 @@ describe('AgentWakeService', () => { expect(logger.warn).not.toHaveBeenCalled(); }); - it('checks checkpoints only for threads with a recorded suspension', async () => { - const never = setup(); - await never.service.attemptWake('thread-1'); - expect(never.checkpointStorage.findSuspendedForThread).not.toHaveBeenCalled(); - expect(never.orchestrator.executeForWake).toHaveBeenCalledTimes(1); - - // A resumed run keeps its suspended status. Only an active checkpoint blocks the wake. - const resumed = setup(); - resumed.executionRepository.hasSuspendedRun.mockResolvedValue(true); - await resumed.service.attemptWake('thread-1'); - expect(resumed.checkpointStorage.findSuspendedForThread).toHaveBeenCalledWith( - 'agent-1', - 'thread-1', - ); - expect(resumed.orchestrator.executeForWake).toHaveBeenCalledTimes(1); - }); - it('does not mark results as delivered after the wake loses its lease', async () => { const { service, lockService, orchestrator, jobRepository } = setup(); const controller = new AbortController(); diff --git a/packages/cli/src/modules/agents/background/agent-wake.service.ts b/packages/cli/src/modules/agents/background/agent-wake.service.ts index ac33b900bb5..90113d10b4f 100644 --- a/packages/cli/src/modules/agents/background/agent-wake.service.ts +++ b/packages/cli/src/modules/agents/background/agent-wake.service.ts @@ -22,9 +22,7 @@ import { } from './background-job-messages'; import type { AgentBackgroundJob } from '../entities/agent-background-job.entity'; import { ChatIntegrationRegistry } from '../integrations/agent-chat-integration'; -import { N8NCheckpointStorage } from '../integrations/n8n-checkpoint-storage'; import { AgentBackgroundJobRepository } from '../repositories/agent-background-job.repository'; -import { AgentExecutionRepository } from '../repositories/agent-execution.repository'; import { AgentRepository } from '../repositories/agent.repository'; import { integrationTypeFromMemoryResourceId, @@ -51,10 +49,8 @@ export class AgentWakeService { constructor( private readonly jobRepository: AgentBackgroundJobRepository, - private readonly executionRepository: AgentExecutionRepository, private readonly agentRepository: AgentRepository, private readonly userRepository: UserRepository, - private readonly checkpointStorage: N8NCheckpointStorage, private readonly integrationRegistry: ChatIntegrationRegistry, private readonly orchestrator: AgentExecutionOrchestratorService, private readonly lockService: LockService, @@ -166,17 +162,6 @@ export class AgentWakeService { return; } - // A running parent is no reason to skip: the wake queues behind it as a - // thread turn. A suspended parent is, because the resume owns the thread. - // Execution records keep their suspended status after a resume, so check - // the checkpoint store to determine whether the thread is still suspended. - if ( - (await this.executionRepository.hasSuspendedRun(threadId)) && - (await this.checkpointStorage.findSuspendedForThread(first.parentAgentId, threadId)) !== null - ) { - return; - } - const agent = await this.agentRepository.findById(first.parentAgentId); if (!agent) { this.recordFailure(threadId, generation, 'Background job parent agent no longer exists'); diff --git a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts index 7d59460c71e..368b0750ad3 100644 --- a/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts +++ b/packages/cli/src/modules/agents/repositories/agent-execution.repository.ts @@ -14,7 +14,7 @@ export type RunningAgentExecution = Pick< /** Another running turn already claims this thread. The caller waits and retries. */ export class AgentThreadClaimConflictError extends OperationalError { - constructor(readonly threadId: string) { + constructor() { super('Another agent turn already holds this thread', { level: 'info' }); } } @@ -66,7 +66,7 @@ export class AgentExecutionRepository extends Repository { return await this.save(this.create(values)); } catch (error) { if (values.activeThreadId !== null && isUniqueConstraintError(error)) { - throw new AgentThreadClaimConflictError(values.activeThreadId); + throw new AgentThreadClaimConflictError(); } throw error; } @@ -96,26 +96,14 @@ export class AgentExecutionRepository extends Repository { return result.affected === 1; } - /** Move a running row to a terminal status. This also releases its thread claim. */ + /** + * Move a running row to a terminal status. This also releases its thread + * claim. With `staleBefore`, only a row without a heartbeat since then + * matches, so a run that is alive again keeps its claim. + */ async updateIfRunning( executionId: string, values: AgentExecutionFinalizationValues, - ): Promise { - return await this.finalizeRunning(executionId, values); - } - - /** Finalize a running row only while its liveness timestamp remains stale. */ - async updateIfAbandoned( - executionId: string, - staleBefore: Date, - values: AgentExecutionFinalizationValues, - ): Promise { - return await this.finalizeRunning(executionId, values, staleBefore); - } - - private async finalizeRunning( - executionId: string, - values: AgentExecutionFinalizationValues, staleBefore?: Date, ): Promise { const result = await this.update( diff --git a/packages/cli/test/integration/database/repositories/agent-background-job.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-background-job.repository.test.ts index e50618c3ad8..c1aaa21d2d4 100644 --- a/packages/cli/test/integration/database/repositories/agent-background-job.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/agent-background-job.repository.test.ts @@ -15,7 +15,6 @@ import { AgentBackgroundJobService } from '@/modules/agents/background/agent-bac import { AgentWakeService, WAKE_DEBOUNCE_MS } from '@/modules/agents/background/agent-wake.service'; import type { AgentBackgroundJob } from '@/modules/agents/entities/agent-background-job.entity'; import type { Agent } from '@/modules/agents/entities/agent.entity'; -import type { N8NCheckpointStorage } from '@/modules/agents/integrations/n8n-checkpoint-storage'; import type { ChatIntegrationRegistry } from '@/modules/agents/integrations/agent-chat-integration'; import { AgentBackgroundJobRepository } from '@/modules/agents/repositories/agent-background-job.repository'; import type { AgentExecutionRepository } from '@/modules/agents/repositories/agent-execution.repository'; @@ -175,9 +174,6 @@ describe('AgentBackgroundJobRepository', () => { }); const executionRepository = mock(); - executionRepository.existsRunningByThread.mockResolvedValue(false); - const checkpointStorage = mock(); - checkpointStorage.findSuspendedForThread.mockResolvedValue(null); const orchestrator = mock(); let firstWakeStarted!: () => void; const firstWake = new Promise((resolve) => (firstWakeStarted = resolve)); @@ -202,10 +198,8 @@ describe('AgentBackgroundJobRepository', () => { }); const wakeService = new AgentWakeService( repository, - executionRepository, agentRepository, userRepository, - checkpointStorage, mock(), orchestrator, lockService, diff --git a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts index cb9b01c7010..4266afb414c 100644 --- a/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts +++ b/packages/cli/test/integration/database/repositories/agent-execution.repository.test.ts @@ -295,9 +295,9 @@ describe('AgentExecutionRepository', () => { { id: execution.id }, { updatedAt: new Date('2026-01-02T00:00:01Z') }, ); - expect( - await repository.updateIfAbandoned(execution.id, staleBefore, finalizationValues), - ).toBe(false); + expect(await repository.updateIfRunning(execution.id, finalizationValues, staleBefore)).toBe( + false, + ); expect(await repository.findOneByOrFail({ id: execution.id })).toMatchObject({ status: 'running', activeThreadId: thread.id, @@ -308,9 +308,9 @@ describe('AgentExecutionRepository', () => { { id: execution.id }, { updatedAt: new Date('2026-01-01T23:59:59Z') }, ); - expect( - await repository.updateIfAbandoned(execution.id, staleBefore, finalizationValues), - ).toBe(true); + expect(await repository.updateIfRunning(execution.id, finalizationValues, staleBefore)).toBe( + true, + ); expect(await repository.findOneByOrFail({ id: execution.id })).toMatchObject({ status: 'interrupted', activeThreadId: null,