feat(engine): Settle steps across loop iterations (no-changelog) (#36604)

packages/@n8n/engine/src/database/__tests__/workflow-step-execution.integration.test.ts

@@ -674,7 +674,7 @@ describe('workflow_step_execution table (integration)', () => {
 		});
 	});
 
-	it('TypeOrmStepStore.loadLatestStep returns the highest-iteration row, or null', async () => {
+	it("TypeOrmStepStore.loadLatestStepSummaries returns each node's highest-iteration row", async () => {
 		const executionId = await createExecution();
 		const store = new TypeOrmStepStore(dataSource.getRepository(WorkflowStepExecution));
 		await store.createSteps(executionId, [
@@ -685,10 +685,24 @@ describe('workflow_step_execution table (integration)', () => {
 		const otherExecutionId = await createExecution();
 		await store.createSteps(otherExecutionId, [{ nodeId: 'b', iteration: 5, status: 'queued' }]);
 
-		const latest = await store.loadLatestStep(executionId, 'b');
-		expect(latest).toMatchObject({ nodeId: 'b', iteration: 1, status: 'completed' });
+		await store.createSteps(executionId, [
+			{ nodeId: 'c', iteration: 0, status: 'completed', outputs: [[{ json: {} }]] },
+		]);
 
-		expect(await store.loadLatestStep(executionId, 'ghost')).toBeNull();
+		const latest = await store.loadLatestStepSummaries(executionId, ['b', 'c', 'ghost']);
+
+		// one row per node asked about, the slim view: which slots were filled, not
+		// what they hold, and a node with no row is absent rather than null
+		expect(latest.b).toMatchObject({
+			nodeId: 'b',
+			iteration: 1,
+			status: 'completed',
+			filledOutputSlots: [true, false],
+		});
+		expect(latest.c).toMatchObject({ nodeId: 'c', iteration: 0 });
+		expect(latest.ghost).toBeUndefined();
+
+		expect(await store.loadLatestStepSummaries(executionId, [])).toEqual({});
 	});
 
 	it('carries the unique key and the failed-rows partial index in the schema', async () => {

packages/@n8n/engine/src/database/typeorm-step-store.ts

@@ -20,6 +20,26 @@ import {
 	type StepSummary,
 } from '../execution/step-store';
 
+/**
+ * Per output slot: whether the step put data there. Computed in the query, so the
+ * potentially large outputs payload is never transferred. A slot counts as filled
+ * unless it holds JSON null.
+ */
+const FILLED_OUTPUT_SLOTS = `COALESCE(
+	(SELECT array_agg(jsonb_typeof(slot.value) <> 'null' ORDER BY slot.ordinality)
+	 FROM jsonb_array_elements(step.outputs) WITH ORDINALITY AS slot),
+	'{}'
+)`;
+
+/** What both summary queries select: every column but the outputs payload. */
+type StepSummaryRow = {
+	id: string;
+	nodeId: string;
+	iteration: number;
+	status: StepStatus;
+	filledOutputSlots: boolean[];
+};
+
 /** RETURNING rows come back keyed by database column name (snake_case). */
 type InsertedStepRow = { id: string; node_id: string; iteration: number };
 type ClaimedStepRow = { id: string; execution_id: string; node_id: string; iteration: number };
@@ -212,30 +232,14 @@ export class TypeOrmStepStore implements StepStore {
 	): Promise<Record<StepKeyId, StepSummary>> {
 		if (keys.length === 0) return {};
 
-		// The per-slot booleans are computed inside the query, so the potentially
-		// large outputs payloads are never transferred. A slot counts as filled
-		// unless it holds JSON null.
 		const { fragment, parameters } = stepKeyFilter(keys);
-		const rows: Array<{
-			id: string;
-			nodeId: string;
-			iteration: number;
-			status: StepStatus;
-			filledOutputSlots: boolean[];
-		}> = await this.repo
+		const rows: StepSummaryRow[] = await this.repo
 			.createQueryBuilder('step')
 			.select('step.id', 'id')
 			.addSelect('step.node_id', 'nodeId')
 			.addSelect('step.iteration', 'iteration')
 			.addSelect('step.status', 'status')
-			.addSelect(
-				`COALESCE(
-					(SELECT array_agg(jsonb_typeof(slot.value) <> 'null' ORDER BY slot.ordinality)
-					 FROM jsonb_array_elements(step.outputs) WITH ORDINALITY AS slot),
-					'{}'
-				)`,
-				'filledOutputSlots',
-			)
+			.addSelect(FILLED_OUTPUT_SLOTS, 'filledOutputSlots')
 			.where('step.execution_id = :executionId', { executionId })
 			.andWhere(fragment, parameters)
 			.getRawMany();
@@ -243,12 +247,27 @@ export class TypeOrmStepStore implements StepStore {
 		return Object.fromEntries(rows.map((row) => [stepKeyId(row), row]));
 	}
 
-	async loadLatestStep(executionId: string, nodeId: string): Promise<StepRecord | null> {
-		// NOTE: `findOne({ where })`, not `findOneBy`, as in `loadStep`.
-		return await this.repo.findOne({
-			where: { executionId, nodeId },
-			order: { iteration: 'DESC' },
-		});
+	async loadLatestStepSummaries(
+		executionId: string,
+		nodeIds: string[],
+	): Promise<Record<string, StepSummary>> {
+		if (nodeIds.length === 0) return {};
+
+		const rows: StepSummaryRow[] = await this.repo
+			.createQueryBuilder('step')
+			.distinctOn(['step.node_id'])
+			.select('step.id', 'id')
+			.addSelect('step.node_id', 'nodeId')
+			.addSelect('step.iteration', 'iteration')
+			.addSelect('step.status', 'status')
+			.addSelect(FILLED_OUTPUT_SLOTS, 'filledOutputSlots')
+			.where('step.execution_id = :executionId', { executionId })
+			.andWhere('step.node_id IN (:...nodeIds)', { nodeIds })
+			.orderBy('step.node_id')
+			.addOrderBy('step.iteration', 'DESC')
+			.getRawMany();
+
+		return Object.fromEntries(rows.map((row) => [row.nodeId, row]));
 	}
 
 	async countSettledSteps(executionId: string): Promise<number> {

packages/@n8n/engine/src/execution/__tests__/completion.test.ts

@@ -0,0 +1,78 @@
+import { describe, expect, it } from 'vitest';
+
+import type { WorkflowLoop } from '../../graph';
+import { countExpectedSettledSteps } from '../completion';
+
+function loop(batchNodeId: string, memberIds: string[]): WorkflowLoop {
+	return {
+		batchNodeId,
+		memberIds: new Set([batchNodeId, ...memberIds]),
+		backEdges: [],
+		entryEdges: [],
+		exitEdges: [],
+	};
+}
+
+describe('countExpectedSettledSteps', () => {
+	it('counts one row per node when there are no loops', () => {
+		expect(countExpectedSettledSteps([], new Set(['trigger', 'a', 'b']), new Map())).toBe(3);
+	});
+
+	it('counts a loop that ran three iterations', () => {
+		// B has rows 0, 1, 2 and 3, x has rows 0, 1 and 2, since the body has none
+		// at the terminal iteration. Plus trigger and d outside the loop.
+		const loops = [loop('B', ['x'])];
+		const reachable = new Set(['trigger', 'B', 'x', 'd']);
+
+		expect(countExpectedSettledSteps(loops, reachable, new Map([['B', 3]]))).toBe(2 + 4 + 3);
+	});
+
+	it('counts a loop that ended immediately', () => {
+		// zero items: B's row 0 is terminal, and the body never ran
+		const loops = [loop('B', ['x'])];
+		const reachable = new Set(['trigger', 'B', 'x', 'd']);
+
+		expect(countExpectedSettledSteps(loops, reachable, new Map([['B', 0]]))).toBe(2 + 1 + 0);
+	});
+
+	it('counts each member of a longer body per iteration', () => {
+		const loops = [loop('B', ['x', 'y', 'z'])];
+		const reachable = new Set(['trigger', 'B', 'x', 'y', 'z']);
+
+		// B: 0..2 is 3 rows, and x, y and z have 2 rows each
+		expect(countExpectedSettledSteps(loops, reachable, new Map([['B', 2]]))).toBe(1 + 3 + 6);
+	});
+
+	it('owes an unknown number of rows while a loop is still running', () => {
+		const loops = [loop('B', ['x'])];
+		const reachable = new Set(['trigger', 'B', 'x', 'd']);
+
+		expect(countExpectedSettledSteps(loops, reachable, new Map())).toBeUndefined();
+	});
+
+	it('ignores a loop the trigger cannot reach, which never receives rows', () => {
+		// the converter permits a loop component nothing points into: waiting on its
+		// ledger would hang an execution that is otherwise finished
+		const loops = [loop('B', ['x']), loop('orphan', ['other'])];
+		const reachable = new Set(['trigger', 'B', 'x']);
+
+		expect(countExpectedSettledSteps(loops, reachable, new Map([['B', 1]]))).toBe(1 + 2 + 1);
+	});
+
+	it('waits on every reachable loop, not just the first', () => {
+		const loops = [loop('B1', ['x']), loop('B2', ['y'])];
+		const reachable = new Set(['trigger', 'B1', 'x', 'B2', 'y']);
+
+		expect(countExpectedSettledSteps(loops, reachable, new Map([['B1', 1]]))).toBeUndefined();
+		expect(
+			countExpectedSettledSteps(
+				loops,
+				reachable,
+				new Map([
+					['B1', 1],
+					['B2', 2],
+				]),
+			),
+		).toBe(1 + (2 + 1) + (3 + 2));
+	});
+});

packages/@n8n/engine/src/execution/__tests__/execution-start-handler.test.ts

@@ -31,7 +31,7 @@ function makeStepStore(createSteps = vi.fn()): StepStore {
 		cancelQueuedSteps: vi.fn(),
 		loadStepsByKeys: vi.fn().mockResolvedValue({}),
 		loadStepSummariesByKeys: vi.fn().mockResolvedValue({}),
-		loadLatestStep: vi.fn().mockResolvedValue(null),
+		loadLatestStepSummaries: vi.fn().mockResolvedValue({}),
 		countSettledSteps: vi.fn(),
 		hasFailedSteps: vi.fn(),
 	};

packages/@n8n/engine/src/execution/__tests__/loop-ledger.test.ts

@@ -0,0 +1,121 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import { deriveLoops, type WorkflowGraph } from '../../graph';
+import { endsLoop, exitSourcesInto, isTerminalStep, loadTerminalIterations } from '../loop-ledger';
+import type { StepStore, StepSummary } from '../step-store';
+
+function tip(iteration: number, filledOutputSlots: boolean[], status = 'completed' as const) {
+	return { id: `step-B-${iteration}`, nodeId: 'B', iteration, status, filledOutputSlots };
+}
+
+/** The canonical loop: trigger into B, body x, exit to d. */
+const loopGraph: WorkflowGraph = {
+	nodes: [
+		{ id: 'trigger', name: 'T', type: 'trigger' },
+		{ id: 'B', name: 'B', type: 'batch' },
+		{ id: 'x', name: 'X', type: 'v1-node' },
+		{ id: 'd', name: 'D', type: 'v1-node' },
+	],
+	edges: [
+		{ from: 'trigger', to: 'B', outputIndex: 0, inputIndex: 0 },
+		{ from: 'B', to: 'x', outputIndex: 1, inputIndex: 0 },
+		{ from: 'x', to: 'B', outputIndex: 0, inputIndex: 0, isBackEdge: true },
+		{ from: 'B', to: 'd', outputIndex: 0, inputIndex: 0 },
+	],
+};
+const loops = deriveLoops(loopGraph);
+
+describe('endsLoop', () => {
+	it('ends the loop when a settled row leaves its loop slot dead', () => {
+		expect(endsLoop('completed', false)).toBe(true);
+	});
+
+	it('does not end the loop while the loop slot is filled', () => {
+		expect(endsLoop('completed', true)).toBe(false);
+	});
+
+	it('ends the loop on a skipped row, which fired nothing', () => {
+		expect(endsLoop('skipped', false)).toBe(true);
+	});
+
+	it('does not end the loop from a row that has not settled', () => {
+		expect(endsLoop('running', false)).toBe(false);
+		expect(endsLoop('queued', false)).toBe(false);
+	});
+});
+
+describe('isTerminalStep', () => {
+	it('is terminal when the row fired the done slot', () => {
+		expect(isTerminalStep(tip(2, [true, false]))).toBe(true);
+	});
+
+	it('is not terminal when the row fired the loop slot', () => {
+		expect(isTerminalStep(tip(2, [false, true]))).toBe(false);
+	});
+
+	it('is terminal when the row fired nothing at all', () => {
+		expect(isTerminalStep(tip(2, []))).toBe(true);
+	});
+});
+
+describe('exitSourcesInto', () => {
+	it('names the batch node whose terminal row a node after the loop reads', () => {
+		expect(exitSourcesInto(loopGraph, loops, ['d'])).toEqual(['B']);
+	});
+
+	it('names nothing for a node inside the loop, which reads its own iteration', () => {
+		expect(exitSourcesInto(loopGraph, loops, ['x'])).toEqual([]);
+		expect(exitSourcesInto(loopGraph, loops, ['B'])).toEqual([]);
+	});
+
+	it('names nothing in a graph without loops, so no tip is ever read', () => {
+		expect(exitSourcesInto(loopGraph, [], ['d'])).toEqual([]);
+	});
+});
+
+describe('loadTerminalIterations', () => {
+	function makeStepStore(latest: Record<string, StepSummary>): StepStore {
+		return { loadLatestStepSummaries: vi.fn().mockResolvedValue(latest) } as unknown as StepStore;
+	}
+
+	it('reports the terminal iteration of a loop that has ended', async () => {
+		const stepStore = makeStepStore({ B: tip(3, [true, false]) });
+
+		expect(await loadTerminalIterations(stepStore, 'exec-1', ['B'])).toEqual(new Map([['B', 3]]));
+		expect(stepStore.loadLatestStepSummaries).toHaveBeenCalledExactlyOnceWith('exec-1', ['B']);
+	});
+
+	it('omits a loop still running, so its exit stays undecidable', async () => {
+		const stepStore = makeStepStore({ B: tip(3, [false, true]) });
+
+		expect(await loadTerminalIterations(stepStore, 'exec-1', ['B'])).toEqual(new Map());
+	});
+
+	it('omits a loop with no rows yet', async () => {
+		const stepStore = makeStepStore({});
+
+		expect(await loadTerminalIterations(stepStore, 'exec-1', ['B'])).toEqual(new Map());
+	});
+
+	it('reads several loops in one query, ended and running alike', async () => {
+		const stepStore = makeStepStore({
+			B1: { ...tip(2, [true, false]), nodeId: 'B1' },
+			B2: { ...tip(5, [false, true]), nodeId: 'B2' },
+		});
+
+		expect(await loadTerminalIterations(stepStore, 'exec-1', ['B1', 'B2'])).toEqual(
+			new Map([['B1', 2]]),
+		);
+		expect(stepStore.loadLatestStepSummaries).toHaveBeenCalledExactlyOnceWith('exec-1', [
+			'B1',
+			'B2',
+		]);
+	});
+
+	it('reads nothing when no loop is named', async () => {
+		const stepStore = makeStepStore({});
+
+		expect(await loadTerminalIterations(stepStore, 'exec-1', [])).toEqual(new Map());
+		expect(stepStore.loadLatestStepSummaries).not.toHaveBeenCalled();
+	});
+});

packages/@n8n/engine/src/execution/__tests__/settlement.test.ts

@@ -1,9 +1,9 @@
 import { describe, expect, it } from 'vitest';
 
 import type { GraphEdge, WorkflowGraph } from '../../graph';
-import { getDescendantNodeIds } from '../../graph';
+import { deriveLoops, getDescendantNodeIds } from '../../graph';
 import { stepKeyId, type StepKey, type StepStatus } from '../execution.types';
-import { decideSuccessors } from '../settlement';
+import { decideSuccessors, type SuccessorDecisions } from '../settlement';
 import type { StepSummary } from '../step-store';
 
 function summary(
@@ -18,11 +18,19 @@ function makeSteps(...summaries: StepSummary[]): Record<string, StepSummary> {
 	return Object.fromEntries(summaries.map((s) => [stepKeyId(s), s]));
 }
 
-/** Every row in these tests sits at iteration 0; loops are CAT-2875 part 2. */
 function keyFor(nodeId: string): StepKey {
 	return { nodeId, iteration: 0 };
 }
 
+/** No loops, so there is no loop set and no loop that could have ended. */
+function decideLoopless(
+	graph: WorkflowGraph,
+	settled: StepKey,
+	steps: Record<string, StepSummary>,
+): SuccessorDecisions {
+	return decideSuccessors(graph, [], settled, steps, new Map());
+}
+
 function makeGraph(
 	edges: Array<Partial<GraphEdge> & Pick<GraphEdge, 'from' | 'to'>>,
 ): WorkflowGraph {
@@ -50,7 +58,7 @@ describe('decideSuccessors', () => {
 	it('queues the live branch and skips the dead one', () => {
 		// if fired only output slot 0; m is not a direct successor, so it is
 		// not considered — b's own settled event will examine it.
-		const decisions = decideSuccessors(
+		const decisions = decideLoopless(
 			diamond,
 			keyFor('if'),
 			makeSteps(summary('trigger', 'completed', [true]), summary('if', 'completed', [true, false])),
@@ -62,7 +70,7 @@ describe('decideSuccessors', () => {
 	it('decides a merge once its last predecessor settles', () => {
 		// b was skipped earlier; a just completed. m has one live and one dead
 		// edge, so it runs on the live data.
-		const decisions = decideSuccessors(
+		const decisions = decideLoopless(
 			diamond,
 			keyFor('a'),
 			makeSteps(
@@ -79,7 +87,7 @@ describe('decideSuccessors', () => {
 	it('leaves a merge undecided while a predecessor is still unsettled', () => {
 		// b's settled event fires while a is still queued: m is not decidable
 		// yet, and a's own settlement will decide it later.
-		const decisions = decideSuccessors(
+		const decisions = decideLoopless(
 			diamond,
 			keyFor('b'),
 			makeSteps(
@@ -107,19 +115,19 @@ describe('decideSuccessors', () => {
 			summary('if', 'completed', [true, false]),
 		);
 
-		expect(decideSuccessors(graph, keyFor('if'), steps)).toEqual({
+		expect(decideLoopless(graph, keyFor('if'), steps)).toEqual({
 			toQueue: [keyFor('a')],
 			toSkip: [keyFor('b')],
 		});
 
 		steps[stepKeyId(keyFor('b'))] = summary('b', 'skipped');
-		expect(decideSuccessors(graph, keyFor('b'), steps)).toEqual({
+		expect(decideLoopless(graph, keyFor('b'), steps)).toEqual({
 			toQueue: [],
 			toSkip: [keyFor('c')],
 		});
 
 		steps[stepKeyId(keyFor('c'))] = summary('c', 'skipped');
-		expect(decideSuccessors(graph, keyFor('c'), steps)).toEqual({
+		expect(decideLoopless(graph, keyFor('c'), steps)).toEqual({
 			toQueue: [],
 			toSkip: [keyFor('d')],
 		});
@@ -142,10 +150,10 @@ describe('decideSuccessors', () => {
 			summary('b', 'skipped'),
 		);
 
-		expect(decideSuccessors(graph, keyFor('b'), steps)).toEqual({ toQueue: [], toSkip: [] });
+		expect(decideLoopless(graph, keyFor('b'), steps)).toEqual({ toQueue: [], toSkip: [] });
 
 		steps[stepKeyId(keyFor('c'))] = summary('c', 'skipped');
-		expect(decideSuccessors(graph, keyFor('c'), steps)).toEqual({
+		expect(decideLoopless(graph, keyFor('c'), steps)).toEqual({
 			toQueue: [],
 			toSkip: [keyFor('d')],
 		});
@@ -155,7 +163,7 @@ describe('decideSuccessors', () => {
 		// duplicate delivery: both successors were planned by the first run.
 		// Nothing is created, so nothing is announced — a lost announcement is
 		// reconciliation's job (CAT-2938), not this planner's.
-		const decisions = decideSuccessors(
+		const decisions = decideLoopless(
 			diamond,
 			keyFor('if'),
 			makeSteps(
@@ -177,7 +185,7 @@ describe('decideSuccessors', () => {
 			{ from: 'switch', to: 'c', outputIndex: 2 },
 		]);
 
-		const decisions = decideSuccessors(
+		const decisions = decideLoopless(
 			graph,
 			keyFor('switch'),
 			makeSteps(
@@ -195,7 +203,7 @@ describe('decideSuccessors', () => {
 			{ from: 'a', to: 'b', outputIndex: 1 },
 		]);
 
-		const decisions = decideSuccessors(
+		const decisions = decideLoopless(
 			graph,
 			keyFor('a'),
 			makeSteps(summary('trigger', 'completed', [true]), summary('a', 'completed', [true])),
@@ -274,7 +282,7 @@ function simulateExecution(
 		if (handleEvent) {
 			const nodeId = settledEvents[0];
 			settledEvents.shift();
-			const { toQueue, toSkip } = decideSuccessors(graph, keyFor(nodeId), steps);
+			const { toQueue, toSkip } = decideLoopless(graph, keyFor(nodeId), steps);
 			for (const { nodeId: id } of toQueue) {
 				steps[stepKeyId(keyFor(id))] = summary(id, 'queued');
 				queuedNodes.push(id);
@@ -336,3 +344,170 @@ describe('the event loop over decideSuccessors matches the reference evaluator',
 		}
 	});
 });
+
+describe('decideSuccessors over loop iterations', () => {
+	/**
+	 * ┌───────┐    ┌───┐ o0    ┌───┐
+	 * │trigger├───►│   ├──────►│ d │
+	 * └───────┘    │ B │       └───┘
+	 *              │   │ o1    ┌───┐
+	 *              │   ├──────►│ x │
+	 *              └─▲─┘       └─┬─┘
+	 *                └──(back)───┘
+	 */
+	const graph = makeGraph([
+		{ from: 'trigger', to: 'B' },
+		{ from: 'B', to: 'x', outputIndex: 1 },
+		{ from: 'x', to: 'B', isBackEdge: true },
+		{ from: 'B', to: 'd', outputIndex: 0 },
+	]);
+	graph.nodes = graph.nodes.map((node) =>
+		node.id === 'B' ? { ...node, type: 'batch' as const } : node,
+	);
+	const loops = deriveLoops(graph);
+
+	/** A row at a given iteration, where the loopless `summary` always builds 0. */
+	function at(
+		nodeId: string,
+		iteration: number,
+		status: StepStatus,
+		filledOutputSlots: boolean[] = [],
+	): StepSummary {
+		return { id: `step-${nodeId}-${iteration}`, nodeId, iteration, status, filledOutputSlots };
+	}
+
+	function decide(
+		settled: StepKey,
+		steps: Record<string, StepSummary>,
+		terminalIterations: Map<string, number> = new Map(),
+	): SuccessorDecisions {
+		return decideSuccessors(graph, loops, settled, steps, terminalIterations);
+	}
+
+	const key = (nodeId: string, iteration: number): StepKey => ({ nodeId, iteration });
+
+	it('queues the body at the batch row iteration, and leaves the exit undecided', () => {
+		// B filled its loop slot, so the loop runs on and its end is unknown
+		const steps = makeSteps(at('B', 0, 'completed', [false, true]));
+
+		expect(decide(key('B', 0), steps)).toEqual({ toQueue: [key('x', 0)], toSkip: [] });
+	});
+
+	it('advances the iteration across the return edge', () => {
+		const steps = makeSteps(
+			at('B', 0, 'completed', [false, true]),
+			at('x', 0, 'completed', [true]),
+		);
+
+		expect(decide(key('x', 0), steps)).toEqual({ toQueue: [key('B', 1)], toSkip: [] });
+	});
+
+	it('skips the next iteration when the body returns nothing', () => {
+		// a dead return edge ends the loop: (B, 1) is skipped at birth, and skipped
+		// rows fire nothing, which makes it the terminal row
+		const steps = makeSteps(
+			at('B', 0, 'completed', [false, true]),
+			at('x', 0, 'completed', [false]),
+		);
+
+		expect(decide(key('x', 0), steps)).toEqual({ toQueue: [], toSkip: [key('B', 1)] });
+	});
+
+	it('queues what follows the loop from the terminal row only', () => {
+		// B filled its done slot instead, so the loop has ended at iteration 2
+		const steps = makeSteps(at('B', 2, 'completed', [true, false]));
+		const terminals = new Map([['B', 2]]);
+
+		expect(decide(key('B', 2), steps, terminals)).toEqual({
+			toQueue: [key('d', 0)],
+			toSkip: [],
+		});
+	});
+
+	it('plans no body row at the terminal iteration', () => {
+		// the cascade this prevents: a skipped body row would settle, its return edge
+		// would plan another batch row, and that row would end the loop in turn
+		const steps = makeSteps(at('B', 2, 'completed', [true, false]));
+		const terminals = new Map([['B', 2]]);
+
+		const decisions = decide(key('B', 2), steps, terminals);
+
+		expect(decisions.toQueue).not.toContainEqual(key('x', 2));
+		expect(decisions.toSkip).not.toContainEqual(key('x', 2));
+	});
+
+	it('leaves what follows the loop undecided while the loop still runs', () => {
+		// B@1 has a dead done slot, and d is not skipped on the strength of it: an
+		// empty toSkip is the assertion, since a later row may still fire that slot
+		const steps = makeSteps(
+			at('B', 0, 'completed', [false, true]),
+			at('x', 0, 'completed', [true]),
+			at('B', 1, 'completed', [false, true]),
+		);
+
+		expect(decide(key('B', 1), steps)).toEqual({ toQueue: [key('x', 1)], toSkip: [] });
+	});
+
+	/**
+	 * The rule above keeps a running loop from deciding its exit, but a node after
+	 * the loop can also be reached from outside it. Then the exit edge is resolved
+	 * for real, finds no terminal row, and holds the decision open.
+	 *
+	 * ┌───────┐    ┌───┐ o1    ┌───┐
+	 * │trigger├───►│ B ├──────►│ x │
+	 * └───┬───┘    └─▲─┘       └─┬─┘
+	 *     │          └──(back)───┘
+	 *     │       ┌───┐ o0
+	 *     └──────►│ p ├──────────► d ◄── B's done slot
+	 *             └───┘
+	 */
+	it('holds a node after the loop undecided when another predecessor settles first', () => {
+		const joined = makeGraph([
+			{ from: 'trigger', to: 'B' },
+			{ from: 'B', to: 'x', outputIndex: 1 },
+			{ from: 'x', to: 'B', isBackEdge: true },
+			{ from: 'B', to: 'd', outputIndex: 0 },
+			{ from: 'trigger', to: 'p', outputIndex: 1 },
+			{ from: 'p', to: 'd', inputIndex: 1 },
+		]);
+		joined.nodes = joined.nodes.map((node) =>
+			node.id === 'B' ? { ...node, type: 'batch' as const } : node,
+		);
+		const joinedLoops = deriveLoops(joined);
+		const steps = makeSteps(
+			at('B', 0, 'completed', [false, true]),
+			at('p', 0, 'completed', [true]),
+		);
+
+		// p settling makes d a candidate, but the loop has not ended, so the exit
+		// edge has no row to read and d gets no fate at all
+		expect(decideSuccessors(joined, joinedLoops, key('p', 0), steps, new Map())).toEqual({
+			toQueue: [],
+			toSkip: [],
+		});
+
+		// once it has ended, the same settlement queues d
+		const ended = makeSteps(
+			at('B', 2, 'completed', [true, false]),
+			at('p', 0, 'completed', [true]),
+		);
+		expect(decideSuccessors(joined, joinedLoops, key('p', 0), ended, new Map([['B', 2]]))).toEqual({
+			toQueue: [key('d', 0)],
+			toSkip: [],
+		});
+	});
+
+	it('reads the entry edge at iteration 0 and the return edge after it', () => {
+		const entry = makeSteps(at('trigger', 0, 'completed', [true]));
+		expect(decide(key('trigger', 0), entry)).toEqual({ toQueue: [key('B', 0)], toSkip: [] });
+
+		// at iteration 1 the entry edge connects nothing, so the return edge alone
+		// decides, and B is queued on its strength
+		const second = makeSteps(
+			at('trigger', 0, 'completed', [true]),
+			at('B', 0, 'completed', [false, true]),
+			at('x', 0, 'completed', [true]),
+		);
+		expect(decide(key('x', 0), second)).toEqual({ toQueue: [key('B', 1)], toSkip: [] });
+	});
+});

packages/@n8n/engine/src/execution/__tests__/step-ready-handler.test.ts

@@ -1,12 +1,12 @@
 import { describe, expect, it, vi } from 'vitest';
 
 import type { ExternalDependencies, IStepExecutor } from '../../dependencies';
-import type { WorkflowGraph } from '../../graph';
+import { deriveLoops, type WorkflowGraph } from '../../graph';
 import type { OrchestrationMessage, WorkQueue } from '../../queue';
 import type { ExecutionRecord, ExecutionStore } from '../execution-store';
 import { stepKeyId, type StepSlots, type StepStatus } from '../execution.types';
-import { StepReadyHandler } from '../step-ready-handler';
-import type { StepRecord, StepStore } from '../step-store';
+import { resolveInputReads, StepReadyHandler } from '../step-ready-handler';
+import type { StepRecord, StepStore, StepSummary } from '../step-store';
 
 /** Key for a `loadStepsByKeys` result at iteration 0, as the handler requests them. */
 const at = (nodeId: string) => stepKeyId({ nodeId, iteration: 0 });
@@ -77,7 +77,7 @@ function makeStepStore(step: Partial<StepRecord> = {}, overrides: Partial<StepSt
 			.fn()
 			.mockResolvedValue({ [at('trigger')]: stepRow('trigger', 'completed', [{}]) }),
 		loadStepSummariesByKeys: vi.fn().mockResolvedValue({}),
-		loadLatestStep: vi.fn().mockResolvedValue(null),
+		loadLatestStepSummaries: vi.fn().mockResolvedValue({}),
 		countSettledSteps: vi.fn().mockResolvedValue(0),
 		hasFailedSteps: vi.fn().mockResolvedValue(false),
 		...overrides,
@@ -684,3 +684,152 @@ describe('StepReadyHandler', () => {
 		},
 	);
 });
+
+describe('StepReadyHandler over loop iterations', () => {
+	/**
+	 * ┌───────┐    ┌───┐ o0    ┌───┐
+	 * │trigger├───►│   ├──────►│ d │
+	 * └───────┘    │ B │       └───┘
+	 *              │   │ o1    ┌───┐
+	 *              │   ├──────►│ x │
+	 *              └─▲─┘       └─┬─┘
+	 *                └──(back)───┘
+	 */
+	const loopGraph: WorkflowGraph = {
+		nodes: [
+			{ id: 'trigger', name: 'T', type: 'trigger' },
+			{ id: 'B', name: 'B', type: 'batch' },
+			{ id: 'x', name: 'X', type: 'v1-node' },
+			{ id: 'd', name: 'D', type: 'v1-node' },
+		],
+		edges: [
+			{ from: 'trigger', to: 'B', outputIndex: 0, inputIndex: 0 },
+			{ from: 'B', to: 'x', outputIndex: 1, inputIndex: 0 },
+			{ from: 'x', to: 'B', outputIndex: 0, inputIndex: 0, isBackEdge: true },
+			{ from: 'B', to: 'd', outputIndex: 0, inputIndex: 0 },
+		],
+	};
+
+	/** A loop's latest row as the store returns it: filled slots, no payloads. */
+	function tipAt(iteration: number, filledOutputSlots: boolean[]): StepSummary {
+		return {
+			id: `step-B-${iteration}`,
+			nodeId: 'B',
+			iteration,
+			status: 'completed',
+			filledOutputSlots,
+		};
+	}
+
+	function rowAt(nodeId: string, iteration: number, outputs: StepRecord['outputs']): StepRecord {
+		return {
+			id: `step-${nodeId}-${iteration}`,
+			executionId: 'exec-1',
+			nodeId,
+			iteration,
+			status: 'completed',
+			outputs,
+			error: null,
+		};
+	}
+
+	it('reads the body input from the batch row at the same iteration', async () => {
+		const executor = makeExecutor();
+		const stepStore = makeStepStore(
+			{ id: 'step-x-2', nodeId: 'x', iteration: 2 },
+			{
+				loadStepsByKeys: vi.fn().mockResolvedValue({
+					[stepKeyId({ nodeId: 'B', iteration: 2 })]: rowAt('B', 2, [null, [{ json: { i: 2 } }]]),
+				}),
+			},
+		);
+		const handler = new StepReadyHandler(
+			makeExecutionStore({ graph: loopGraph }),
+			stepStore,
+			makeQueue(),
+			{ v1StepExecutor: executor },
+		);
+
+		await handler.handle({ ...event, stepId: 'step-x-2' });
+
+		expect(stepStore.loadStepsByKeys).toHaveBeenCalledWith('exec-1', [
+			{ nodeId: 'B', iteration: 2 },
+		]);
+		expect(executor.execute).toHaveBeenCalledWith(
+			expect.objectContaining({ inputs: [[{ json: { i: 2 } }]] }),
+		);
+	});
+
+	/**
+	 * A batch node's slot 0 carries both its entry edge and its return edge, which
+	 * the old check rejected outright as two edges into one slot. Asserted on the
+	 * resolution rather than through the handler, since running a batch step needs
+	 * an executor that does not exist yet.
+	 */
+	it('reads the batch node from the entry edge at iteration 0 and the return edge after it', () => {
+		const loops = deriveLoops(loopGraph);
+		const intoB = loopGraph.edges.filter((edge) => edge.to === 'B');
+
+		const readsAt = (iteration: number) =>
+			resolveInputReads(
+				intoB,
+				loops,
+				{ id: `step-B-${iteration}`, nodeId: 'B', iteration },
+				new Map(),
+			).map(({ edge, key }) => ({ from: edge.from, key }));
+
+		expect(readsAt(0)).toEqual([{ from: 'trigger', key: { nodeId: 'trigger', iteration: 0 } }]);
+		expect(readsAt(1)).toEqual([{ from: 'x', key: { nodeId: 'x', iteration: 0 } }]);
+	});
+
+	it('reads what follows the loop from the terminal row, whatever iteration that is', async () => {
+		const executor = makeExecutor();
+		const stepStore = makeStepStore(
+			{ id: 'step-d-0', nodeId: 'd', iteration: 0 },
+			{
+				loadLatestStepSummaries: vi.fn().mockResolvedValue({ B: tipAt(4, [true, false]) }),
+				loadStepsByKeys: vi.fn().mockResolvedValue({
+					[stepKeyId({ nodeId: 'B', iteration: 4 })]: rowAt('B', 4, [
+						[{ json: { done: true } }],
+						null,
+					]),
+				}),
+			},
+		);
+		const handler = new StepReadyHandler(
+			makeExecutionStore({ graph: loopGraph }),
+			stepStore,
+			makeQueue(),
+			{ v1StepExecutor: executor },
+		);
+
+		await handler.handle({ ...event, stepId: 'step-d-0' });
+
+		expect(stepStore.loadStepsByKeys).toHaveBeenCalledWith('exec-1', [
+			{ nodeId: 'B', iteration: 4 },
+		]);
+		expect(executor.execute).toHaveBeenCalledWith(
+			expect.objectContaining({ inputs: [[{ json: { done: true } }]] }),
+		);
+	});
+
+	it('throws, running nothing, when the loop it reads across has not ended', async () => {
+		// only the planner should queue such a step, so the rows and the plan disagree
+		const executor = makeExecutor();
+		const stepStore = makeStepStore(
+			{ id: 'step-d-0', nodeId: 'd', iteration: 0 },
+			{ loadLatestStepSummaries: vi.fn().mockResolvedValue({ B: tipAt(4, [false, true]) }) },
+		);
+		const handler = new StepReadyHandler(
+			makeExecutionStore({ graph: loopGraph }),
+			stepStore,
+			makeQueue(),
+			{ v1StepExecutor: executor },
+		);
+
+		await expect(handler.handle({ ...event, stepId: 'step-d-0' })).rejects.toThrow(
+			/across a loop that has not ended/,
+		);
+		expect(executor.execute).not.toHaveBeenCalled();
+	});
+});

packages/@n8n/engine/src/execution/__tests__/step-settled-handler.test.ts

@@ -103,7 +103,7 @@ function makeStepStore(
 			);
 		}),
 		loadStepsByKeys: vi.fn().mockResolvedValue({}),
-		loadLatestStep: vi.fn().mockResolvedValue(null),
+		loadLatestStepSummaries: vi.fn().mockResolvedValue({}),
 		// far from settled, so finish tests opt in explicitly
 		countSettledSteps: vi.fn().mockResolvedValue(0),
 		hasFailedSteps: vi.fn().mockResolvedValue(false),
@@ -201,7 +201,7 @@ describe('StepSettledHandler', () => {
 		expect(orchestrationQueue.publish).not.toHaveBeenCalled();
 	});
 
-	it('loads the decision rows in one query: the successors and their predecessors', async () => {
+	it('loads the decision rows in one query: the settled row, the successors, and what they read', async () => {
 		const stepStore = makeStepStore({ id: 'step-b', nodeId: 'b' }, {}, [
 			...defaultSummaries,
 			summary('b', 'completed', [true]),
@@ -210,10 +210,10 @@ describe('StepSettledHandler', () => {
 
 		await handler.handle({ ...event, stepId: 'step-b' });
 
-		// b's only successor is m; m reads from both b and c
+		// b's only successor is m, and m reads from both b and c
 		expect(stepStore.loadStepSummariesByKeys).toHaveBeenCalledExactlyOnceWith('exec-1', [
-			{ nodeId: 'm', iteration: 0 },
 			{ nodeId: 'b', iteration: 0 },
+			{ nodeId: 'm', iteration: 0 },
 			{ nodeId: 'c', iteration: 0 },
 		]);
 	});

packages/@n8n/engine/src/execution/completion.ts

@@ -0,0 +1,42 @@
+import type { WorkflowLoop } from '../graph';
+
+/**
+ * How many settled steps a finished execution owes, or `undefined` while a loop
+ * the trigger reaches is still running.
+ *
+ * Without loops this is the reachable node count, one step per node. A loop
+ * replaces its members' single steps with one per iteration.
+ *
+ * While a loop runs, how many steps it will create is not known, so there is no
+ * total to give. That matters between two iterations, when every step so far has
+ * settled: with a total, the execution would look finished when the next
+ * iteration simply has not been planned yet.
+ *
+ * Only loops the trigger reaches count. The converter allows a loop component
+ * nothing points into, and those never receive steps, so waiting on one would
+ * hang the execution.
+ */
+export function countExpectedSettledSteps(
+	loops: WorkflowLoop[],
+	reachable: Set<string>,
+	terminalIterations: Map<string, number>,
+): number | undefined {
+	let expected = 0;
+	const members = new Set<string>();
+
+	for (const loop of loops.filter((l) => reachable.has(l.batchNodeId))) {
+		const terminal = terminalIterations.get(loop.batchNodeId);
+		if (terminal === undefined) return undefined;
+
+		// iterations 0 to t for the batch node, then t each for the rest: the body
+		// has no step at the terminal iteration
+		expected += terminal + 1 + terminal * (loop.memberIds.size - 1);
+		for (const memberId of loop.memberIds) members.add(memberId);
+	}
+
+	for (const nodeId of reachable) {
+		if (!members.has(nodeId)) expected += 1;
+	}
+
+	return expected;
+}

packages/@n8n/engine/src/execution/loop-ledger.ts

@@ -0,0 +1,72 @@
+import type { WorkflowGraph, WorkflowLoop } from '../graph';
+import { isSettledStatus, type StepStatus } from './execution.types';
+import { classifyEdge } from './iteration-mapping';
+import type { StepStore, StepSummary } from './step-store';
+
+/** A batch node's output slots: 0 is done, 1 is loop. */
+const LOOP_SLOT = 1;
+
+/**
+ * A loop's batch node steps form a ledger: one per iteration, written strictly
+ * in order, since iteration `i + 1` is planned only once iteration `i` has
+ * settled. The last one is the terminal step, and it is what says the loop is
+ * over.
+ *
+ * A step ends the loop when it settles without filling its loop slot: it fired
+ * the done slot instead, or it never ran at all, as a skip records. Nothing can
+ * advance the loop past it, so it is always the last one.
+ */
+export function endsLoop(status: StepStatus, loopSlotFilled: boolean): boolean {
+	return isSettledStatus(status) && !loopSlotFilled;
+}
+
+/** `endsLoop` for a step, which carries both of the facts it asks for. */
+export function isTerminalStep(step: StepSummary): boolean {
+	return endsLoop(step.status, Boolean(step.filledOutputSlots[LOOP_SLOT]));
+}
+
+/**
+ * The batch nodes whose loop-ending step has to be read to resolve the edges
+ * into `targetNodeIds`. Only an exit edge reads one, so a target with no exit
+ * edge into it needs none: every other edge class resolves against the target's
+ * own iteration.
+ */
+export function exitSourcesInto(
+	graph: WorkflowGraph,
+	loops: WorkflowLoop[],
+	targetNodeIds: string[],
+): string[] {
+	if (loops.length === 0) return [];
+
+	const targets = new Set(targetNodeIds);
+	const sources = new Set<string>();
+	for (const edge of graph.edges) {
+		if (targets.has(edge.to) && classifyEdge(edge, loops) === 'exit') sources.add(edge.from);
+	}
+	return [...sources];
+}
+
+/**
+ * The terminal iteration of each loop asked about, by batch node id, omitting
+ * the loops that have not ended.
+ *
+ * Only the latest step can be terminal, since they are written in order, so one
+ * query over every loop answers this. It reads the slim view because the step
+ * ending a loop holds everything that loop accumulated, and this runs on every
+ * settlement.
+ */
+export async function loadTerminalIterations(
+	stepStore: StepStore,
+	executionId: string,
+	batchNodeIds: string[],
+): Promise<Map<string, number>> {
+	if (batchNodeIds.length === 0) return new Map();
+
+	const latest = await stepStore.loadLatestStepSummaries(executionId, batchNodeIds);
+
+	const terminalIterations = new Map<string, number>();
+	for (const [batchNodeId, step] of Object.entries(latest)) {
+		if (isTerminalStep(step)) terminalIterations.set(batchNodeId, step.iteration);
+	}
+	return terminalIterations;
+}

packages/@n8n/engine/src/execution/settlement.ts

@@ -1,90 +1,214 @@
-import { getSuccessorNodeIds, type GraphEdge, type WorkflowGraph } from '../graph';
+import type { GraphEdge, WorkflowGraph, WorkflowLoop } from '../graph';
 import { stepKeyId, isSettledStatus, type StepKey, type StepKeyId } from './execution.types';
+import { classifyEdge, sourceRow, targetKey, type EdgeClass } from './iteration-mapping';
+import { isTerminalStep } from './loop-ledger';
 import type { StepSummary } from './step-store';
 
 /**
+ * A **step** is one run of one node, identified by `StepKey`, which is
+ * `(nodeId, iteration)`. Outside a loop a node has a single step, at iteration
+ * 0. A loop member has one step per pass.
+ *
  * Settlement rules (design: CAT-2874):
  *
- * 1. Every node the execution considers eventually settles: completed,
- *    failed, skipped, or cancelled. Settled fates are immutable. The
- *    execution is finished exactly when every reachable node has settled.
+ * 1. Every step the execution creates eventually settles: completed, failed,
+ *    skipped, or cancelled. Settled fates are immutable. The execution is
+ *    finished exactly when every step it owes has settled, which
+ *    `completion.ts` counts.
  * 2. An edge is live iff its source completed and filled the edge's output
  *    slot; it is dead if the source settled any other way, or left the slot
  *    null. An edge whose source is unsettled is neither yet.
- * 3. A node is decidable once every predecessor is settled. Decidable with at
- *    least one live incoming edge -> it runs (queued). Decidable with none ->
- *    it is skipped: settled at birth, its own out-edges all dead.
+ * 3. A step is decidable once every step its incoming edges read has settled.
+ *    Decidable with at least one live incoming edge -> it runs (queued).
+ *    Decidable with none -> it is skipped: settled at birth, its own out-edges
+ *    all dead.
  * 4. A settlement decides only its direct successors; the cascade is the
  *    event loop. A skip is announced as settled like any other settlement,
  *    and its handler decides the next hop.
  * 5. A planner commits its decisions in one batch, then announces only the
- *    rows it created. A settlement whose announcement is lost is detectable
- *    from rows alone — a settled row with decidable but rowless successors —
- *    and re-announced by reconciliation (CAT-2938).
+ *    steps it created. A settlement whose announcement is lost is detectable
+ *    from the steps alone — a settled one whose successors are decidable but
+ *    absent — and re-announced by reconciliation (CAT-2938).
  *
- * Fates are pure functions of settled rows, so any planner, at any time,
+ * Fates are pure functions of settled steps, so any planner, at any time,
  * recomputes the same decisions; duplicates and races converge instead of
  * corrupting.
  *
- * Rows are identified per `(nodeId, iteration)`: loop members run once per
- * pass (CAT-2875). Forward edges stay within a pass; the back-edge and exit
- * mappings land with loop execution, and until then back-edges are rejected.
+ * Loops extend those rules rather than replacing them (CAT-2875).
+ * `iteration-mapping.ts` decides which iterations an edge connects, and two
+ * consequences belong to loops alone:
+ *
+ * - A batch node's step decides one side of its loop. While the loop runs it
+ *   decides the body, and on the terminal step, the one whose loop slot stayed
+ *   dead, it decides the nodes after the loop instead. Deciding the other side
+ *   too early would skip it, and a later step could not take that back.
+ * - Body steps exist for running iterations only. The terminal iteration has
+ *   none at all, not even skipped ones, or those skips would cascade through
+ *   the body into a further iteration, and on forever. The exclusion lives in
+ *   `decideNodeFate`, so anything recomputing fates reaches the same answer,
+ *   reconciliation included.
  */
 
 export interface SuccessorDecisions {
-	/** Successors with a live input, to enqueue — in edge order. */
+	/** Successor steps with a live input, to enqueue — in edge order. */
 	toQueue: StepKey[];
-	/** Successors with settled but all-dead inputs, to record as skipped. */
+	/** Successor steps with settled but all-dead inputs, to record as skipped. */
 	toSkip: StepKey[];
 }
 
 /**
- * Decides the direct successors of the settled step (rules 2–4). `steps`
- * holds the existing rows for those successors and their predecessors, keyed
- * by `stepKeyId`, including the settled step itself.
+ * Decides the direct successors of the settled step (rules 2–4).
+ *
+ * `steps` holds the steps `decisionKeys` names, keyed by `stepKeyId`. Anything
+ * less and a decision reads an absent step as one that has not settled, leaving
+ * the successor undecided forever. `terminalIterations` holds each loop's
+ * terminal iteration by batch node id, omitting the loops that have not ended.
  */
 export function decideSuccessors(
 	graph: WorkflowGraph,
+	loops: WorkflowLoop[],
 	settled: StepKey,
 	steps: Record<StepKeyId, StepSummary>,
+	terminalIterations: Map<string, number>,
 ): SuccessorDecisions {
 	const decisions: SuccessorDecisions = { toQueue: [], toSkip: [] };
-	for (const successorNodeId of getSuccessorNodeIds(graph, settled.nodeId)) {
-		const successor: StepKey = { nodeId: successorNodeId, iteration: settled.iteration };
-		// An existing row was decided by an earlier settlement, which announced it.
-		if (steps[stepKeyId(successor)]) continue;
-		const fate = decideNodeFate(graph, successor, steps);
-		if (fate === 'queued') decisions.toQueue.push(successor);
-		else if (fate === 'skipped') decisions.toSkip.push(successor);
+	const batchStep = loops.some((loop) => loop.batchNodeId === settled.nodeId)
+		? steps[stepKeyId(settled)]
+		: undefined;
+	const decided = new Set<StepKeyId>();
+	const outgoingEdges = graph.edges.filter((edge) => edge.from === settled.nodeId);
+
+	for (const edge of outgoingEdges) {
+		const edgeClass = classifyEdge(edge, loops);
+		if (batchStep && !batchStepDecides(edgeClass, batchStep)) continue;
+
+		const target = targetKey(edge, edgeClass, settled);
+		const targetId = stepKeyId(target);
+		// An existing step was decided by an earlier settlement, which announced it.
+		// Two edges into one step are one candidate, decided once.
+		if (steps[targetId] || decided.has(targetId)) continue;
+		decided.add(targetId);
+
+		const fate = decideNodeFate(graph, loops, target, steps, terminalIterations);
+		if (fate === 'queued') decisions.toQueue.push(target);
+		else if (fate === 'skipped') decisions.toSkip.push(target);
 	}
+
 	return decisions;
 }
 
-/** One candidate's fate under rules 2–3; undecidable while a predecessor is unsettled. */
+/** Which side of its loop a batch step decides: the body, or what follows. */
+function batchStepDecides(edgeClass: EdgeClass, batchStep: StepSummary): boolean {
+	return isTerminalStep(batchStep) ? edgeClass === 'exit' : edgeClass !== 'exit';
+}
+
+/**
+ * One candidate's fate under rules 2–3.
+ *
+ * `undecidable` covers a predecessor that has not settled and a loop that has
+ * not ended alike, since either way the step an edge reads does not exist yet.
+ * `outside` means the candidate is not part of the execution at this iteration
+ * and gets no step at all, which is what keeps a loop that has ended from
+ * cascading skips through its own body.
+ */
 function decideNodeFate(
 	graph: WorkflowGraph,
+	loops: WorkflowLoop[],
 	candidate: StepKey,
 	steps: Record<StepKeyId, StepSummary>,
-): 'queued' | 'skipped' | 'undecidable' {
-	// Back-edges aside: loop iteration is CAT-2875.
-	const incoming = graph.edges.filter((edge) => edge.to === candidate.nodeId && !edge.isBackEdge);
-	const predecessors = [...new Set(incoming.map((edge) => edge.from))];
-	const settled = predecessors.every((nodeId) => {
-		const row = steps[stepKeyId({ nodeId, iteration: candidate.iteration })];
-		return row !== undefined && isSettledStatus(row.status);
-	});
-	if (!settled) return 'undecidable';
-	return incoming.some((edge) => isLiveEdge(edge, candidate.iteration, steps))
-		? 'queued'
-		: 'skipped';
+	terminalIterations: Map<string, number>,
+): 'queued' | 'skipped' | 'undecidable' | 'outside' {
+	if (isPastLoopEnd(loops, candidate, steps)) return 'outside';
+
+	let applicable = 0;
+	let live = false;
+	const incomingEdges = graph.edges.filter((edge) => edge.to === candidate.nodeId);
+
+	for (const edge of incomingEdges) {
+		const source = sourceRow(
+			edge,
+			classifyEdge(edge, loops),
+			candidate,
+			terminalIterations.get(edge.from),
+		);
+		if (source.kind === 'none') continue;
+		if (source.kind === 'pending') return 'undecidable';
+
+		applicable += 1;
+		const sourceStep = steps[stepKeyId(source.key)];
+		if (!sourceStep || !isSettledStatus(sourceStep.status)) return 'undecidable';
+		if (isLive(sourceStep, edge)) live = true;
+	}
+
+	// No edge connects to this step at this iteration, so nothing produces it.
+	if (applicable === 0) return 'outside';
+
+	return live ? 'queued' : 'skipped';
+}
+
+/**
+ * Whether the candidate is a body step of a loop that has already ended. Read
+ * from the batch step rather than from `terminalIterations`, so the exclusion
+ * holds for anything recomputing fates from the steps alone.
+ */
+function isPastLoopEnd(
+	loops: WorkflowLoop[],
+	candidate: StepKey,
+	steps: Record<StepKeyId, StepSummary>,
+): boolean {
+	const loop = loops.find((l) => l.memberIds.has(candidate.nodeId));
+	if (!loop || candidate.nodeId === loop.batchNodeId) return false;
+
+	const batchStep = steps[stepKeyId({ nodeId: loop.batchNodeId, iteration: candidate.iteration })];
+	return batchStep !== undefined && isTerminalStep(batchStep);
 }
 
 /** Rule 2. A slot beyond the produced list reads undefined — dead, like null. */
-function isLiveEdge(
-	edge: GraphEdge,
-	iteration: number,
-	steps: Record<StepKeyId, StepSummary>,
-): boolean {
-	const source = steps[stepKeyId({ nodeId: edge.from, iteration })];
-	return source?.status === 'completed' && Boolean(source.filledOutputSlots[edge.outputIndex]);
+function isLive(source: StepSummary, edge: GraphEdge): boolean {
+	return source.status === 'completed' && Boolean(source.filledOutputSlots[edge.outputIndex]);
+}
+
+/**
+ * The steps `decideSuccessors` reads for this settlement:
+ *
+ * - the settled step, which for a batch node says which side to decide
+ * - each candidate successor, since an existing one means it was decided already
+ * - the steps each candidate's own incoming edges read
+ * - for a candidate inside a loop, that loop's batch step at the same iteration
+ *
+ * Derived from the same mapping the decision uses, so a caller cannot load a
+ * different set from the one that gets read.
+ */
+export function decisionKeys(
+	graph: WorkflowGraph,
+	loops: WorkflowLoop[],
+	settled: StepKey,
+	terminalIterations: Map<string, number>,
+): StepKey[] {
+	const keys = new Map<StepKeyId, StepKey>([[stepKeyId(settled), settled]]);
+	const add = (key: StepKey) => keys.set(stepKeyId(key), key);
+
+	const outgoingEdges = graph.edges.filter((edge) => edge.from === settled.nodeId);
+
+	for (const edge of outgoingEdges) {
+		const target = targetKey(edge, classifyEdge(edge, loops), settled);
+		add(target);
+
+		const incomingEdges = graph.edges.filter((inEdge) => inEdge.to === target.nodeId);
+
+		for (const inEdge of incomingEdges) {
+			const source = sourceRow(
+				inEdge,
+				classifyEdge(inEdge, loops),
+				target,
+				terminalIterations.get(inEdge.from),
+			);
+			if (source.kind === 'row') add(source.key);
+		}
+
+		const loop = loops.find((l) => l.memberIds.has(target.nodeId));
+		if (loop) add({ nodeId: loop.batchNodeId, iteration: target.iteration });
+	}
+
+	return [...keys.values()];
 }

packages/@n8n/engine/src/execution/step-ready-handler.ts

@@ -1,9 +1,17 @@
 import { UnexpectedError, UnimplementedError, type JsonValue } from '../common';
 import type { ExternalDependencies, IStepExecutor } from '../dependencies';
-import type { GraphEdge, GraphNode } from '../graph';
+import { deriveLoops, type GraphEdge, type GraphNode, type WorkflowLoop } from '../graph';
 import type { OrchestrationMessage, StepReadyEvent, WorkQueue } from '../queue';
 import type { ExecutionRecord, ExecutionStore } from './execution-store';
-import { stepKeyId, isSettledStatus, type StepKeyId, type StepSlots } from './execution.types';
+import {
+	stepKeyId,
+	isSettledStatus,
+	type StepKey,
+	type StepKeyId,
+	type StepSlots,
+} from './execution.types';
+import { classifyEdge, sourceRow } from './iteration-mapping';
+import { exitSourcesInto, loadTerminalIterations } from './loop-ledger';
 import type { StepError, StepRecord, StepStore } from './step-store';
 import { validateStepContext } from './validate-step-context';
 
@@ -102,7 +110,7 @@ export class StepReadyHandler {
 		return outputs;
 	}
 
-	/** Inputs for `node`, each slot taken from its predecessor's output. */
+	/** Inputs for `node`, each slot taken from the row its edge reads. */
 	private async gatherInputs(execution: ExecutionRecord, step: StepRecord): Promise<StepSlots> {
 		// These are all the edges that feed into the node this step runs.
 		const incomingEdges = execution.graph.edges.filter((edge) => edge.to === step.nodeId);
@@ -114,23 +122,31 @@ export class StepReadyHandler {
 			);
 		}
 
-		validateIncomingEdges(incomingEdges, step);
-
-		const predecessorNodeIds = [...new Set(incomingEdges.map((edge) => edge.from))];
-		const predecessorSteps = await this.stepStore.loadStepsByKeys(
+		const loops = deriveLoops(execution.graph);
+		// Only an exit edge reads the row that ended a loop, so a step with no exit
+		// edge needs no such read at all.
+		const terminalIterations = await loadTerminalIterations(
+			this.stepStore,
 			execution.id,
-			predecessorNodeIds.map((nodeId) => ({ nodeId, iteration: step.iteration })),
+			exitSourcesInto(execution.graph, loops, [step.nodeId]),
 		);
+		const reads = resolveInputReads(incomingEdges, loops, step, terminalIterations);
+
+		// One key per distinct row: a predecessor wired to two input slots is read
+		// twice but loaded once.
+		const rows = await this.stepStore.loadStepsByKeys(execution.id, [
+			...new Map(reads.map(({ key }) => [stepKeyId(key), key])).values(),
+		]);
 
 		// Array of length equal to the highest input slot plus one.
 		// The entries are `null` placeholders filled by the loop immediately below.
 		const inputs: StepSlots = Array.from(
-			{ length: Math.max(...incomingEdges.map((edge) => edge.inputIndex)) + 1 },
+			{ length: Math.max(...reads.map(({ edge }) => edge.inputIndex)) + 1 },
 			() => null,
 		);
 
-		for (const edge of incomingEdges) {
-			inputs[edge.inputIndex] = readEdgeValue(edge, step, predecessorSteps);
+		for (const { edge, key } of reads) {
+			inputs[edge.inputIndex] = readEdgeValue(edge, key, step, rows);
 		}
 
 		return inputs;
@@ -155,11 +171,44 @@ export class StepReadyHandler {
 	}
 }
 
-// Validate that the incoming edges meet our constraints. filledSlots tracks the filled
-// input slots so we can detect multiple edges into the same slot.
-function validateIncomingEdges(incomingEdges: GraphEdge[], step: StepRecord): void {
+/**
+ * Which row each incoming edge reads for this step, one per input slot.
+ *
+ * An edge that connects nothing at this iteration is dropped, which is what lets
+ * a batch node carry both an entry edge and a return edge on slot 0: they never
+ * apply at the same iteration, so per iteration the slot still has one source.
+ * Two edges that do both apply are the unsupported convergence case.
+ */
+export function resolveInputReads(
+	incomingEdges: GraphEdge[],
+	loops: WorkflowLoop[],
+	step: Pick<StepRecord, 'id' | 'nodeId' | 'iteration'>,
+	terminalIterations: Map<string, number>,
+): Array<{ edge: GraphEdge; key: StepKey }> {
+	const reads = incomingEdges.flatMap((edge) => {
+		const source = sourceRow(
+			edge,
+			classifyEdge(edge, loops),
+			step,
+			terminalIterations.get(edge.from),
+		);
+		if (source.kind === 'row') return [{ edge, key: source.key }];
+		if (source.kind === 'none') return [];
+		// The planner queues a step only once every row it reads exists, so a loop
+		// that has not ended means the rows and the plan disagree.
+		throw new UnexpectedError(
+			`step ${step.id} reads node ${edge.from} across a loop that has not ended`,
+		);
+	});
+
+	if (reads.length === 0) {
+		throw new UnexpectedError(
+			`step ${step.id} runs node ${step.nodeId}, which no edge reaches at iteration ${step.iteration}`,
+		);
+	}
+
 	const filledSlots: Set<number> = new Set();
-	for (const edge of incomingEdges) {
+	for (const { edge } of reads) {
 		if (filledSlots.has(edge.inputIndex)) {
 			// TODO(CAT-3982): same-slot convergence gets a defined meaning. We
 			// should have rejected this graph at validation time.
@@ -169,6 +218,8 @@ function validateIncomingEdges(incomingEdges: GraphEdge[], step: StepRecord): vo
 		}
 		filledSlots.add(edge.inputIndex);
 	}
+
+	return reads;
 }
 
 /**
@@ -178,10 +229,11 @@ function validateIncomingEdges(incomingEdges: GraphEdge[], step: StepRecord): vo
  */
 function readEdgeValue(
 	edge: GraphEdge,
+	source: StepKey,
 	step: StepRecord,
-	predecessorSteps: Record<StepKeyId, StepRecord>,
+	rows: Record<StepKeyId, StepRecord>,
 ): JsonValue {
-	const row = predecessorSteps[stepKeyId({ nodeId: edge.from, iteration: step.iteration })];
+	const row = rows[stepKeyId(source)];
 	if (!row || !isSettledStatus(row.status)) {
 		// A step is planned only once every predecessor settled, so running on
 		// a fabricated empty input would mask a planner/store inconsistency.

packages/@n8n/engine/src/execution/step-settled-handler.ts

@@ -1,22 +1,19 @@
 import { UnexpectedError } from '../common';
-import {
-	findTriggerNode,
-	getDescendantNodeIds,
-	getPredecessorNodeIds,
-	getSuccessorNodeIds,
-} from '../graph';
+import { deriveLoops, findTriggerNode, getDescendantNodeIds, getSuccessorNodeIds } from '../graph';
 import type { OrchestrationMessage, StepMessage, StepSettledEvent, WorkQueue } from '../queue';
+import { countExpectedSettledSteps } from './completion';
 import type { ExecutionRecord, ExecutionStore } from './execution-store';
 import { stepKeyId, type StepKey, type StepKeyId } from './execution.types';
-import { decideSuccessors } from './settlement';
-import type { StepRecord, StepStore, StepSummary } from './step-store';
+import { exitSourcesInto, loadTerminalIterations } from './loop-ledger';
+import { decideSuccessors, decisionKeys } from './settlement';
+import type { StepRecord, StepStore } from './step-store';
 import { validateStepContext } from './validate-step-context';
 
 /**
  * Handles the `step:settled` orchestration event: decides the fate of the
  * settled step's direct successors — queued when a live edge feeds them,
  * skipped when every input is settled dead (see the rules in `settlement.ts`)
- * — and records the execution's outcome once every reachable node has
+ * — and records the execution's outcome once every step the execution owes has
  * settled. Skips are settlements too: each one is announced back onto the
  * orchestration queue, and handling it here decides the next hop, so a dead
  * region cascades through the event loop one settlement at a time.
@@ -57,8 +54,7 @@ export class StepSettledHandler {
 				return;
 			}
 
-			const steps = await this.loadDecisionSteps(execution, step);
-			queued = await this.planSuccessors(execution, step, steps);
+			queued = await this.planSuccessors(execution, step);
 		}
 
 		// If we've queued steps, we know the execution isn't done yet, so we
@@ -74,12 +70,27 @@ export class StepSettledHandler {
 	}
 
 	/** Plans the settled step's direct successors, returning how many were queued. */
-	private async planSuccessors(
-		execution: ExecutionRecord,
-		step: StepRecord,
-		steps: Record<StepKeyId, StepSummary>,
-	): Promise<number> {
-		const { toQueue, toSkip } = decideSuccessors(execution.graph, step, steps);
+	private async planSuccessors(execution: ExecutionRecord, step: StepRecord): Promise<number> {
+		const loops = deriveLoops(execution.graph);
+		// Only the candidates' own edges are resolved, so only the loops those
+		// edges leave need their latest row read.
+		const candidates = getSuccessorNodeIds(execution.graph, step.nodeId);
+		const terminalIterations = await loadTerminalIterations(
+			this.stepStore,
+			execution.id,
+			exitSourcesInto(execution.graph, loops, candidates),
+		);
+		const steps = await this.stepStore.loadStepSummariesByKeys(
+			execution.id,
+			decisionKeys(execution.graph, loops, step, terminalIterations),
+		);
+		const { toQueue, toSkip } = decideSuccessors(
+			execution.graph,
+			loops,
+			step,
+			steps,
+			terminalIterations,
+		);
 		if (toQueue.length === 0 && toSkip.length === 0) return 0;
 
 		// One batch, so a settlement's consequence lands atomically and a fan-out
@@ -96,28 +107,6 @@ export class StepSettledHandler {
 		return await this.announceCreatedSteps(execution.id, created, new Set(toQueue.map(stepKeyId)));
 	}
 
-	/**
-	 * The rows a successor decision reads: the successors themselves (an
-	 * existing row means already decided) and their predecessors (settledness
-	 * and slot liveness), which include the settled node itself.
-	 */
-	private async loadDecisionSteps(
-		execution: ExecutionRecord,
-		step: StepRecord,
-	): Promise<Record<StepKeyId, StepSummary>> {
-		const successors = getSuccessorNodeIds(execution.graph, step.nodeId);
-		const nodeIds = [
-			...new Set([
-				...successors,
-				...successors.flatMap((id) => getPredecessorNodeIds(execution.graph, id)),
-			]),
-		];
-		// Forward edges stay within a pass, so every row read sits at the
-		// settled row's iteration.
-		const keys = nodeIds.map((nodeId) => ({ nodeId, iteration: step.iteration }));
-		return await this.stepStore.loadStepSummariesByKeys(execution.id, keys);
-	}
-
 	/**
 	 * Announces the created rows — `step:ready` for queued ones, `step:settled`
 	 * for skips, which settle at birth — and returns how many were queued.
@@ -141,27 +130,43 @@ export class StepSettledHandler {
 	}
 
 	/**
-	 * Records the execution's outcome once every reachable node has settled:
-	 * `failed` if any step failed, `completed` otherwise. Settled rows are
-	 * unique per node, only exist for reachable nodes, and never unsettle, so
+	 * Records the execution's outcome once every step it owes has settled: `failed`
+	 * if any step failed, `completed` otherwise. Steps are unique per
+	 * `(node, iteration)`, only exist for reachable nodes, and never unsettle, so
 	 * the count comparison cannot pass early — in-flight events and unplanned
-	 * successors both leave reachable nodes unsettled.
+	 * successors both leave steps outstanding.
 	 */
 	private async finishExecutionIfDone(execution: ExecutionRecord): Promise<void> {
+		const reachable = this.reachableNodeIds(execution);
+		const loops = deriveLoops(execution.graph);
+		const terminalIterations = await loadTerminalIterations(
+			this.stepStore,
+			execution.id,
+			// A loop the trigger cannot reach never receives rows, and the count below
+			// leaves it out, so its tip is not worth asking for.
+			loops
+				.filter((loop) => reachable.has(loop.batchNodeId))
+				.map((loop) => loop.batchNodeId),
+		);
+		const expected = countExpectedSettledSteps(loops, reachable, terminalIterations);
+		// A loop still running owes an unknown number of steps, so there is nothing
+		// to compare against yet.
+		if (expected === undefined) return;
+
 		const settled = await this.stepStore.countSettledSteps(execution.id);
-		if (settled < this.reachableNodeCount(execution)) return;
+		if (settled < expected) return;
 
 		const failed = await this.stepStore.hasFailedSteps(execution.id);
 		await this.executionStore.finishExecution(execution.id, failed ? 'failed' : 'completed');
 	}
 
-	private reachableNodeCount(execution: ExecutionRecord): number {
+	private reachableNodeIds(execution: ExecutionRecord): Set<string> {
 		const trigger = findTriggerNode(execution.graph);
 		if (!trigger) {
 			// The start boundary rejects triggerless graphs, so this execution
 			// should never have been created.
 			throw new UnexpectedError(`Execution ${execution.id} has no trigger node in its graph`);
 		}
-		return 1 + getDescendantNodeIds(execution.graph, trigger.id).length;
+		return new Set([trigger.id, ...getDescendantNodeIds(execution.graph, trigger.id)]);
 	}
 }

packages/@n8n/engine/src/execution/step-store.ts

@@ -139,16 +139,26 @@ export interface StepStore {
 	): Promise<Record<StepKeyId, StepSummary>>;
 
 	/**
-	 * The node's highest-iteration row, or `null` when it has none. For a
-	 * batch node this is the loop's ledger tip.
+	 * Planning view of each named node's highest-iteration row, keyed by node id,
+	 * omitting the nodes with no row. For a batch node this is the row that says
+	 * whether its loop has ended.
+	 *
+	 * One query for every node asked about, since a settlement can span several
+	 * loops. The row ending a loop holds everything that loop accumulated, so this
+	 * returns the same slim view as `loadStepSummariesByKeys`, not the whole row.
 	 */
-	loadLatestStep(executionId: string, nodeId: string): Promise<StepRecord | null>;
+	loadLatestStepSummaries(
+		executionId: string,
+		nodeIds: string[],
+	): Promise<Record<string, StepSummary>>;
 
 	/**
 	 * How many of the execution's steps have settled (completed, failed,
-	 * skipped, or cancelled). Rows are unique per node and only exist for
-	 * reachable nodes, so comparing this against the graph's reachable node
-	 * count answers "has everything settled?" exactly.
+	 * skipped, or cancelled). Rows are unique per `(node, iteration)`, only exist
+	 * for reachable nodes, and never unsettle, so comparing this against the
+	 * number of rows the execution owes answers "has everything settled?" exactly.
+	 * A loop makes that number more than the node count, so the comparison runs
+	 * against `expectedSettledRows` rather than against the graph.
 	 */
 	countSettledSteps(executionId: string): Promise<number>;