feat(engine): Route poll trigger runs to the engine 2.0 data plane (#37803)

packages/cli/src/services/__tests__/engine-v2-payload-guard.service.test.ts

@@ -34,21 +34,21 @@ describe('EngineV2PayloadGuard', () => {
 			{ name: 'a null slot', slots: [null] },
 			{ name: 'items with no binary key', slots: [[{ json: {} }]] },
 			{ name: 'an empty binary map', slots: [[{ json: {}, binary: {} }]] },
-		])('allows a payload with $name', async ({ slots }) => {
-			await expect(guard.assertNoFiles(slots, REASON)).resolves.toBeUndefined();
+		])('allows a payload with $name', ({ slots }) => {
+			expect(() => guard.assertNoFiles(slots, REASON)).not.toThrow();
 			expect(binaryDataService.deleteManyByBinaryDataId).not.toHaveBeenCalled();
 		});
 
-		it('refuses a stored file and deletes it, because no execution will own it', async () => {
+		it('refuses a stored file and deletes it, because no execution will own it', () => {
 			const slots = [[{ json: {}, binary: { attachment: stored('filesystem:abc') } }]];
 
-			await expect(guard.assertNoFiles(slots, REASON)).rejects.toThrow(REASON);
+			expect(() => guard.assertNoFiles(slots, REASON)).toThrow(REASON);
 			expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledExactlyOnceWith([
 				'filesystem:abc',
 			]);
 		});
 
-		it('deletes every stored file across slots and items', async () => {
+		it('deletes every stored file across slots and items', () => {
 			const slots: Array<INodeExecutionData[] | null> = [
 				[
 					{ json: {}, binary: { a: stored('filesystem:1') } },
@@ -58,7 +58,7 @@ describe('EngineV2PayloadGuard', () => {
 				[{ json: {}, binary: { d: stored('filesystem:3') } }],
 			];
 
-			await expect(guard.assertNoFiles(slots, REASON)).rejects.toThrow(UserError);
+			expect(() => guard.assertNoFiles(slots, REASON)).toThrow(UserError);
 			expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledExactlyOnceWith([
 				'filesystem:1',
 				'filesystem:2',
@@ -66,10 +66,10 @@ describe('EngineV2PayloadGuard', () => {
 			]);
 		});
 
-		it('refuses an in-memory file without a delete, as it has no id', async () => {
+		it('refuses an in-memory file without a delete, as it has no id', () => {
 			const slots = [[{ json: {}, binary: { attachment: inMemory() } }]];
 
-			await expect(guard.assertNoFiles(slots, REASON)).rejects.toThrow(REASON);
+			expect(() => guard.assertNoFiles(slots, REASON)).toThrow(REASON);
 			expect(binaryDataService.deleteManyByBinaryDataId).not.toHaveBeenCalled();
 		});
 
@@ -77,35 +77,13 @@ describe('EngineV2PayloadGuard', () => {
 			binaryDataService.deleteManyByBinaryDataId.mockRejectedValue(new Error('store is down'));
 			const slots = [[{ json: {}, binary: { attachment: stored('s3:abc') } }]];
 
-			await expect(guard.assertNoFiles(slots, REASON)).rejects.toThrow(REASON);
-			expect(logger.error).toHaveBeenCalledWith(
-				'Failed to delete the files of a rejected engine 2.0 payload',
-				expect.objectContaining({ error: expect.any(Error) }),
+			expect(() => guard.assertNoFiles(slots, REASON)).toThrow(REASON);
+			await vi.waitFor(() =>
+				expect(logger.error).toHaveBeenCalledWith(
+					'Failed to delete the files of a rejected engine 2.0 payload',
+					expect.objectContaining({ error: expect.any(Error) }),
+				),
 			);
 		});
 	});
-
-	describe('discardFiles', () => {
-		it('deletes the stored files without refusing', async () => {
-			const slots = [[{ json: {}, binary: { attachment: stored('filesystem:abc') } }]];
-
-			await expect(guard.discardFiles(slots)).resolves.toBeUndefined();
-			expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledExactlyOnceWith([
-				'filesystem:abc',
-			]);
-		});
-
-		it('does nothing for a payload with no files', async () => {
-			await expect(guard.discardFiles([[{ json: {} }]])).resolves.toBeUndefined();
-			expect(binaryDataService.deleteManyByBinaryDataId).not.toHaveBeenCalled();
-		});
-
-		it('never throws when the delete fails, so the caller keeps its own answer', async () => {
-			binaryDataService.deleteManyByBinaryDataId.mockRejectedValue(new Error('store is down'));
-			const slots = [[{ json: {}, binary: { attachment: stored('s3:abc') } }]];
-
-			await expect(guard.discardFiles(slots)).resolves.toBeUndefined();
-			expect(logger.error).toHaveBeenCalled();
-		});
-	});
 });

packages/cli/src/services/engine-v2-payload-guard.service.ts

@@ -32,27 +32,21 @@ export class EngineV2PayloadGuard {
 	 *
 	 * `reason` names the surface, so the user hears which trigger could not be
 	 * served rather than a generic message.
+	 *
+	 * Refuses synchronously, so a caller that cannot await (a poll's `__emit`)
+	 * still refuses before doing anything that only makes sense for a run it is
+	 * about to start. The delete runs detached: nobody reads its result, and
+	 * {@link deleteStoredFiles} never throws.
 	 */
-	async assertNoFiles(slots: PayloadSlots, reason: string): Promise<void> {
+	assertNoFiles(slots: PayloadSlots, reason: string): void {
 		const files = this.filesIn(slots);
 		if (files.length === 0) return;
 
-		await this.deleteStoredFiles(files);
+		void this.deleteStoredFiles(files);
 
 		throw new UserError(reason);
 	}
 
-	/**
-	 * Deletes the files of a payload that is being discarded for some other
-	 * reason. Never throws: the caller's own refusal is the answer.
-	 */
-	async discardFiles(slots: PayloadSlots): Promise<void> {
-		const files = this.filesIn(slots);
-		if (files.length === 0) return;
-
-		await this.deleteStoredFiles(files);
-	}
-
 	/** An empty `binary` map carries no file, so it does not count. */
 	private filesIn(slots: PayloadSlots): IBinaryData[] {
 		return slots.flatMap((slot) => slot ?? []).flatMap((item) => Object.values(item.binary ?? {}));
@@ -60,8 +54,9 @@ export class EngineV2PayloadGuard {
 
 	/**
 	 * Only stored modes give a file an id; in memory the data rides on the item and
-	 * there is nothing to delete. A failed delete leaks a file, which must not
-	 * replace the caller's reason with a storage error.
+	 * there is nothing to delete. Never throws: a failed delete leaks a file, which
+	 * must not replace the caller's reason with a storage error, and callers detach
+	 * this call.
 	 */
 	private async deleteStoredFiles(files: IBinaryData[]): Promise<void> {
 		const storedIds = files.map((file) => file.id).filter((id) => id !== undefined);

packages/cli/src/webhooks/engine-v2-webhooks.ts

@@ -113,8 +113,8 @@ export class EngineV2Webhooks {
 	 * Only the webhook node's own output says whether the request brought a file,
 	 * so this runs after the node, unlike {@link assertSupported}.
 	 */
-	async assertPayloadSupported(webhookResultData: IWebhookResponseData): Promise<void> {
-		await this.payloadGuard.assertNoFiles(
+	assertPayloadSupported(webhookResultData: IWebhookResponseData): void {
+		this.payloadGuard.assertNoFiles(
 			webhookResultData.workflowData ?? [],
 			'Engine 2.0 cannot receive files from a webhook yet.',
 		);

packages/cli/src/webhooks/webhook-helpers.ts

@@ -916,7 +916,7 @@ export async function executeWebhook(
 
 		// The node's output is the only place a file shows up, so this cannot run with
 		// the checks above.
-		if (routesToEngineV2) await engineV2Webhooks.assertPayloadSupported(webhookResultData);
+		if (routesToEngineV2) engineV2Webhooks.assertPayloadSupported(webhookResultData);
 
 		// Reactive credential-status gate. Runs only once we know the workflow will
 		// execute (workflowData is defined), so a falsy "Only Run If" short-circuits

packages/cli/src/workflows/__tests__/workflow-execution.service.test.ts

@@ -414,6 +414,108 @@ describe('WorkflowExecutionService', () => {
 		});
 	});
 
+	describe('runPolledWorkflowV2()', () => {
+		const node = mock<INode>({ id: 'node-1', name: 'Poll Node' });
+		const workflow = mock<IWorkflowBase>({
+			id: 'wf-1',
+			active: true,
+			activeVersionId: 'some-version-id',
+			nodes: [node],
+		});
+		const cursor = { lastItemId: 'a' };
+		const pollItems = [[{ json: { id: 1 } }]];
+		let responsePromise: MockProxy<IDeferredPromise<IExecuteResponsePromiseData>>;
+
+		const runPolledWorkflowV2 = async () =>
+			await workflowExecutionService.runPolledWorkflowV2(
+				workflow,
+				node,
+				pollItems,
+				additionalData,
+				'trigger',
+				cursor,
+				responsePromise,
+			);
+
+		beforeEach(() => {
+			vi.clearAllMocks();
+			responsePromise = mock<IDeferredPromise<IExecuteResponsePromiseData>>();
+			workflowRunner.run.mockResolvedValue('exec-v2');
+			pollCursorService.commitCursorOnly.mockResolvedValue(true);
+		});
+
+		test('starts the run before committing the cursor, and returns its id', async () => {
+			const returned = await runPolledWorkflowV2();
+
+			expect(returned).toBe('exec-v2');
+			expect(workflowRunner.run).toHaveBeenCalledWith(
+				expect.objectContaining({
+					workflowData: workflow,
+					executionData: expect.objectContaining({
+						executionData: expect.objectContaining({
+							nodeExecutionStack: [{ node, data: { main: pollItems }, source: null }],
+						}),
+					}),
+				}),
+				true,
+				undefined,
+				undefined,
+				responsePromise,
+			);
+			expect(pollCursorService.commitCursorOnly).toHaveBeenCalledWith({
+				workflowId: 'wf-1',
+				nodeId: 'node-1',
+				cursor,
+				fence: undefined,
+			});
+			// The cursor must not advance until the data plane has confirmed the
+			// run: a commit before that could advance past items nothing carried.
+			expect(workflowRunner.run.mock.invocationCallOrder[0]).toBeLessThan(
+				pollCursorService.commitCursorOnly.mock.invocationCallOrder[0],
+			);
+		});
+
+		test('never commits the cursor when the run fails to start', async () => {
+			const runError = new Error('engine 2.0 rejected the run');
+			workflowRunner.run.mockRejectedValue(runError);
+
+			await expect(runPolledWorkflowV2()).rejects.toBe(runError);
+
+			expect(pollCursorService.commitCursorOnly).not.toHaveBeenCalled();
+		});
+
+		test('logs, but does not fail the run, when the lease is gone by the time it commits', async () => {
+			pollCursorService.commitCursorOnly.mockResolvedValue(false);
+
+			const returned = await runPolledWorkflowV2();
+
+			expect(returned).toBe('exec-v2');
+			expect(logger.warn).toHaveBeenCalledWith(
+				'Poll cursor commit skipped after its execution already started: the poll no longer holds its lease',
+				{ workflowId: 'wf-1', nodeId: 'node-1', nodeName: 'Poll Node', executionId: 'exec-v2' },
+			);
+		});
+
+		test('passes the fence through to the cursor commit', async () => {
+			const fence = { taskId: 'task-1', leaseEpoch: 3 };
+
+			await workflowExecutionService.runPolledWorkflowV2(
+				workflow,
+				node,
+				pollItems,
+				additionalData,
+				'trigger',
+				cursor,
+				responsePromise,
+				fence,
+			);
+
+			expect(pollCursorService.commitCursorOnly).toHaveBeenCalledWith(
+				expect.objectContaining({ fence }),
+			);
+		});
+	});
+
 	describe('executeManually()', () => {
 		beforeEach(() => {
 			workflowRunner.run.mockClear();

packages/cli/src/workflows/triggers/__tests__/engine-v2-active-triggers.test.ts

@@ -16,8 +16,7 @@ describe('EngineV2ActiveTriggers', () => {
 
 	beforeEach(() => {
 		vi.clearAllMocks();
-		payloadGuard.assertNoFiles.mockResolvedValue(undefined);
-		payloadGuard.discardFiles.mockResolvedValue(undefined);
+		payloadGuard.assertNoFiles.mockReturnValue(undefined);
 		engineV2ActiveTriggers = new EngineV2ActiveTriggers(dispatcher, payloadGuard);
 	});
 
@@ -60,10 +59,10 @@ describe('EngineV2ActiveTriggers', () => {
 	});
 
 	describe('assertPayloadSupported', () => {
-		it('asks the guard to refuse files, naming the trigger surface', async () => {
+		it('asks the guard to refuse files, naming the trigger surface', () => {
 			const slots = [[{ json: {}, binary: { data: mock<IBinaryData>() } }]];
 
-			await engineV2ActiveTriggers.assertPayloadSupported(slots);
+			engineV2ActiveTriggers.assertPayloadSupported(slots);
 
 			expect(payloadGuard.assertNoFiles).toHaveBeenCalledWith(
 				slots,
@@ -71,29 +70,12 @@ describe('EngineV2ActiveTriggers', () => {
 			);
 		});
 
-		it('surfaces the refusal the guard raises', async () => {
-			payloadGuard.assertNoFiles.mockRejectedValue(new UserError('nope'));
+		it('surfaces the refusal the guard raises, synchronously', () => {
+			payloadGuard.assertNoFiles.mockImplementation(() => {
+				throw new UserError('nope');
+			});
 
-			await expect(engineV2ActiveTriggers.assertPayloadSupported([[]])).rejects.toThrow('nope');
-		});
-	});
-
-	describe('discardFiles', () => {
-		it('hands the payload to the guard', async () => {
-			const slots = [[{ json: {} }]];
-
-			await engineV2ActiveTriggers.discardFiles(slots);
-
-			expect(payloadGuard.discardFiles).toHaveBeenCalledWith(slots);
-		});
-	});
-
-	describe('assertPollSupported', () => {
-		it('always refuses', () => {
-			expect(() => engineV2ActiveTriggers.assertPollSupported()).toThrow(UserError);
-			expect(() => engineV2ActiveTriggers.assertPollSupported()).toThrow(
-				'Engine 2.0 cannot run polling triggers yet.',
-			);
+			expect(() => engineV2ActiveTriggers.assertPayloadSupported([[]])).toThrow('nope');
 		});
 	});
 });

packages/cli/src/workflows/triggers/__tests__/trigger-execution-context.factory.test.ts

@@ -947,6 +947,47 @@ describe('TriggerExecutionContextFactory', () => {
 			expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled();
 		});
 
+		test('routes to runPolledWorkflowV2, not the transactional runPolledWorkflow, on engine 2.0', async () => {
+			engineV2Dispatcher.handlesWorkflow.mockReturnValue(true);
+			workflowExecutionService.runPolledWorkflowV2.mockResolvedValue('exec-v2');
+			const responsePromise = createDeferredPromise<IExecuteResponsePromiseData>();
+
+			await context.__runPoll(async () => {
+				Object.assign(context.getWorkflowStaticData('node'), { lastItemId: 'a' });
+				context.__emit(pollData, responsePromise);
+			});
+			await sleep(0);
+
+			expect(workflowExecutionService.runPolledWorkflowV2).toHaveBeenCalledWith(
+				expect.objectContaining({ id: 'wf-1' }),
+				node,
+				pollData,
+				additionalData,
+				mode,
+				{ lastItemId: 'a' },
+				responsePromise,
+				undefined,
+			);
+			expect(workflowExecutionService.runPolledWorkflow).not.toHaveBeenCalled();
+			expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled();
+		});
+
+		test('refuses a migrated poll that carries a file, and commits no cursor, on engine 2.0', async () => {
+			engineV2Dispatcher.handlesWorkflow.mockReturnValue(true);
+			const storedFile = mock<IBinaryData>({ id: 'filesystem:abc', mimeType: 'text/plain' });
+			const fileData: INodeExecutionData[][] = [[{ json: {}, binary: { attachment: storedFile } }]];
+
+			await context.__runPoll(async () => {
+				Object.assign(context.getWorkflowStaticData('node'), { lastItemId: 'a' });
+				expect(() => context.__emit(fileData)).toThrow(
+					'Engine 2.0 cannot receive files from a trigger yet.',
+				);
+			});
+
+			expect(pollCursorService.commitCursorOnly).not.toHaveBeenCalled();
+			expect(workflowExecutionService.runPolledWorkflowV2).not.toHaveBeenCalled();
+		});
+
 		// A poll that neither emitted nor committed leaves its mutation behind; the next
 		// poll must not pick it up and commit an advance past items nobody carried.
 		const abandonedStaging = [
@@ -1823,16 +1864,58 @@ describe('TriggerExecutionContextFactory', () => {
 				]);
 			});
 
-			test('refuses the emit and commits no cursor', () => {
-				expect(() => pollContext().__emit([[{ json: {} }]])).toThrow(
-					'Engine 2.0 cannot run polling triggers yet.',
+			test('refuses an emit carrying a file, and commits no cursor', () => {
+				const data: INodeExecutionData[][] = [[{ json: {}, binary: { attachment: storedFile } }]];
+
+				expect(() => pollContext().__emit(data)).toThrow(
+					'Engine 2.0 cannot receive files from a trigger yet.',
 				);
 				expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled();
-				expect(workflowExecutionService.runPolledWorkflow).not.toHaveBeenCalled();
+				expect(workflowExecutionService.runPolledWorkflowV2).not.toHaveBeenCalled();
 				expect(pollCursorService.commitWithExecution).not.toHaveBeenCalled();
 				expect(pollCursorService.commitCursorOnly).not.toHaveBeenCalled();
 				expect(workflowStaticDataService.saveStaticData).not.toHaveBeenCalled();
 			});
+
+			test('hands an unmigrated emit to the plain runner, which dispatches it', async () => {
+				pollContext().__emit([[{ json: {} }]]);
+				await sleep(0);
+
+				expect(workflowExecutionService.runWorkflow).toHaveBeenCalled();
+				expect(workflowExecutionService.runPolledWorkflowV2).not.toHaveBeenCalled();
+			});
+
+			test('re-checks the payload against fresh data when the registration snapshot missed the transition to engine 2.0', async () => {
+				const staleWorkflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
+				const freshWorkflowData = mock<WorkflowEntity>({ id: 'wf-1', name: 'Test Workflow' });
+				// The registration snapshot still says v1; only the fresh read reflects
+				// the workflow having since been republished onto engine 2.0.
+				engineV2Dispatcher.handlesWorkflow.mockImplementation((wf) => wf === freshWorkflowData);
+
+				const getPollFunctions = factory.getExecutePollFunctions(
+					staleWorkflowData,
+					additionalData,
+					mode,
+					activation,
+					async () => freshWorkflowData,
+				);
+				const context = getPollFunctions(
+					workflow,
+					mock<INode>({ name: 'Poll Node' }),
+					additionalData,
+					mode,
+					activation,
+				);
+
+				const data: INodeExecutionData[][] = [[{ json: {}, binary: { attachment: storedFile } }]];
+				context.__emit(data);
+				await sleep(0);
+
+				expect(workflowExecutionService.runWorkflow).not.toHaveBeenCalled();
+				expect(binaryDataService.deleteManyByBinaryDataId).toHaveBeenCalledExactlyOnceWith([
+					'filesystem:abc',
+				]);
+			});
 		});
 
 		describe('a workflow that did not opt in', () => {

packages/cli/src/workflows/triggers/engine-v2-active-triggers.ts

@@ -68,30 +68,11 @@ export class EngineV2ActiveTriggers {
 	 * the node's own output says whether it produced one, so this runs on the emit
 	 * rather than at activation. Email Read IMAP is the case that matters today:
 	 * it downloads attachments and passes no promise, so nothing else refuses it.
-	 */
-	async assertPayloadSupported(slots: Array<INodeExecutionData[] | null>): Promise<void> {
-		await this.payloadGuard.assertNoFiles(
-			slots,
-			'Engine 2.0 cannot receive files from a trigger yet.',
-		);
-	}
-
-	/**
-	 * Deletes the files of an emit refused for another reason, so a payload the
-	 * poll path turns away does not leave them behind. Never throws.
-	 */
-	async discardFiles(slots: Array<INodeExecutionData[] | null>): Promise<void> {
-		await this.payloadGuard.discardFiles(slots);
-	}
-
-	/**
-	 * Rejects a polled emit.
 	 *
-	 * A migrated poll node commits its cursor in the same transaction as the
-	 * execution row, which a v2 run does not create.
-	 * TODO(CAT-4078): commit the cursor against a data-plane execution instead.
+	 * Refuses synchronously, so a poll's `__emit` can call it before it takes the
+	 * staged cursor.
 	 */
-	assertPollSupported(): never {
-		throw new UserError('Engine 2.0 cannot run polling triggers yet.');
+	assertPayloadSupported(slots: Array<INodeExecutionData[] | null>): void {
+		this.payloadGuard.assertNoFiles(slots, 'Engine 2.0 cannot receive files from a trigger yet.');
 	}
 }

packages/cli/src/workflows/triggers/trigger-execution-context.factory.ts

@@ -116,15 +116,15 @@ export class TriggerExecutionContextFactory {
 	 * The done promise needs no help here: {@link settleDonePromise} rejects it
 	 * from the returned promise chain.
 	 */
-	private async assertEngineV2Supported(
+	private assertEngineV2Supported(
 		data: INodeExecutionData[][],
 		emit: EngineV2ActiveTriggerEmit,
-	): Promise<void> {
+	): void {
 		try {
 			// Files first, because this check deletes what it refuses: a refusal for
 			// any other reason would otherwise leave the stored files behind, owned by
 			// no execution.
-			await this.engineV2ActiveTriggers.assertPayloadSupported(data);
+			this.engineV2ActiveTriggers.assertPayloadSupported(data);
 			this.engineV2ActiveTriggers.assertSupported(emit);
 		} catch (error) {
 			emit.responsePromise?.reject(ensureError(error));
@@ -209,7 +209,7 @@ export class TriggerExecutionContextFactory {
 						// Checked against the fresh data, so this agrees with the dispatcher,
 						// which decides on the same copy.
 						if (this.engineV2ActiveTriggers.handles(freshWorkflowData, mode)) {
-							await this.assertEngineV2Supported(data, { responsePromise, donePromise });
+							this.assertEngineV2Supported(data, { responsePromise, donePromise });
 						}
 
 						return await this.workflowExecutionService.runWorkflow(
@@ -372,12 +372,7 @@ export class TriggerExecutionContextFactory {
 				// retried. Reads the registration's copy of the workflow rather than the
 				// fresh one for the same reason: the fresh read comes too late.
 				if (this.engineV2ActiveTriggers.handles(workflowData, mode)) {
-					// Detached because `__emit` is synchronous. The poll may have stored
-					// attachments, and no execution will ever own them.
-					void this.engineV2ActiveTriggers
-						.discardFiles(data)
-						.catch((error: unknown) => this.logTriggerExecutionFailure(error, workflowData, node));
-					this.engineV2ActiveTriggers.assertPollSupported();
+					this.engineV2ActiveTriggers.assertPayloadSupported(data);
 				}
 
 				const cursor = takeStagedCursor();
@@ -390,15 +385,42 @@ export class TriggerExecutionContextFactory {
 				// TODO(CAT-3202): resolves workflow data via callback so we
 				// can feature-flag between in-memory data and the published data
 				// service. Once the flag is removed, we'll call the service directly.
-				const executePromise = resolveWorkflowData().then(async (freshWorkflowData) =>
-					cursor === null
-						? await this.workflowExecutionService.runWorkflow(
+				const executePromise = resolveWorkflowData().then(async (freshWorkflowData) => {
+					// The registration snapshot above can be stale by the time this
+					// resolves (e.g. the workflow was just republished onto engine 2.0),
+					// so a payload that slipped past that check is guarded again here,
+					// against the copy that actually decides where this run goes.
+					const routesToV2 = this.engineV2ActiveTriggers.handles(freshWorkflowData, mode);
+					if (routesToV2) {
+						try {
+							this.engineV2ActiveTriggers.assertPayloadSupported(data);
+						} catch (error) {
+							responsePromise?.reject(ensureError(error));
+							throw error;
+						}
+					}
+
+					if (cursor === null) {
+						return await this.workflowExecutionService.runWorkflow(
+							freshWorkflowData,
+							node,
+							data,
+							additionalData,
+							mode,
+							responsePromise,
+						);
+					}
+
+					return routesToV2
+						? await this.workflowExecutionService.runPolledWorkflowV2(
 								freshWorkflowData,
 								node,
 								data,
 								additionalData,
 								mode,
+								cursor,
 								responsePromise,
+								fence,
 							)
 						: await this.workflowExecutionService.runPolledWorkflow(
 								freshWorkflowData,
@@ -409,8 +431,8 @@ export class TriggerExecutionContextFactory {
 								cursor,
 								responsePromise,
 								fence,
-							),
-				);
+							);
+				});
 
 				if (donePromise) this.settleDonePromise(executePromise, donePromise);
 

packages/cli/src/workflows/workflow-execution.service.ts

@@ -259,6 +259,53 @@ export class WorkflowExecutionService {
 		return executionId;
 	}
 
+	/**
+	 * Starts a polled execution on engine 2.0, then advances the cursor.
+	 *
+	 * The v2 path keeps no control-plane execution row, so the cursor cannot
+	 * commit in the same transaction as the run the way {@link runPolledWorkflow}
+	 * does. It commits after the data plane confirms the run started instead: a
+	 * crash between the two calls can duplicate a poll, never lose one.
+	 * TODO(CAT-4078): add a dedup key once `StartExecutionRequest` supports one.
+	 */
+	async runPolledWorkflowV2(
+		workflowData: IWorkflowBase,
+		node: INode,
+		data: INodeExecutionData[][],
+		additionalData: IWorkflowExecuteAdditionalData,
+		mode: WorkflowExecuteMode,
+		cursor: PollCursor,
+		responsePromise?: IDeferredPromise<IExecuteResponsePromiseData>,
+		fence?: PollLeaseFence,
+	): Promise<string> {
+		const executionId = await this.runWorkflow(
+			workflowData,
+			node,
+			data,
+			additionalData,
+			mode,
+			responsePromise,
+		);
+
+		const committed = await this.pollCursorService.commitCursorOnly({
+			workflowId: workflowData.id,
+			nodeId: node.id,
+			cursor,
+			fence,
+		});
+
+		if (!committed) {
+			// The run already started; the lease loss only means a concurrent
+			// poller may repeat the same window, not that anything was lost.
+			this.logger.warn(
+				'Poll cursor commit skipped after its execution already started: the poll no longer holds its lease',
+				{ workflowId: workflowData.id, nodeId: node.id, nodeName: node.name, executionId },
+			);
+		}
+
+		return executionId;
+	}
+
 	/**
 	 * Marks a committed row that failed to start as crashed, so it does not sit at
 	 * `new` indefinitely.