From fb65257ca54956956f2503a154cfdfe551903a67 Mon Sep 17 00:00:00 2001 From: Robin Braumann Date: Sat, 12 Sep 2026 12:07:30 +0200 Subject: [PATCH] refactor(core): Simplify turn lease loss and claim finalization Abort the thread lease signal with a reason in the coordinator, so the orchestrator no longer names lease loss itself. Merge the three finalization methods of the execution repository into one with an optional stale guard. Remove the wake service suspension pre-check; the orchestrator keeps the post-admission check. Drop the unused conflict error field and the one-field waiter interface. Co-authored-by: Cursor --- ...ent-execution-orchestrator.service.test.ts | 6 +-- .../__tests__/agent-execution.service.test.ts | 7 ++- .../agent-thread-turn-coordinator.test.ts | 6 +-- .../agent-execution-orchestrator.service.ts | 13 ++---- .../modules/agents/agent-execution.service.ts | 5 +-- .../agents/agent-thread-turn-coordinator.ts | 38 ++++++++-------- .../__tests__/agent-wake.service.test.ts | 43 ------------------- .../agents/background/agent-wake.service.ts | 15 ------- .../agent-execution.repository.ts | 26 +++-------- .../agent-background-job.repository.test.ts | 6 --- .../agent-execution.repository.test.ts | 12 +++--- 11 files changed, 47 insertions(+), 130 deletions(-) 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,