fix(core): Memoize pairedItem ancestry traversal (#38085)

Co-authored-by: Claude Fable 5 <[email protected]>
Co-authored-by: Emilia <[email protected]>

packages/testing/performance/benchmarks/paired-item/traversal.bench.ts

@@ -0,0 +1,250 @@
+/**
+ * Paired item ancestry traversal benchmarks.
+ *
+ * `$('Node').item` / `.itemMatching()` walk an item's `pairedItem` chain back
+ * through every ancestor, once per expression that uses them. Ancestry is a DAG,
+ * so an item reachable through many paths must be expanded once, not once per path.
+ *
+ * Keep the chains short: `itemMatching` also runs a `getParentNodes` connection
+ * check that is superlinear in the chain length and would otherwise dominate.
+ */
+import { bench, describe } from 'vitest';
+import { Workflow, WorkflowDataProxy } from 'n8n-workflow';
+import type {
+	IConnections,
+	IExecuteData,
+	INode,
+	INodeExecutionData,
+	INodeTypes,
+	IPairedItemData,
+	IRunExecutionData,
+	ITaskData,
+} from 'n8n-workflow';
+
+import { BENCH_OPTIONS } from '../bench-options';
+
+const MAIN = 'main';
+
+/** An unknown node type leaves the node as given; the traversal reads only run data and connections. */
+const nodeTypes: INodeTypes = {
+	getByName: () => undefined as never,
+	getByNameAndVersion: () => undefined as never,
+	getKnownTypes: () => ({}),
+};
+
+type ChainSpec = {
+	/** Items in each node's output; its length is the chain length. */
+	widths: number[];
+	pairing: (d: number, i: number, widths: number[]) => IPairedItemData | IPairedItemData[];
+};
+
+/** A -> B -> C -> ... with run data attached to every node. */
+function buildChain({ widths, pairing }: ChainSpec) {
+	const nodes: INode[] = [];
+	const connections: IConnections = {};
+	const runData: IRunExecutionData['resultData']['runData'] = {};
+
+	for (let d = 0; d < widths.length; d++) {
+		const name = `n${d}`;
+		nodes.push({
+			id: name,
+			name,
+			type: 'n8n-nodes-base.noOp',
+			typeVersion: 1,
+			position: [d * 100, 0],
+			parameters: {},
+		});
+
+		if (d < widths.length - 1) {
+			connections[name] = { [MAIN]: [[{ node: `n${d + 1}`, type: MAIN, index: 0 }]] };
+		}
+
+		const items: INodeExecutionData[] = [];
+		for (let i = 0; i < widths[d]!; i++) {
+			const item: INodeExecutionData = { json: { d, i } };
+			if (d > 0) item.pairedItem = pairing(d, i, widths);
+			items.push(item);
+		}
+
+		runData[name] = [
+			{
+				startTime: 0,
+				executionTime: 0,
+				executionIndex: d,
+				source: d === 0 ? [] : [{ previousNode: `n${d - 1}`, previousNodeRun: 0 }],
+				data: { [MAIN]: [items] },
+			} satisfies ITaskData,
+		];
+	}
+
+	return { nodes, connections, runData };
+}
+
+/** A proxy on the last node of the chain, ready to trace back to the first. */
+function proxyForChain(spec: ChainSpec) {
+	const { nodes, connections, runData } = buildChain(spec);
+	const activeNode = `n${spec.widths.length - 1}`;
+	const taskData = runData[activeNode]![0]!;
+	const inputData = taskData.data![MAIN]![0]!;
+
+	const executeData: IExecuteData = {
+		data: taskData.data!,
+		node: nodes[nodes.length - 1]!,
+		source: { [MAIN]: taskData.source },
+	};
+
+	return new WorkflowDataProxy(
+		new Workflow({ id: 'bench', name: 'bench', nodes, connections, active: false, nodeTypes }),
+		{ resultData: { runData } } as IRunExecutionData,
+		0,
+		0,
+		activeNode,
+		inputData,
+		{},
+		'manual',
+		{},
+		executeData,
+	).getDataProxy();
+}
+
+/** The last node's items carry a single paired item, the proxy's entry point into the walk. */
+const isLastNode = (d: number, widths: number[]) => d === widths.length - 1;
+
+/**
+ * origin -> {bN, cN} -> mN -> {bN+1, cN+1} -> ... : 2^diamonds paths to the last
+ * merge, where the chain shapes above reach a shared ancestor from one node.
+ */
+function diamondProxy(diamonds: number) {
+	const nodes: INode[] = [];
+	const connections: IConnections = {};
+	const runData: IRunExecutionData['resultData']['runData'] = {};
+
+	const addNode = (name: string) => {
+		nodes.push({
+			id: name,
+			name,
+			type: 'n8n-nodes-base.noOp',
+			typeVersion: 1,
+			position: [nodes.length * 100, 0],
+			parameters: {},
+		});
+	};
+	const link = (from: string, to: string, index: number) => {
+		connections[from] ??= { [MAIN]: [[]] };
+		connections[from][MAIN][0]!.push({ node: to, type: MAIN, index });
+	};
+	const task = (
+		source: ITaskData['source'],
+		pairedItem: IPairedItemData | IPairedItemData[] | undefined,
+	) =>
+		[
+			{
+				startTime: 0,
+				executionTime: 0,
+				executionIndex: 0,
+				source,
+				data: { [MAIN]: [[{ json: {}, ...(pairedItem ? { pairedItem } : {}) }]] },
+			} satisfies ITaskData,
+		] as ITaskData[];
+
+	addNode('origin');
+	runData.origin = task([], undefined);
+
+	let cur = 'origin';
+	for (let i = 0; i < diamonds; i++) {
+		const [b, c, m] = [`b${i}`, `c${i}`, `m${i}`];
+		for (const branch of [b, c]) {
+			addNode(branch);
+			link(cur, branch, 0);
+			runData[branch] = task([{ previousNode: cur, previousNodeRun: 0 }], { item: 0, input: 0 });
+		}
+		addNode(m);
+		link(b, m, 0);
+		link(c, m, 1);
+		runData[m] = task(
+			[
+				{ previousNode: b, previousNodeRun: 0 },
+				{ previousNode: c, previousNodeRun: 0 },
+			],
+			[
+				{ item: 0, input: 0 },
+				{ item: 0, input: 1 },
+			],
+		);
+		cur = m;
+	}
+
+	addNode('end');
+	link(cur, 'end', 0);
+	runData.end = task([{ previousNode: cur, previousNodeRun: 0 }], { item: 0, input: 0 });
+
+	const taskData = runData.end![0]!;
+	return new WorkflowDataProxy(
+		new Workflow({ id: 'bench', name: 'bench', nodes, connections, active: false, nodeTypes }),
+		{ resultData: { runData } } as IRunExecutionData,
+		0,
+		0,
+		'end',
+		taskData.data![MAIN]![0]!,
+		{},
+		'manual',
+		{},
+		{
+			data: taskData.data!,
+			node: nodes[nodes.length - 1]!,
+			source: { [MAIN]: taskData.source },
+		},
+	).getDataProxy();
+}
+
+describe('Paired item ancestry traversal', () => {
+	// Every item pairs to the same ancestor twice: 2^depth paths over `depth`
+	// distinct items. Exponential paths, must stay linear.
+	const recombining = proxyForChain({
+		widths: Array<number>(20).fill(1),
+		pairing: (d, i, widths) =>
+			isLastNode(d, widths)
+				? { item: i, input: 0 }
+				: [
+						{ item: 0, input: 0 },
+						{ item: 0, input: 0 },
+					],
+	});
+	bench(
+		'itemMatching: recombining chain (20 nodes, 2^20 paths)',
+		() => {
+			recombining.$('n0').itemMatching(0);
+		},
+		BENCH_OPTIONS,
+	);
+
+	// One item paired to 10k inputs that all trace to the same origin: no path is
+	// walked twice, so every visit is new work.
+	const FAN = 10_000;
+	const fanIn = proxyForChain({
+		widths: [1, FAN, 1, 1],
+		pairing: (d, i, widths) => {
+			if (isLastNode(d, widths)) return { item: i, input: 0 };
+			if (d === 2) return Array.from({ length: FAN }, (_, k) => ({ item: k, input: 0 }));
+			return { item: 0, input: 0 };
+		},
+	});
+	bench(
+		`itemMatching: fan-in (one item paired to ${FAN} ancestors)`,
+		() => {
+			fanIn.$('n0').itemMatching(0);
+		},
+		BENCH_OPTIONS,
+	);
+
+	// Branches converge from two separate parents, so a shared ancestor is reached
+	// under one state from several routes. Exponential paths, must stay linear.
+	const diamonds = diamondProxy(14);
+	bench(
+		'itemMatching: diamond chain (14 diamonds, 2^14 paths)',
+		() => {
+			diamonds.$('origin').itemMatching(0);
+		},
+		BENCH_OPTIONS,
+	);
+});

packages/workflow/src/workflow-data-proxy-paired-item-memo.ts

@@ -0,0 +1,42 @@
+import type { INodeExecutionData, IPairedItemData, ISourceData } from './interfaces';
+
+type PairedItemMemoResult =
+	| { ok: true; result: INodeExecutionData }
+	| { ok: false; error: unknown };
+
+/**
+ * Memoizes one paired item traversal per (node run output, item) state.
+ * For a fixed destination, revisiting a state yields the same result or error.
+ */
+export class PairedItemMemo {
+	private readonly entries = new Map<string, PairedItemMemoResult>();
+
+	/**
+	 * Resolves the memoized result for the given source data and paired item,
+	 * or computes and memoizes a new result if none exists.
+	 */
+	resolve(
+		sourceData: ISourceData,
+		pairedItem: IPairedItemData,
+		compute: () => INodeExecutionData,
+	): INodeExecutionData {
+		const key = this.toKey(sourceData, pairedItem);
+
+		let entry = this.entries.get(key);
+		if (!entry) {
+			try {
+				entry = { ok: true, result: compute() };
+			} catch (error) {
+				entry = { ok: false, error };
+			}
+			this.entries.set(key, entry);
+		}
+
+		if (entry.ok) return entry.result;
+		throw entry.error;
+	}
+
+	private toKey(sourceData: ISourceData, pairedItem: IPairedItemData): string {
+		return `${sourceData.previousNodeRun ?? 0}|${sourceData.previousNodeOutput ?? 0}|${pairedItem.item}|${sourceData.previousNode}`;
+	}
+}

packages/workflow/src/workflow-data-proxy.ts

@@ -43,6 +43,7 @@ import type { Workflow } from './workflow';
 import type { EnvProviderState } from './workflow-data-proxy-env-provider';
 import { createEnvProvider, createEnvProviderState } from './workflow-data-proxy-env-provider';
 import { getPinDataIfManualExecution } from './workflow-data-proxy-helpers';
+import { PairedItemMemo } from './workflow-data-proxy-paired-item-memo';
 
 const isScriptingNode = (nodeName: string, workflow: Workflow) => {
 	const node = workflow.getNode(nodeName);
@@ -1034,7 +1035,26 @@ export class WorkflowDataProxy {
 			incomingSourceData: ISourceData | null,
 			initialPairedItem: IPairedItemData,
 			usedMethodName: PairedItemMethod = PAIRED_ITEM_METHOD.$GET_PAIRED_ITEM,
-			nodeBeforeLast?: string,
+		): INodeExecutionData =>
+			resolvePairedItem(
+				destinationNodeName,
+				incomingSourceData,
+				initialPairedItem,
+				usedMethodName,
+				undefined,
+				// Ancestry is a DAG: branches recombine on shared ancestors (e.g. an
+				// Aggregate output pairing to all its inputs), so without memoization
+				// the walk revisits the same item exponentially often.
+				new PairedItemMemo(),
+			);
+
+		const resolvePairedItem = (
+			destinationNodeName: string,
+			incomingSourceData: ISourceData | null,
+			initialPairedItem: IPairedItemData,
+			usedMethodName: PairedItemMethod,
+			nodeBeforeLast: string | undefined,
+			memo: PairedItemMemo,
 		): INodeExecutionData => {
 			// Normalize inputs
 			const [pairedItem, sourceData] = normalizeInputs(initialPairedItem, incomingSourceData);
@@ -1043,6 +1063,26 @@ export class WorkflowDataProxy {
 				throw createPairedItemNotFound(destinationNodeName, nodeBeforeLast);
 			}
 
+			return memo.resolve(sourceData, pairedItem, () =>
+				resolvePairedItemUncached(
+					destinationNodeName,
+					sourceData,
+					pairedItem,
+					usedMethodName,
+					nodeBeforeLast,
+					memo,
+				),
+			);
+		};
+
+		const resolvePairedItemUncached = (
+			destinationNodeName: string,
+			sourceData: ISourceData,
+			pairedItem: IPairedItemData,
+			usedMethodName: PairedItemMethod,
+			nodeBeforeLast: string | undefined,
+			memo: PairedItemMemo,
+		): INodeExecutionData => {
 			const taskData = getTaskData(sourceData);
 			const outputData = getNodeOutput(taskData, sourceData, nodeBeforeLast);
 			const item = outputData[pairedItem.item];
@@ -1074,12 +1114,13 @@ export class WorkflowDataProxy {
 
 				try {
 					return createResultOk(
-						getPairedItem(
+						resolvePairedItem(
 							destinationNodeName,
 							nextSource,
 							{ ...nextPairedItem, input: inputIndex },
 							usedMethodName,
 							sourceData.previousNode,
+							memo,
 						),
 					);
 				} catch (error) {

packages/workflow/test/workflow-data-proxy-paired-item-memo.test.ts

@@ -0,0 +1,55 @@
+import type { INodeExecutionData } from '../src/interfaces';
+import { PairedItemMemo } from '../src/workflow-data-proxy-paired-item-memo';
+
+describe('PairedItemMemo', () => {
+	const source = { previousNode: 'A', previousNodeRun: 1, previousNodeOutput: 2 };
+	const item: INodeExecutionData = { json: {} };
+
+	test('computes once per state and replays the result', () => {
+		const memo = new PairedItemMemo();
+		const compute = vi.fn(() => item);
+
+		expect(memo.resolve(source, { item: 0 }, compute)).toBe(item);
+		expect(memo.resolve(source, { item: 0 }, compute)).toBe(item);
+		expect(compute).toHaveBeenCalledTimes(1);
+	});
+
+	test('replays the same error without recomputing', () => {
+		const memo = new PairedItemMemo();
+		const error = new Error('boom');
+		const compute = vi.fn(() => {
+			throw error;
+		});
+
+		expect(() => memo.resolve(source, { item: 0 }, compute)).toThrow(error);
+		expect(() => memo.resolve(source, { item: 0 }, compute)).toThrow(error);
+		expect(compute).toHaveBeenCalledTimes(1);
+	});
+
+	test('treats omitted run and output as 0', () => {
+		const memo = new PairedItemMemo();
+		const compute = vi.fn(() => item);
+
+		memo.resolve({ previousNode: 'A' }, { item: 0 }, compute);
+		memo.resolve(
+			{ previousNode: 'A', previousNodeRun: 0, previousNodeOutput: 0 },
+			{ item: 0 },
+			compute,
+		);
+		expect(compute).toHaveBeenCalledTimes(1);
+	});
+
+	test.each([
+		['node', { ...source, previousNode: 'B' }, { item: 0 }],
+		['run', { ...source, previousNodeRun: 0 }, { item: 0 }],
+		['output', { ...source, previousNodeOutput: 0 }, { item: 0 }],
+		['item', source, { item: 1 }],
+	])('keys a different %s as a distinct state', (_, otherSource, otherItem) => {
+		const memo = new PairedItemMemo();
+		const compute = vi.fn(() => item);
+
+		memo.resolve(source, { item: 0 }, compute);
+		memo.resolve(otherSource, otherItem, compute);
+		expect(compute).toHaveBeenCalledTimes(2);
+	});
+});

packages/workflow/test/workflow-data-proxy.test.ts

@@ -9,6 +9,8 @@ import {
 	type IDataObject,
 	type IExecuteData,
 	type INode,
+	type INodeExecutionData,
+	type IPairedItemData,
 	type IPinData,
 	type IRun,
 	type IWorkflowBase,
@@ -2264,3 +2266,360 @@ describe('WorkflowDataProxy', () => {
 		});
 	});
 });
+
+describe('WorkflowDataProxy → pairedItem traversal through recombining ancestry', () => {
+	// Models a polling loop whose Aggregate-style node pairs its single output
+	// item to all of its inputs: ancestry branches fan out at every pass and
+	// recombine on the shared ancestor, forming a DAG. Traversal must resolve
+	// this in polynomial time - a tree walk would take ROWS^PASSES steps.
+	const ROWS = 8;
+	const PASSES = 25;
+
+	const makeWorkflow = (): IWorkflowBase => ({
+		id: '1',
+		name: 'test',
+		nodes: (['Start', 'Rows', 'Agg', 'End'] as const).map((name, i) => ({
+			id: `uuid-${i}`,
+			name,
+			type: 'n8n-nodes-base.code',
+			typeVersion: 1,
+			position: [i * 100, 0] as [number, number],
+			parameters: {},
+		})),
+		connections: {
+			Start: { main: [[{ node: 'Rows', type: NodeConnectionTypes.Main, index: 0 }]] },
+			Rows: { main: [[{ node: 'Agg', type: NodeConnectionTypes.Main, index: 0 }]] },
+			Agg: { main: [[{ node: 'End', type: NodeConnectionTypes.Main, index: 0 }]] },
+		},
+		active: false,
+		activeVersionId: null,
+		isArchived: false,
+		createdAt: new Date(),
+		updatedAt: new Date(),
+	});
+
+	const taskData = (source: Array<{ previousNode: string; previousNodeRun?: number } | null>) => ({
+		startTime: 0,
+		executionTime: 0,
+		executionIndex: 0,
+		source,
+		data: { main: [[]] as INodeExecutionData[][] },
+	});
+
+	const makeRun = (
+		startItems: INodeExecutionData[],
+		rowsPairing: (row: number) => number,
+	): IRun => {
+		const start = taskData([null]);
+		start.data.main[0] = startItems;
+
+		const rowsRuns = [];
+		const aggRuns = [];
+		for (let pass = 0; pass < PASSES; pass++) {
+			const rows = taskData([
+				pass === 0 ? { previousNode: 'Start' } : { previousNode: 'Agg', previousNodeRun: pass - 1 },
+			]);
+			// only pass 0 is sourced from Start (multiple items); later passes are
+			// sourced from Agg, which emits a single item
+			rows.data.main[0] = Array.from({ length: ROWS }, (_, row) => ({
+				json: { row },
+				pairedItem: { item: pass === 0 ? rowsPairing(row) : 0 },
+			}));
+			rowsRuns.push(rows);
+
+			const agg = taskData([{ previousNode: 'Rows', previousNodeRun: pass }]);
+			agg.data.main[0] = [
+				{ json: { pass }, pairedItem: Array.from({ length: ROWS }, (_, item) => ({ item })) },
+			];
+			aggRuns.push(agg);
+		}
+
+		const end = taskData([{ previousNode: 'Agg', previousNodeRun: PASSES - 1 }]);
+		end.data.main[0] = [{ json: {}, pairedItem: { item: 0 } }];
+
+		return {
+			data: createRunExecutionData({
+				resultData: { runData: { Start: [start], Rows: rowsRuns, Agg: aggRuns, End: [end] } },
+			}),
+			mode: 'manual',
+			startedAt: new Date(),
+			status: 'success',
+			storedAt: 'db',
+		};
+	};
+
+	test('resolves when all branches recombine on one ancestor item', () => {
+		const run = makeRun([{ json: { origin: true }, pairedItem: { item: 0 } }], () => 0);
+		const proxy = getProxyFromFixture(makeWorkflow(), run, 'End');
+
+		expect(proxy.$('Start').item.json).toEqual({ origin: true });
+	});
+
+	test('still detects ambiguity when branches resolve to different ancestor items', () => {
+		const startItems = Array.from({ length: ROWS }, (_, item) => ({
+			json: { item },
+			pairedItem: { item },
+		}));
+		// each row pairs back to its own Start item, so resolution is ambiguous
+		const run = makeRun(startItems, (row) => row);
+		const proxy = getProxyFromFixture(makeWorkflow(), run, 'End');
+
+		expect(() => proxy.$('Start').item).toThrowError('Multiple matches');
+	});
+});
+
+describe('WorkflowDataProxy → pairedItem traversal with sourceOverwrite routes', () => {
+	// One item pairs to the same ancestor node run twice: once through its task's
+	// default source and once through a `sourceOverwrite` redirect. The two routes
+	// must resolve as distinct states.
+	const makeWorkflow = (): IWorkflowBase => ({
+		id: '1',
+		name: 'test',
+		nodes: (['Start', 'Mid', 'Join', 'End'] as const).map((name, i) => ({
+			id: `uuid-${i}`,
+			name,
+			type: 'n8n-nodes-base.code',
+			typeVersion: 1,
+			position: [i * 100, 0] as [number, number],
+			parameters: {},
+		})),
+		connections: {
+			Start: { main: [[{ node: 'Mid', type: NodeConnectionTypes.Main, index: 0 }]] },
+			Mid: { main: [[{ node: 'Join', type: NodeConnectionTypes.Main, index: 0 }]] },
+			Join: { main: [[{ node: 'End', type: NodeConnectionTypes.Main, index: 0 }]] },
+		},
+		active: false,
+		activeVersionId: null,
+		isArchived: false,
+		createdAt: new Date(),
+		updatedAt: new Date(),
+	});
+
+	const midRun = (outputs: INodeExecutionData[][]) => ({
+		startTime: 0,
+		executionTime: 0,
+		executionIndex: 0,
+		source: [{ previousNode: 'Start' }],
+		data: { main: outputs },
+	});
+
+	// Mid run N emits one item paired to Start item `midRunPairing[N]`
+	const midRunsPairedTo = (midRunPairing: number[]) =>
+		midRunPairing.map((startItem) => midRun([[{ json: {}, pairedItem: { item: startItem } }]]));
+
+	const defaultJoinPairing: IPairedItemData[] = [
+		{ item: 0 },
+		{ item: 0, sourceOverwrite: { previousNode: 'Mid', previousNodeRun: 1 } },
+	];
+
+	const makeRun = (
+		startItems: INodeExecutionData[],
+		midRuns: ReturnType<typeof midRun>[],
+		joinPairing: IPairedItemData[] = defaultJoinPairing,
+	): IRun => {
+		const start = {
+			startTime: 0,
+			executionTime: 0,
+			executionIndex: 0,
+			source: [null],
+			data: { main: [startItems] },
+		};
+
+		const join = {
+			startTime: 0,
+			executionTime: 0,
+			executionIndex: 0,
+			source: [{ previousNode: 'Mid', previousNodeRun: 0 }],
+			data: { main: [[{ json: {}, pairedItem: joinPairing }]] },
+		};
+
+		const end = {
+			startTime: 0,
+			executionTime: 0,
+			executionIndex: 0,
+			source: [{ previousNode: 'Join' }],
+			data: { main: [[{ json: {}, pairedItem: { item: 0 } }]] },
+		};
+
+		return {
+			data: createRunExecutionData({
+				resultData: { runData: { Start: [start], Mid: midRuns, Join: [join], End: [end] } },
+			}),
+			mode: 'manual',
+			startedAt: new Date(),
+			status: 'success',
+			storedAt: 'db',
+		};
+	};
+
+	test('resolves when the default and overwritten routes agree on the ancestor item', () => {
+		const run = makeRun(
+			[{ json: { origin: true }, pairedItem: { item: 0 } }],
+			midRunsPairedTo([0, 0]),
+		);
+		const proxy = getProxyFromFixture(makeWorkflow(), run, 'End');
+
+		expect(proxy.$('Start').item.json).toEqual({ origin: true });
+	});
+
+	test('detects ambiguity when the overwritten route reaches a different ancestor item', () => {
+		const startItems = [
+			{ json: { item: 0 }, pairedItem: { item: 0 } },
+			{ json: { item: 1 }, pairedItem: { item: 1 } },
+		];
+		const run = makeRun(startItems, midRunsPairedTo([0, 1]));
+		const proxy = getProxyFromFixture(makeWorkflow(), run, 'End');
+
+		expect(() => proxy.$('Start').item).toThrowError('Multiple matches');
+	});
+
+	test('returns the resolving route when another route fails, also on its memoized revisit', () => {
+		const startItems = [{ json: { origin: true }, pairedItem: { item: 0 } }];
+		// Mid run 1 has no pairedItem, so the overwritten route cannot resolve.
+		// It is listed twice so the second visit replays the memoized error.
+		const brokenRoute = { item: 0, sourceOverwrite: { previousNode: 'Mid', previousNodeRun: 1 } };
+		const midRuns = [...midRunsPairedTo([0]), midRun([[{ json: {} }]])];
+
+		const okFirst = getProxyFromFixture(
+			makeWorkflow(),
+			makeRun(startItems, midRuns, [{ item: 0 }, brokenRoute, brokenRoute]),
+			'End',
+		);
+		expect(okFirst.$('Start').item.json).toEqual({ origin: true });
+
+		const okLast = getProxyFromFixture(
+			makeWorkflow(),
+			makeRun(startItems, midRuns, [brokenRoute, brokenRoute, { item: 0 }]),
+			'End',
+		);
+		expect(okLast.$('Start').item.json).toEqual({ origin: true });
+
+		const allBroken = getProxyFromFixture(
+			makeWorkflow(),
+			makeRun(startItems, midRuns, [brokenRoute, brokenRoute]),
+			'End',
+		);
+		expect(() => allBroken.$('Start').item).toThrowError(
+			"Paired item data for item from node 'Mid' is unavailable",
+		);
+	});
+
+	test('treats the same item index on different outputs of one run as distinct states', () => {
+		const startItems = [
+			{ json: { item: 0 }, pairedItem: { item: 0 } },
+			{ json: { item: 1 }, pairedItem: { item: 1 } },
+		];
+		// Mid run 0 output 0 item 0 pairs to Start 0, output 1 item 0 pairs to Start 1
+		const midRuns = [
+			midRun([[{ json: {}, pairedItem: { item: 0 } }], [{ json: {}, pairedItem: { item: 1 } }]]),
+		];
+		const run = makeRun(startItems, midRuns, [
+			{
+				item: 0,
+				sourceOverwrite: { previousNode: 'Mid', previousNodeRun: 0, previousNodeOutput: 0 },
+			},
+			{
+				item: 0,
+				sourceOverwrite: { previousNode: 'Mid', previousNodeRun: 0, previousNodeOutput: 1 },
+			},
+		]);
+		const proxy = getProxyFromFixture(makeWorkflow(), run, 'End');
+
+		expect(() => proxy.$('Start').item).toThrowError('Multiple matches');
+	});
+});
+
+describe('WorkflowDataProxy → pairedItem traversal through a diamond', () => {
+	// Start fans out to Left and Right, which join again in Merge. A Start item is
+	// reached under one state from two different parent nodes.
+	const makeWorkflow = (): IWorkflowBase => ({
+		id: '1',
+		name: 'test',
+		nodes: (['Start', 'Left', 'Right', 'Merge', 'End'] as const).map((name, i) => ({
+			id: `uuid-${i}`,
+			name,
+			type: 'n8n-nodes-base.code',
+			typeVersion: 1,
+			position: [i * 100, 0] as [number, number],
+			parameters: {},
+		})),
+		connections: {
+			Start: {
+				main: [
+					[
+						{ node: 'Left', type: NodeConnectionTypes.Main, index: 0 },
+						{ node: 'Right', type: NodeConnectionTypes.Main, index: 0 },
+					],
+				],
+			},
+			Left: { main: [[{ node: 'Merge', type: NodeConnectionTypes.Main, index: 0 }]] },
+			Right: { main: [[{ node: 'Merge', type: NodeConnectionTypes.Main, index: 1 }]] },
+			Merge: { main: [[{ node: 'End', type: NodeConnectionTypes.Main, index: 0 }]] },
+		},
+		active: false,
+		activeVersionId: null,
+		isArchived: false,
+		createdAt: new Date(),
+		updatedAt: new Date(),
+	});
+
+	const task = (source: Array<{ previousNode: string } | null>, items: INodeExecutionData[]) => ({
+		startTime: 0,
+		executionTime: 0,
+		executionIndex: 0,
+		source,
+		data: { main: [items] },
+	});
+
+	const makeRun = (
+		startItems: INodeExecutionData[],
+		leftItem: number,
+		rightItem: number,
+	): IRun => ({
+		data: createRunExecutionData({
+			resultData: {
+				runData: {
+					Start: [task([null], startItems)],
+					Left: [task([{ previousNode: 'Start' }], [{ json: {}, pairedItem: { item: leftItem } }])],
+					Right: [
+						task([{ previousNode: 'Start' }], [{ json: {}, pairedItem: { item: rightItem } }]),
+					],
+					Merge: [
+						task(
+							[{ previousNode: 'Left' }, { previousNode: 'Right' }],
+							[
+								{
+									json: {},
+									pairedItem: [
+										{ item: 0, input: 0 },
+										{ item: 0, input: 1 },
+									],
+								},
+							],
+						),
+					],
+					End: [task([{ previousNode: 'Merge' }], [{ json: {}, pairedItem: { item: 0 } }])],
+				},
+			},
+		}),
+		mode: 'manual',
+		startedAt: new Date(),
+		status: 'success',
+		storedAt: 'db',
+	});
+
+	test('resolves when both routes reach one ancestor item', () => {
+		const run = makeRun([{ json: { origin: true }, pairedItem: { item: 0 } }], 0, 0);
+		const proxy = getProxyFromFixture(makeWorkflow(), run, 'End');
+
+		expect(proxy.$('Start').item.json).toEqual({ origin: true });
+	});
+
+	test('detects ambiguity when the routes reach different ancestor items', () => {
+		const startItems = [0, 1].map((item) => ({ json: { item }, pairedItem: { item } }));
+		const run = makeRun(startItems, 0, 1);
+		const proxy = getProxyFromFixture(makeWorkflow(), run, 'End');
+
+		expect(() => proxy.$('Start').item).toThrowError('Multiple matches');
+	});
+});