fix(ai-builder): Enforce declared field names on eval pin data (no-changelog) (#34960)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-09-01 05:38:33 +08:00 · 2026-07-28 14:11:02 +01:00
parent 313be7c401
commit 6d8b01dc79
15 changed files with 928 additions and 31 deletions
@@ -11,6 +11,7 @@ This is a test environment. No real credentials or API connections exist. ALL HT
 IMPORTANT: Nodes receiving mock responses instead of real API responses is EXPECTED. Missing or mock credentials is EXPECTED. Don't flag these as issues — they are the testing mechanism itself.
 IMPORTANT: When an AI root node such as an AI Agent is pinned, its connected AI subnodes (language model, memory, tools, retrievers, parsers) often do not run. This is expected. Evaluate those subnodes from the saved workflow structure, connections, and all-node configs instead of failing only because the subnode did not execute.
 IMPORTANT: AI root output shapes differ by node type, and pinned outputs follow the REAL node behavior. An Agent root wraps its result in \`{ "output": ... }\`. A Basic LLM Chain (\`chainLlm\`) does NOT: with a structured output parser attached it emits the parsed fields FLAT at the top level of \`json\` (the parser unwraps any \`output\` envelope); without a parser it emits \`{ "text": "..." }\`. A downstream expression reading \`$json.output.*\` from a chainLlm root can never resolve against the real node either — that is a builder_issue (wrong expression for the node type), NOT a mock/pin issue. Do not claim the pin "should have had an output wrapper" for chainLlm.
+IMPORTANT: When the harness resolved the table's real columns, pinned Data Table read outputs mirror that column schema — it reads the actual columns off the table the builder created and enforces them on the pinned rows. In that case, when a downstream expression reads a field that is not among the pinned rows' keys (e.g. \`$json.contact_email\` while the rows carry \`email\`), the same expression would resolve undefined against the real table too — that is a builder_issue (expression references a column the builder never created), NOT a mock/pin issue.

 Credential ID values in the workflow JSON (real, placeholder strings, or stale references) never cause execution failures. When a credential ID cannot be resolved, the framework substitutes a mock credential and execution proceeds. Do not cite credential ID values as a root cause of failure under any circumstance.

@@ -264,10 +264,12 @@ describe('emit-instance-ai', () => {
 			// Mock/pin-data generation building blocks (src/mock-data/) — eval
 			// and simulated-verification tooling, never in workflow bodies
 			'buildDateAnchors',
+			'buildFieldViolationRetryMessage',
 			'buildNodeSchemaSection',
 			'buildPinDataUserPrompt',
 			'buildSchemaContexts',
 			'collectDownstreamConsumers',
+			'collectPinFieldViolations',
 			'describeAiRootShape',
 			'findEnvelopeKey',
 			'findOutputParserTargets',
@@ -1,42 +1,191 @@
-import type { NodeSchemaContext, OutputParserContext, OutputSchemaLookup } from './types';
+import { DATA_TABLE_SYSTEM_COLUMNS } from 'n8n-workflow';
+
+import { findEnvelopeKey } from './ai-root-shapes';
+import type {
+	DataTableColumnInfo,
+	DeclaredFieldContract,
+	NodeSchemaContext,
+	OutputParserContext,
+	OutputSchemaLookup,
+} from './types';
 import type { NodeJSON, WorkflowJSON } from '../types/base';

+export const INFORMATION_EXTRACTOR_NODE_TYPE = '@n8n/n8n-nodes-langchain.informationExtractor';
+
 /**
 * Assemble the per-node contexts the generation prompt is built from.
 * Schema enrichment happens through the injected lookup (consumers pass
 * n8n-core's `__schema__` resolver); absent lookup = no schema sections.
+ * `dataTableColumns` (node name → real table columns) comes from consumers
+ * with instance access — the pinned rows must mirror those exact keys.
 */
 export function buildSchemaContexts(
 	nodes: NodeJSON[],
 	outputSchemaLookup?: OutputSchemaLookup,
 	outputParserTargets?: Map<string, OutputParserContext>,
+	dataTableColumns?: Record<string, DataTableColumnInfo[]>,
 ): NodeSchemaContext[] {
 	return nodes.map((node) => {
 		const params = node.parameters as Record<string, unknown> | undefined;
 		const resource = typeof params?.resource === 'string' ? params.resource : undefined;
 		const operation = typeof params?.operation === 'string' ? params.operation : undefined;
-		const outputParser = node.name ? outputParserTargets?.get(node.name) : undefined;
+		// An information extractor declares its output schema in its OWN
+		// parameters (there is no parser sub-node to read it from) — surface it
+		// through the same outputParser slot so the prompt embeds it.
+		const outputParser =
+			(node.name ? outputParserTargets?.get(node.name) : undefined) ??
+			(node.type === INFORMATION_EXTRACTOR_NODE_TYPE
+				? extractInformationExtractorSchema(params)
+				: undefined);

 		const schema = outputSchemaLookup?.({
 			type: node.type,
 			typeVersion: node.typeVersion,
 			resource,
 			operation,
-			hasOutputParser: outputParser !== undefined,
+			hasOutputParser: node.name ? outputParserTargets?.has(node.name) === true : false,
 		});

+		const nodeName = node.name ?? node.type;
+		const columns = dataTableColumns?.[nodeName];
+
 		return {
-			nodeName: node.name ?? node.type,
+			nodeName,
 			nodeType: node.type,
 			typeVersion: node.typeVersion,
 			resource,
 			operation,
 			schema,
 			outputParser,
+			dataTableColumns: columns,
+			declaredFields: buildDeclaredFieldContract(node.type, schema, outputParser, columns),
 		};
 	});
 }

+/**
+ * Envelope key the parsed fields sit under, if any. Shared with the prompt
+ * builder so the shape asked for and the shape enforced can't drift: the
+ * extractor always wraps in `output` even when the `__schema__` lookup is
+ * unavailable, while parser targets get theirs from the resolved with-parser
+ * schema variant.
+ */
+export function resolveEnvelopeKey(
+	nodeType: string,
+	schema: Record<string, unknown> | undefined,
+): string | undefined {
+	return (
+		findEnvelopeKey(schema) ?? (nodeType === INFORMATION_EXTRACTOR_NODE_TYPE ? 'output' : undefined)
+	);
+}
+
+/**
+ * Derive the field-name contract pinned items are validated against.
+ * Data Table columns are exact (real rows always carry every column);
+ * schema-declared fields allow a subset (optional fields may be absent).
+ */
+function buildDeclaredFieldContract(
+	nodeType: string,
+	schema: Record<string, unknown> | undefined,
+	outputParser: OutputParserContext | undefined,
+	columns: DataTableColumnInfo[] | undefined,
+): DeclaredFieldContract | undefined {
+	if (columns && columns.length > 0) {
+		return {
+			keys: [...DATA_TABLE_SYSTEM_COLUMNS, ...columns.map((c) => c.name)],
+			exact: true,
+			source: 'data-table-columns',
+		};
+	}
+
+	if (outputParser?.schemaText) {
+		const keys = deriveTopLevelKeys(outputParser.schemaText, outputParser.schemaIsExample);
+		if (keys.length > 0) {
+			const envelopeKey = resolveEnvelopeKey(nodeType, schema);
+			return { keys, envelopeKey, exact: false, source: 'declared-schema' };
+		}
+	}
+
+	return undefined;
+}
+
+/** Top-level field names of a JSON Schema (`properties` keys) or an example object. */
+function deriveTopLevelKeys(schemaText: string, isExample: boolean): string[] {
+	let parsed: unknown;
+	try {
+		parsed = JSON.parse(schemaText);
+	} catch {
+		return [];
+	}
+	if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) return [];
+	const record = parsed as Record<string, unknown>;
+	if (isExample) return Object.keys(record);
+	const properties = record.properties;
+	if (typeof properties !== 'object' || properties === null) return [];
+	return Object.keys(properties);
+}
+
+/**
+ * Read the information extractor's own declared schema off its parameters:
+ * `fromAttributes` holds an attribute list (synthesized into a JSON Schema
+ * here), `fromJson` an example object, `manual` a JSON Schema — mirroring the
+ * fields the structured output parser node uses.
+ */
+function extractInformationExtractorSchema(
+	params: Record<string, unknown> | undefined,
+): OutputParserContext | undefined {
+	if (!params) return undefined;
+	const schemaType = typeof params.schemaType === 'string' ? params.schemaType : 'fromAttributes';
+
+	if (schemaType === 'fromAttributes') {
+		const attributesWrapper = params.attributes as Record<string, unknown> | undefined;
+		const attributes = Array.isArray(attributesWrapper?.attributes)
+			? attributesWrapper.attributes
+			: [];
+		const properties: Record<string, unknown> = {};
+		const required: string[] = [];
+		for (const attribute of attributes) {
+			if (typeof attribute !== 'object' || attribute === null) continue;
+			const { name, type, description } = attribute as Record<string, unknown>;
+			if (typeof name !== 'string' || name.length === 0) continue;
+			properties[name] = {
+				type: typeof type === 'string' ? type : 'string',
+				...(typeof description === 'string' && description ? { description } : {}),
+			};
+			if ((attribute as Record<string, unknown>).required === true) required.push(name);
+		}
+		if (Object.keys(properties).length === 0) return undefined;
+		return {
+			schemaText: JSON.stringify({ type: 'object', properties, required }, null, 2),
+			schemaIsExample: false,
+		};
+	}
+
+	const candidate = schemaType === 'manual' ? params.inputSchema : params.jsonSchemaExample;
+	const schemaText = schemaDeclarationText(candidate);
+	if (schemaText) {
+		return { schemaText, schemaIsExample: schemaType !== 'manual' };
+	}
+	return undefined;
+}
+
+/**
+ * Schema declarations are JSON strings in the editor, but eval builders
+ * sometimes store them as raw objects — those still declare the field names
+ * the pin must follow, so read both forms.
+ */
+export function schemaDeclarationText(candidate: unknown): string | undefined {
+	if (typeof candidate === 'string' && candidate.trim().length > 0) return candidate.trim();
+	if (typeof candidate === 'object' && candidate !== null) {
+		try {
+			return JSON.stringify(candidate);
+		} catch {
+			return undefined;
+		}
+	}
+	return undefined;
+}
+
 /**
 * Map AI root node name → structured output parser context, discovered from
 * `ai_outputParser` connections (parser node is the connection SOURCE, the
@@ -87,8 +236,9 @@ function extractParserContext(parserNode: NodeJSON | undefined): OutputParserCon
 		[example, true],
 		[legacySchema, false],
 	] as Array<[unknown, boolean]>) {
-		if (typeof candidate === 'string' && candidate.trim().length > 0) {
-			return { schemaText: candidate.trim(), schemaIsExample: isExample };
+		const schemaText = schemaDeclarationText(candidate);
+		if (schemaText) {
+			return { schemaText, schemaIsExample: isExample };
 		}
 	}

@@ -4,6 +4,8 @@ export type {
 	OutputParserContext,
 	NodeSchemaContext,
 	PinDataGenerationInstructions,
+	DataTableColumnInfo,
+	DeclaredFieldContract,
 } from './types';
 export {
 	AGENT_NODE_TYPE,
@@ -25,3 +27,8 @@ export {
 	type BuildPinDataUserPromptOptions,
 } from './prompt';
 export { parsePinDataResponse, repairStructuredOutput } from './parse';
+export {
+	collectPinFieldViolations,
+	buildFieldViolationRetryMessage,
+	type PinFieldViolation,
+} from './validate';
@@ -3,8 +3,9 @@ import { buildDateAnchors } from './date-anchors';
 import { workflowToMermaid } from './mermaid';
 import { parsePinDataResponse, repairStructuredOutput } from './parse';
 import { buildNodeSchemaSection, buildPinDataUserPrompt } from './prompt';
-import type { OutputSchemaLookup } from './types';
-import type { WorkflowJSON } from '../types/base';
+import type { NodeSchemaContext, OutputSchemaLookup } from './types';
+import { buildFieldViolationRetryMessage, collectPinFieldViolations } from './validate';
+import type { NodeJSON, WorkflowJSON } from '../types/base';

 const workflow = {
 	nodes: [
@@ -333,6 +334,255 @@ describe('repairStructuredOutput', () => {
 	});
 });

+describe('information extractor own-schema enrichment', () => {
+	const extractorNode = {
+		name: 'Extract Invoice Details',
+		type: '@n8n/n8n-nodes-langchain.informationExtractor',
+		typeVersion: 1.2,
+		parameters: {
+			schemaType: 'fromAttributes',
+			attributes: {
+				attributes: [
+					{ name: 'invoice_number', type: 'string', description: 'The invoice id', required: true },
+					{ name: 'total_amount', type: 'number', required: false },
+				],
+			},
+		},
+	} as unknown as NodeJSON;
+	const extractorWorkflow = {
+		nodes: [extractorNode],
+		connections: {},
+	} as unknown as WorkflowJSON;
+
+	it('surfaces the declared attributes as a schema and a field contract', () => {
+		const [ctx] = buildSchemaContexts([extractorNode]);
+
+		expect(ctx.outputParser?.schemaIsExample).toBe(false);
+		expect(ctx.outputParser?.schemaText).toContain('"invoice_number"');
+		expect(ctx.outputParser?.schemaText).toContain('"total_amount"');
+		expect(ctx.outputParser?.schemaText).toContain('"required": [\n    "invoice_number"\n  ]');
+		expect(ctx.declaredFields).toEqual({
+			keys: ['invoice_number', 'total_amount'],
+			envelopeKey: 'output',
+			exact: false,
+			source: 'declared-schema',
+		});
+	});
+
+	it('embeds the attribute names in the prompt schema section', () => {
+		const [ctx] = buildSchemaContexts([extractorNode]);
+		const section = buildNodeSchemaSection(ctx).join('\n');
+
+		expect(section).toContain('total_amount');
+		expect(section).toContain('use its exact field names');
+	});
+
+	it('reads fromJson examples and manual schemas off the extractor parameters', () => {
+		const fromJson = {
+			...extractorNode,
+			parameters: { schemaType: 'fromJson', jsonSchemaExample: '{"po_number": "PO-1"}' },
+		};
+		expect(buildSchemaContexts([fromJson])[0].declaredFields).toMatchObject({
+			keys: ['po_number'],
+			envelopeKey: 'output',
+		});
+
+		const manual = {
+			...extractorNode,
+			parameters: {
+				schemaType: 'manual',
+				inputSchema: '{"type":"object","properties":{"po_number":{}}}',
+			},
+		};
+		expect(buildSchemaContexts([manual])[0].declaredFields).toMatchObject({
+			keys: ['po_number'],
+		});
+	});
+
+	it('reads schema declarations stored as raw objects (eval-builder quirk)', () => {
+		const rawObjectExample = {
+			...extractorNode,
+			parameters: {
+				schemaType: 'fromJson',
+				jsonSchemaExample: { amount: 4750, po_number: 'PO-1' },
+			},
+		};
+		expect(buildSchemaContexts([rawObjectExample])[0].declaredFields).toMatchObject({
+			keys: ['amount', 'po_number'],
+			envelopeKey: 'output',
+		});
+
+		const parserWorkflow = {
+			nodes: [
+				{
+					name: 'Invoice Schema',
+					type: '@n8n/n8n-nodes-langchain.outputParserStructured',
+					typeVersion: 1.2,
+					parameters: { schemaType: 'fromJson', jsonSchemaExample: { amount: 1 } },
+				},
+			],
+			connections: {
+				'Invoice Schema': {
+					ai_outputParser: [[{ node: 'AI Root', type: 'ai_outputParser', index: 0 }]],
+				},
+			},
+		} as unknown as WorkflowJSON;
+		expect(findOutputParserTargets(parserWorkflow).get('AI Root')).toEqual({
+			schemaText: '{"amount":1}',
+			schemaIsExample: true,
+		});
+	});
+
+	it('repairs flat extractor pins into the output envelope without a parser connection', () => {
+		const flat = { 'Extract Invoice Details': [{ json: { invoice_number: 'INV-1' } }] };
+
+		expect(
+			repairStructuredOutput(flat, extractorWorkflow, buildSchemaContexts([extractorNode]))[
+				'Extract Invoice Details'
+			][0],
+		).toEqual({ json: { output: { invoice_number: 'INV-1' } } });
+	});
+});
+
+describe('data table column contracts', () => {
+	const dataTableNode = workflow.nodes[0]; // Get Rows
+
+	it('builds an exact contract including the system columns', () => {
+		const [ctx] = buildSchemaContexts([dataTableNode], undefined, undefined, {
+			'Get Rows': [
+				{ name: 'contact_email', type: 'string' },
+				{ name: 'contact_name', type: 'string' },
+			],
+		});
+
+		expect(ctx.declaredFields).toEqual({
+			keys: ['id', 'createdAt', 'updatedAt', 'contact_email', 'contact_name'],
+			exact: true,
+			source: 'data-table-columns',
+		});
+	});
+
+	it('renders the real columns as the authoritative row shape in the prompt', () => {
+		const [ctx] = buildSchemaContexts(
+			[dataTableNode],
+			() => ({ type: 'object', properties: { id: {} } }),
+			undefined,
+			{ 'Get Rows': [{ name: 'contact_email', type: 'string' }] },
+		);
+		const section = buildNodeSchemaSection(ctx).join('\n');
+
+		expect(section).toContain('REAL Data Table columns');
+		expect(section).toContain('contact_email (string)');
+		// The static `__schema__` (system columns only) is superseded, not embedded.
+		expect(section).not.toContain('Output JSON Schema');
+	});
+});
+
+describe('collectPinFieldViolations', () => {
+	const contexts = [
+		{
+			nodeName: 'Get Rows',
+			nodeType: 'n8n-nodes-base.dataTable',
+			typeVersion: 1,
+			declaredFields: {
+				keys: ['id', 'createdAt', 'updatedAt', 'contact_email'],
+				exact: true,
+				source: 'data-table-columns',
+			},
+		},
+		{
+			nodeName: 'Extract',
+			nodeType: '@n8n/n8n-nodes-langchain.informationExtractor',
+			typeVersion: 1.2,
+			declaredFields: {
+				keys: ['total_amount', 'po_number'],
+				envelopeKey: 'output',
+				exact: false,
+				source: 'declared-schema',
+			},
+		},
+	] satisfies NodeSchemaContext[];
+
+	it('flags renamed and missing keys on exact contracts', () => {
+		const violations = collectPinFieldViolations(
+			{
+				'Get Rows': [{ json: { id: 1, createdAt: 'x', updatedAt: 'x', email: 'a@example.com' } }],
+			},
+			contexts,
+		);
+
+		expect(violations).toEqual([
+			{
+				nodeName: 'Get Rows',
+				unknownKeys: ['email'],
+				missingKeys: ['contact_email'],
+				declaredKeys: ['id', 'createdAt', 'updatedAt', 'contact_email'],
+				envelopeKey: undefined,
+			},
+		]);
+	});
+
+	it('checks fields inside the declared envelope and allows subsets there', () => {
+		const drifted = { Extract: [{ json: { output: { invoice_amount: 5 } } }] };
+		expect(collectPinFieldViolations(drifted, contexts)).toMatchObject([
+			{ nodeName: 'Extract', unknownKeys: ['invoice_amount'], missingKeys: [] },
+		]);
+
+		const subset = { Extract: [{ json: { output: { total_amount: 5 } } }] };
+		expect(collectPinFieldViolations(subset, contexts)).toEqual([]);
+	});
+
+	it.each([
+		['a non-object json payload', 'not a row'],
+		['an array json payload', [] as unknown],
+	])('flags %s under an exact contract as missing every declared key', (_label, json) => {
+		// Skipping malformed items let the pin pass with rows no downstream column
+		// expression can resolve — it must take the correction/failure path instead.
+		expect(collectPinFieldViolations({ 'Get Rows': [{ json }] }, contexts)).toEqual([
+			{
+				nodeName: 'Get Rows',
+				unknownKeys: [],
+				missingKeys: ['id', 'createdAt', 'updatedAt', 'contact_email'],
+				declaredKeys: ['id', 'createdAt', 'updatedAt', 'contact_email'],
+				envelopeKey: undefined,
+			},
+		]);
+	});
+
+	it('leaves malformed items alone when the contract is not exact', () => {
+		// A plain-text agent answer is a legitimate non-object payload.
+		expect(collectPinFieldViolations({ Extract: [{ json: 'plain answer' }] }, contexts)).toEqual(
+			[],
+		);
+	});
+
+	it('accepts empty pins, conforming rows, and nodes without contracts', () => {
+		const pinData = {
+			'Get Rows': [],
+			Extract: [{ json: { output: { total_amount: 5, po_number: 'PO-1' } } }],
+			Unrelated: [{ json: { whatever: true } }],
+		};
+
+		expect(collectPinFieldViolations(pinData, contexts)).toEqual([]);
+	});
+
+	it('builds an actionable retry message', () => {
+		const message = buildFieldViolationRetryMessage([
+			{
+				nodeName: 'Get Rows',
+				unknownKeys: ['email'],
+				missingKeys: ['contact_email'],
+				declaredKeys: ['id', 'contact_email'],
+				envelopeKey: undefined,
+			},
+		]);
+
+		expect(message).toContain('Get Rows');
+		expect(message).toContain('remove/rename these unknown fields: email');
+		expect(message).toContain('every item must also carry: contact_email');
+	});
+});
+
 describe('workflowToMermaid', () => {
 	it('renders nodes with resource/operation labels and connections', () => {
 		const mermaid = workflowToMermaid(workflow);
@@ -1,5 +1,5 @@
 import { findEnvelopeKey } from './ai-root-shapes';
-import { findOutputParserTargets } from './context';
+import { findOutputParserTargets, INFORMATION_EXTRACTOR_NODE_TYPE } from './context';
 import type { NodeSchemaContext, PinData } from './types';
 import type { WorkflowJSON } from '../types/base';

@@ -91,10 +91,25 @@ export function repairStructuredOutput(
 	);
 	const repaired: PinData = { ...pinData };

-	for (const nodeName of findOutputParserTargets(workflow).keys()) {
+	// Information extractors wrap in `{ output: ... }` like parser targets do,
+	// but have no ai_outputParser connection — include them explicitly. Their
+	// envelope is always `output`, even when no `__schema__` resolves; other
+	// roots must declare theirs (chainLlm with a parser emits fields FLAT).
+	const targets = new Set<string>(findOutputParserTargets(workflow).keys());
+	const extractorNames = new Set<string>();
+	for (const node of workflow.nodes) {
+		if (node.name && node.type === INFORMATION_EXTRACTOR_NODE_TYPE) {
+			targets.add(node.name);
+			extractorNames.add(node.name);
+		}
+	}
+
+	for (const nodeName of targets) {
 		const items = repaired[nodeName];
 		if (!items) continue;
-		const envelopeKey = findEnvelopeKey(schemaByName.get(nodeName));
+		const envelopeKey =
+			findEnvelopeKey(schemaByName.get(nodeName)) ??
+			(extractorNames.has(nodeName) ? 'output' : undefined);
 		if (!envelopeKey) continue;

 		repaired[nodeName] = items.map((item) => {
@@ -1,5 +1,5 @@
-import { describeAiRootShape, findEnvelopeKey, isAiRootNodeType } from './ai-root-shapes';
-import { collectDownstreamConsumers } from './context';
+import { describeAiRootShape, isAiRootNodeType } from './ai-root-shapes';
+import { collectDownstreamConsumers, resolveEnvelopeKey } from './context';
 import { workflowToMermaid } from './mermaid';
 import type { NodeSchemaContext, PinDataGenerationInstructions } from './types';
 import type { WorkflowJSON } from '../types/base';
@@ -44,7 +44,7 @@ export function buildNodeSchemaSection(ctx: NodeSchemaContext): string[] {
 			lines.push(`- AI ROOT OUTPUT SHAPE: every item MUST be ${describeAiRootShape(ctx.nodeType)}`);
 		}
 		if (ctx.outputParser?.schemaText) {
-			const envelopeKey = findEnvelopeKey(ctx.schema);
+			const envelopeKey = resolveEnvelopeKey(ctx.nodeType, ctx.schema);
 			const target = envelopeKey ? `The \`${envelopeKey}\` object` : 'The top-level `json` fields';
 			const label = ctx.outputParser.schemaIsExample
 				? `- ${target} must have the same shape and field names as this example:`
@@ -61,6 +61,19 @@ export function buildNodeSchemaSection(ctx: NodeSchemaContext): string[] {
 		}
 	}

+	// Real table columns are authoritative and supersede the static `__schema__`
+	// (which only knows the system columns).
+	if (ctx.dataTableColumns && ctx.dataTableColumns.length > 0) {
+		const columnList = ctx.dataTableColumns.map((c) => `${c.name} (${c.type})`).join(', ');
+		lines.push(
+			'- REAL Data Table columns — every pinned row MUST contain exactly these keys plus ' +
+				'`id` (a NUMBER, auto-incremented: 1, 2, 3…), `createdAt`, `updatedAt` (ISO ' +
+				'timestamps), and no others (values may be empty/null when the scenario calls ' +
+				`for it): ${columnList}`,
+		);
+		return lines;
+	}
+
 	if (ctx.schema) {
 		const schemaStr = JSON.stringify(ctx.schema, null, 2);
 		const truncated =
@@ -31,6 +31,29 @@ export interface OutputParserContext {
 	schemaIsExample: boolean;
 }

+/** A real Data Table column, passed in by consumers with instance access. */
+export interface DataTableColumnInfo {
+	name: string;
+	type: string;
+}
+
+/**
+ * The authoritative field-name contract for a pinned node's items, when one
+ * exists: an information extractor's declared attributes, a structured output
+ * parser's schema keys, or a Data Table's real columns. Generated pin data is
+ * validated against it — drifted names (`invoice_amount` where the schema says
+ * `total_amount`) are the top residual mock-quality defect in eval runs.
+ */
+export interface DeclaredFieldContract {
+	/** The declared field names. */
+	keys: string[];
+	/** Envelope key the fields live under (e.g. `output` for extractor roots); absent = top-level `json`. */
+	envelopeKey?: string;
+	/** True when items must carry exactly `keys` (Data Table rows); false allows a subset (optional schema fields). */
+	exact: boolean;
+	source: 'declared-schema' | 'data-table-columns';
+}
+
 /** Per-node context assembled for the generation prompt. */
 export interface NodeSchemaContext {
 	nodeName: string;
@@ -40,6 +63,9 @@ export interface NodeSchemaContext {
 	operation?: string;
 	schema?: Record<string, unknown>;
 	outputParser?: OutputParserContext;
+	/** Real Data Table columns for dataTable reads — rendered in the prompt as the authoritative row shape. */
+	dataTableColumns?: DataTableColumnInfo[];
+	declaredFields?: DeclaredFieldContract;
 }

 export interface PinDataGenerationInstructions {
@@ -0,0 +1,110 @@
+import type { NodeSchemaContext, PinData } from './types';
+
+/** One pinned node whose item keys deviate from its declared field-name contract. */
+export interface PinFieldViolation {
+	nodeName: string;
+	/** Keys present on pinned items but absent from the declared contract. */
+	unknownKeys: string[];
+	/** Declared keys missing from pinned items — reported only for `exact` contracts. */
+	missingKeys: string[];
+	declaredKeys: string[];
+	envelopeKey?: string;
+}
+
+/**
+ * Compare generated pin data against each node's declared field-name contract
+ * (extractor attributes, parser schema keys, real Data Table columns).
+ * Near-miss renames (`invoice_amount` for `total_amount`, `email` for
+ * `contact_email`) are the dominant residual mock defect in eval runs — they
+ * make correctly-built downstream expressions resolve undefined. Run this
+ * AFTER `repairStructuredOutput` so envelope shape is already canonical.
+ *
+ * Deliberately detect-only: renaming keys here could fabricate
+ * scenario-relevant data and mask real generation defects — callers should
+ * regenerate on violations and fail loud when drift persists.
+ */
+export function collectPinFieldViolations(
+	pinData: PinData,
+	contexts: NodeSchemaContext[],
+): PinFieldViolation[] {
+	const violations: PinFieldViolation[] = [];
+
+	for (const ctx of contexts) {
+		const contract = ctx.declaredFields;
+		if (!contract) continue;
+		const items = pinData[ctx.nodeName];
+		if (!items || items.length === 0) continue; // `[]` is a valid zero-item pin
+
+		const declared = new Set(contract.keys);
+		const unknownKeys = new Set<string>();
+		// Exact contracts (Data Table rows) require every declared key on EVERY
+		// item — real rows always carry every column.
+		const missingKeySet = new Set<string>();
+
+		for (const item of items) {
+			const json = item.json;
+			if (typeof json !== 'object' || json === null || Array.isArray(json)) {
+				// A malformed item (`{"json": "not a row"}`, `{"json": []}`) carries no
+				// field names at all. Under an exact contract that IS the violation —
+				// skipping it let the pin pass validation with rows no downstream
+				// column expression can resolve.
+				if (contract.exact) for (const key of contract.keys) missingKeySet.add(key);
+				continue;
+			}
+			let fields = json as Record<string, unknown>;
+			if (contract.envelopeKey) {
+				const enveloped = fields[contract.envelopeKey];
+				// Non-object envelopes (e.g. a plain-text agent answer) carry no
+				// field names to check.
+				if (typeof enveloped !== 'object' || enveloped === null || Array.isArray(enveloped)) {
+					continue;
+				}
+				fields = enveloped as Record<string, unknown>;
+			}
+			for (const key of Object.keys(fields)) {
+				if (!declared.has(key)) unknownKeys.add(key);
+			}
+			if (contract.exact) {
+				for (const key of contract.keys) {
+					if (!(key in fields)) missingKeySet.add(key);
+				}
+			}
+		}
+
+		const missingKeys = [...missingKeySet];
+
+		if (unknownKeys.size > 0 || missingKeys.length > 0) {
+			violations.push({
+				nodeName: ctx.nodeName,
+				unknownKeys: [...unknownKeys],
+				missingKeys,
+				declaredKeys: contract.keys,
+				envelopeKey: contract.envelopeKey,
+			});
+		}
+	}
+
+	return violations;
+}
+
+/** Corrective follow-up message for a regeneration attempt after field-name drift. */
+export function buildFieldViolationRetryMessage(violations: PinFieldViolation[]): string {
+	const lines = [
+		'Your previous response used field names that do not exist on the nodes below.',
+		'Regenerate the COMPLETE JSON object (every node, same format), keeping all values scenario-consistent, but use EXACTLY the declared field names — do not rename, synonymize, or invent fields.',
+		'Keep the SAME number of items per node as before — do not satisfy a correction by dropping items or returning an empty array unless the scenario itself requires zero items:',
+		'',
+	];
+	for (const v of violations) {
+		const location = v.envelopeKey ? ` (inside the \`${v.envelopeKey}\` object)` : '';
+		lines.push(`- ${v.nodeName}${location}:`);
+		if (v.unknownKeys.length > 0) {
+			lines.push(`  - remove/rename these unknown fields: ${v.unknownKeys.join(', ')}`);
+		}
+		if (v.missingKeys.length > 0) {
+			lines.push(`  - every item must also carry: ${v.missingKeys.join(', ')}`);
+		}
+		lines.push(`  - the ONLY valid field names are: ${v.declaredKeys.join(', ')}`);
+	}
+	return lines.join('\n');
+}
@@ -17,7 +17,9 @@ import type { ActiveExecutions } from '@/active-executions';
 import type { LoadNodesAndCredentials } from '@/load-nodes-and-credentials';
 import type { NodeTypes } from '@/node-types';
 import type { PostHogClient } from '@/posthog';
+import type { DataTableService } from '@/modules/data-table/data-table.service';
 import type { WorkflowRunner } from '@/workflow-runner';
+import type { OwnershipService } from '@/services/ownership.service';
 import type { WorkflowFinderService } from '@/workflows/workflow-finder.service';
 import type { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';

@@ -44,6 +46,8 @@ vi.mock('../workflow-analysis', () => ({
 	generateMockHints: vi.fn(),
 	identifyNodesForHints: vi.fn(),
 	identifyNodesForPinData: vi.fn(),
+	isDataTableRead: vi.fn().mockReturnValue(false),
+	emitsDataTableRows: vi.fn().mockReturnValue(false),
 	detectBinaryDependencies: vi.fn(),
 }));

@@ -99,6 +103,7 @@ import { createLlmMockHandler } from '../mock-handler';
 import { generatePinData } from '../pin-data-generator';
 import {
 	detectBinaryDependencies,
+	emitsDataTableRows,
 	generateMockHints,
 	identifyNodesForHints,
 	identifyNodesForPinData,
@@ -114,6 +119,7 @@ const generateMockHintsMock = vi.mocked(generateMockHints);
 const detectBinaryDependenciesMock = vi.mocked(detectBinaryDependencies);
 const identifyNodesForHintsMock = vi.mocked(identifyNodesForHints);
 const identifyNodesForPinDataMock = vi.mocked(identifyNodesForPinData);
+const emitsDataTableRowsMock = vi.mocked(emitsDataTableRows);
 const partitionAiRootsMock = vi.mocked(partitionAiRoots);
 const createLlmMockHandlerMock = vi.mocked(createLlmMockHandler);
 const generatePinDataMock = vi.mocked(generatePinData);
@@ -209,6 +215,8 @@ describe('EvalExecutionService', () => {
 	const binaryDataService = mock<BinaryDataService>();
 	const workflowStaticDataService = mock<WorkflowStaticDataService>();
 	const loadNodesAndCredentials = mock<LoadNodesAndCredentials>();
+	const ownershipService = mock<OwnershipService>();
+	const dataTableService = mock<DataTableService>();

 	// Captured configureAdditionalData closure so tests can re-invoke it on a
 	// stub additionalData without booting the real runner.
@@ -243,6 +251,8 @@ describe('EvalExecutionService', () => {
 			binaryDataService,
 			workflowStaticDataService,
 			loadNodesAndCredentials,
+			ownershipService,
+			dataTableService,
 		);
 		// Reset to safe default — tests that flip queue mode reassign in-test.
 		Object.assign(executionsConfig, { mode: 'regular' });
@@ -1478,6 +1488,100 @@ describe('EvalExecutionService', () => {
 		});
 	});

+	// ── Data Table column contracts ──────────────────────────────────
+
+	describe('resolveDataTableColumns (via execution)', () => {
+		function makeDataTableNode(dataTableId: unknown, operation = 'get'): INode {
+			return {
+				id: 'node-dt',
+				name: 'Get Rows',
+				type: 'n8n-nodes-base.dataTable',
+				typeVersion: 1,
+				position: [200, 0],
+				parameters: { resource: 'row', operation, dataTableId },
+			} as INode;
+		}
+
+		function makeDataTableWorkflow(dataTableId: unknown, operation = 'get') {
+			const node = makeDataTableNode(dataTableId, operation);
+			// The SUT maps these to names, so the mock must yield node objects.
+			identifyNodesForPinDataMock.mockReturnValue([node]);
+			return makeWorkflowEntity({ nodes: [makeStartNode(), node] });
+		}
+
+		beforeEach(() => {
+			// Mirrors the real predicate: only `get` emits stored rows.
+			emitsDataTableRowsMock.mockImplementation(
+				(node: INode) =>
+					node.type === 'n8n-nodes-base.dataTable' &&
+					(node.parameters as { operation?: string } | undefined)?.operation === 'get',
+			);
+			ownershipService.getWorkflowProjectCached.mockResolvedValue({ id: 'proj-1' } as never);
+			dataTableService.getColumns.mockResolvedValue([
+				{ name: 'contact_email', type: 'string' },
+			] as never);
+		});
+
+		it('passes an id-mode locator straight through to the column lookup', async () => {
+			workflowFinderService.findWorkflowForUser.mockResolvedValue(
+				makeDataTableWorkflow({ __rl: true, mode: 'id', value: 'dt-42' }) as never,
+			);
+
+			await service.executeWithLlmMock('wf-1', makeUser());
+
+			expect(dataTableService.getColumns).toHaveBeenCalledWith('dt-42', 'proj-1');
+			expect(generatePinDataMock.mock.calls[0][0].dataTableColumns).toEqual({
+				'Get Rows': [{ name: 'contact_email', type: 'string' }],
+			});
+		});
+
+		it('resolves a name-mode locator to its id before fetching columns', async () => {
+			workflowFinderService.findWorkflowForUser.mockResolvedValue(
+				makeDataTableWorkflow({ __rl: true, mode: 'name', value: 'Customers' }) as never,
+			);
+			dataTableService.findDataTablesByNamesInProject.mockResolvedValue([
+				{ id: 'dt-9', name: 'Customers' },
+			]);
+
+			await service.executeWithLlmMock('wf-1', makeUser());
+
+			// A name passed to the id lookup used to miss, silently dropping the node
+			// to prompt-only generation with invented column names.
+			expect(dataTableService.findDataTablesByNamesInProject).toHaveBeenCalledWith('proj-1', [
+				'Customers',
+			]);
+			expect(dataTableService.getColumns).toHaveBeenCalledWith('dt-9', 'proj-1');
+		});
+
+		it.each(['rowExists', 'rowNotExists'])(
+			'skips the column contract for %s, which emits the input item not table rows',
+			async (operation) => {
+				workflowFinderService.findWorkflowForUser.mockResolvedValue(
+					makeDataTableWorkflow({ __rl: true, mode: 'id', value: 'dt-42' }, operation) as never,
+				);
+
+				await service.executeWithLlmMock('wf-1', makeUser());
+
+				// Enforcing table columns here would demand a fixture the real node
+				// never emits, then blame the resulting mismatch on the builder.
+				expect(dataTableService.getColumns).not.toHaveBeenCalled();
+				expect(generatePinDataMock.mock.calls[0][0].dataTableColumns).toBeUndefined();
+			},
+		);
+
+		it('degrades to prompt-only generation when no table matches the name', async () => {
+			workflowFinderService.findWorkflowForUser.mockResolvedValue(
+				makeDataTableWorkflow({ __rl: true, mode: 'name', value: 'Missing' }) as never,
+			);
+			dataTableService.findDataTablesByNamesInProject.mockResolvedValue([]);
+
+			await service.executeWithLlmMock('wf-1', makeUser());
+
+			expect(dataTableService.getColumns).not.toHaveBeenCalled();
+			expect(generatePinDataMock.mock.calls[0][0].dataTableColumns).toBeUndefined();
+		});
+	});
+
 	// ── hints passthrough ────────────────────────────────────────────

 	describe('hints in result', () => {
@@ -106,4 +106,59 @@ describe('generatePinData', () => {
 		expect(result).toEqual({});
 		expect(createEvalAgentMock).not.toHaveBeenCalled();
 	});
+
+	describe('field-name drift', () => {
+		const dataTableColumns = {
+			'Get Posted Keys': [{ name: 'contact_email', type: 'string' }],
+		};
+		const conforming = JSON.stringify({
+			'Get Posted Keys': [
+				{ json: { id: 1, createdAt: 'x', updatedAt: 'x', contact_email: 'a@b.c' } },
+			],
+		});
+		const drifted = JSON.stringify({
+			'Get Posted Keys': [{ json: { id: 1, createdAt: 'x', updatedAt: 'x', email: 'a@b.c' } }],
+		});
+
+		it('embeds the real columns in the prompt and accepts conforming rows first try', async () => {
+			respondWith(conforming);
+
+			const result = await generatePinData({
+				workflow,
+				nodeNames: ['Get Posted Keys'],
+				dataTableColumns,
+			});
+
+			expect(result['Get Posted Keys'][0].json).toMatchObject({ contact_email: 'a@b.c' });
+			expect(generateMock).toHaveBeenCalledTimes(1);
+			expect(generateMock.mock.calls[0][0]).toContain('REAL Data Table columns');
+		});
+
+		it('regenerates once with corrections when pinned keys drift', async () => {
+			generateMock.mockResolvedValue({});
+			extractTextMock.mockReturnValueOnce(drifted).mockReturnValueOnce(conforming);
+
+			const result = await generatePinData({
+				workflow,
+				nodeNames: ['Get Posted Keys'],
+				dataTableColumns,
+			});
+
+			expect(result['Get Posted Keys'][0].json).toMatchObject({ contact_email: 'a@b.c' });
+			expect(generateMock).toHaveBeenCalledTimes(2);
+			const retryPrompt = generateMock.mock.calls[1][0] as string;
+			expect(retryPrompt).toContain('## Correction required');
+			expect(retryPrompt).toContain('remove/rename these unknown fields: email');
+		});
+
+		it('fails loud instead of serving a still-drifted fixture after the retry', async () => {
+			generateMock.mockResolvedValue({});
+			extractTextMock.mockReturnValue(drifted);
+
+			await expect(
+				generatePinData({ workflow, nodeNames: ['Get Posted Keys'], dataTableColumns }),
+			).rejects.toThrow('drifted from declared field names after retry');
+			expect(generateMock).toHaveBeenCalledTimes(2);
+		});
+	});
 });
@@ -14,9 +14,11 @@ import { UserError } from 'n8n-workflow';
 import {
 	buildVendorLlmRouting,
 	detectBinaryDependencies,
+	emitsDataTableRows,
 	generateMockHints,
 	identifyNodesForHints,
 	identifyNodesForPinData,
+	isDataTableRead,
 	partitionAiRoots,
 } from '../workflow-analysis';

@@ -47,6 +49,35 @@ function makeWorkflow(nodes: INode[], connections: IConnections = {}): IWorkflow
 	};
 }

+describe('Data Table read predicates', () => {
+	function makeDataTableNode(parameters: INodeParameters): INode {
+		return makeNode({ name: 'Table', type: 'n8n-nodes-base.dataTable', parameters });
+	}
+
+	it.each(['get', 'rowExists', 'rowNotExists'])('treats %s as a read', (operation) => {
+		expect(isDataTableRead(makeDataTableNode({ resource: 'row', operation }))).toBe(true);
+	});
+
+	it.each(['insert', 'update', 'deleteRows'])('treats %s as a write', (operation) => {
+		expect(isDataTableRead(makeDataTableNode({ resource: 'row', operation }))).toBe(false);
+	});
+
+	it('only counts `get` as row-emitting', () => {
+		// rowExists/rowNotExists return `[this.getInputData()[index]]` — the input
+		// item passed through — so the table's column contract does not apply.
+		expect(emitsDataTableRows(makeDataTableNode({ resource: 'row', operation: 'get' }))).toBe(true);
+		for (const operation of ['rowExists', 'rowNotExists', 'insert']) {
+			expect(emitsDataTableRows(makeDataTableNode({ resource: 'row', operation }))).toBe(false);
+		}
+	});
+
+	it('ignores non-Data-Table nodes', () => {
+		const node = makeNode({ name: 'HTTP', type: 'n8n-nodes-base.httpRequest' });
+		expect(isDataTableRead(node)).toBe(false);
+		expect(emitsDataTableRows(node)).toBe(false);
+	});
+});
+
 describe('identifyNodesForPinData', () => {
 	it('should identify AI root nodes as needing pin data', () => {
 		const nodes = [
@@ -9,7 +9,7 @@ import { ensureHostsBypassProxy } from '@n8n/backend-network/proxy';
 import { ExecutionsConfig } from '@n8n/config';
 import type { User } from '@n8n/db';
 import { Service } from '@n8n/di';
-import type { WorkflowJSON } from '@n8n/workflow-sdk';
+import type { DataTableColumnInfo, WorkflowJSON } from '@n8n/workflow-sdk';
 import { normalizePinData } from '@n8n/workflow-sdk';
 import {
 	BinaryDataService,
@@ -41,8 +41,10 @@ import { randomUUID } from 'node:crypto';

 import { ActiveExecutions } from '@/active-executions';
 import { LoadNodesAndCredentials } from '@/load-nodes-and-credentials';
+import { DataTableService } from '@/modules/data-table/data-table.service';
 import { NodeTypes } from '@/node-types';
 import { PostHogClient } from '@/posthog';
+import { OwnershipService } from '@/services/ownership.service';
 import { WorkflowRunner } from '@/workflow-runner';
 import { WorkflowFinderService } from '@/workflows/workflow-finder.service';
 import { WorkflowStaticDataService } from '@/workflows/workflow-static-data.service';
@@ -61,6 +63,7 @@ import { generatePinData } from './pin-data-generator';
 import {
 	buildVendorLlmRouting,
 	detectBinaryDependencies,
+	emitsDataTableRows,
 	generateMockHints,
 	identifyNodesForHints,
 	identifyNodesForPinData,
@@ -97,6 +100,8 @@ export class EvalExecutionService {
 		private readonly binaryDataService: BinaryDataService,
 		private readonly workflowStaticDataService: WorkflowStaticDataService,
 		private readonly loadNodesAndCredentials: LoadNodesAndCredentials,
+		private readonly ownershipService: OwnershipService,
+		private readonly dataTableService: DataTableService,
 	) {}

 	async executeWithLlmMock(
@@ -283,6 +288,8 @@ export class EvalExecutionService {
 		if (bypassNodeNames.length === 0) return {};

 		try {
+			const dataTableColumns = await this.resolveDataTableColumns(workflowEntity, bypassNodeNames);
+
 			// Keep the scenario separate from the general context: the pin generator
 			// treats "Test Scenario" as authoritative, and merging them into one blob
 			// lets invented context override scenario-specified stored state.
@@ -298,6 +305,7 @@ export class EvalExecutionService {
 								? { dataDescription: globalContext, testScenario: scenarioHints }
 								: undefined,
 						outputSchemaLookup: this.loadNodesAndCredentials.createOutputSchemaLookup(),
+						dataTableColumns,
 					}),
 			);

@@ -326,6 +334,73 @@ export class EvalExecutionService {
 		}
 	}

+	/**
+	 * Real column names for each pinned dataTable-read node, read from the
+	 * builder-created table itself. They are the authoritative row shape —
+	 * without them the pin generator invents plausible-but-wrong column names
+	 * (`email` where the table says `contact_email`) and correctly-built
+	 * downstream expressions resolve undefined. Best-effort: a missing table or
+	 * unresolved id degrades that node to prompt-only generation.
+	 *
+	 * Only row-emitting reads qualify — `rowExists`/`rowNotExists` pass the input
+	 * item through, so enforcing table columns on them would demand a fixture the
+	 * real node never emits.
+	 */
+	private async resolveDataTableColumns(
+		workflowEntity: IWorkflowBase,
+		bypassNodeNames: string[],
+	): Promise<Record<string, DataTableColumnInfo[]> | undefined> {
+		const bypassSet = new Set(bypassNodeNames);
+		const readNodes = workflowEntity.nodes.filter(
+			(node) => bypassSet.has(node.name) && emitsDataTableRows(node),
+		);
+		if (readNodes.length === 0) return undefined;
+
+		const columnsByNode: Record<string, DataTableColumnInfo[]> = {};
+		let projectId: string | undefined;
+		for (const node of readNodes) {
+			try {
+				const locator = node.parameters?.dataTableId as
+					| { mode?: unknown; value?: unknown }
+					| string
+					| undefined;
+				const locatorValue = typeof locator === 'string' ? locator : locator?.value;
+				if (typeof locatorValue !== 'string' || locatorValue.length === 0) continue;
+
+				projectId ??= (await this.ownershipService.getWorkflowProjectCached(workflowEntity.id)).id;
+
+				// `name` mode carries a table name, not an id (the node runtime resolves
+				// it via `resolveDataTableId`) — passing it straight to an id lookup
+				// dropped named tables to prompt-only generation. Exact name match only;
+				// a near-miss still degrades gracefully below.
+				let tableId = locatorValue;
+				if ((typeof locator === 'string' ? 'id' : locator?.mode) === 'name') {
+					const matches = await this.dataTableService.findDataTablesByNamesInProject(projectId, [
+						locatorValue,
+					]);
+					const resolved = matches.at(0)?.id;
+					if (!resolved) {
+						this.logger.warn(
+							`[EvalMock] No Data Table named "${locatorValue}" for node "${node.name}" — pinned rows fall back to prompt-only generation`,
+						);
+						continue;
+					}
+					tableId = resolved;
+				}
+
+				const columns = await this.dataTableService.getColumns(tableId, projectId);
+				columnsByNode[node.name] = columns.map(({ name, type }) => ({ name, type }));
+			} catch (error) {
+				this.logger.warn(
+					`[EvalMock] Could not resolve Data Table columns for node "${node.name}" — pinned rows fall back to prompt-only generation`,
+					{ error: error instanceof Error ? error.message : String(error) },
+				);
+			}
+		}
+
+		return Object.keys(columnsByNode).length > 0 ? columnsByNode : undefined;
+	}
+
 	// ── Phase 2: Mock execution ────────────────────────────────────────────

 	private async execute(
@@ -16,11 +16,14 @@ import type {
 	WorkflowJSON,
 	OutputSchemaLookup,
 	PinDataGenerationInstructions,
+	DataTableColumnInfo,
 } from '@n8n/workflow-sdk';
 import {
 	buildDateAnchors,
+	buildFieldViolationRetryMessage,
 	buildPinDataUserPrompt,
 	buildSchemaContexts,
+	collectPinFieldViolations,
 	findOutputParserTargets,
 	parsePinDataResponse,
 	PIN_DATA_SYSTEM_PROMPT,
@@ -52,6 +55,11 @@ export interface GeneratePinDataOptions {
 	 * Absent lookup degrades to API-knowledge-only generation.
 	 */
 	outputSchemaLookup?: OutputSchemaLookup;
+	/**
+	 * Real Data Table columns per pinned dataTable-read node name — the
+	 * authoritative row shape; pinned rows are validated against these keys.
+	 */
+	dataTableColumns?: Record<string, DataTableColumnInfo[]>;
 }

 /**
@@ -65,7 +73,7 @@ export interface GeneratePinDataOptions {
 * @throws when generation fails or a node is missing — a silently unpinned node runs for real.
 */
 export async function generatePinData(options: GeneratePinDataOptions): Promise<PinData> {
-	const { workflow, nodeNames, instructions, outputSchemaLookup } = options;
+	const { workflow, nodeNames, instructions, outputSchemaLookup, dataTableColumns } = options;

 	if (nodeNames.length === 0) return {};

@@ -76,7 +84,12 @@ export async function generatePinData(options: GeneratePinDataOptions): Promise<
 	// Build schema contexts with optional __schema__ enrichment and
 	// structured-output-parser envelopes for AI roots
 	const outputParserTargets = findOutputParserTargets(workflow);
-	const contexts = buildSchemaContexts(targetNodes, outputSchemaLookup, outputParserTargets);
+	const contexts = buildSchemaContexts(
+		targetNodes,
+		outputSchemaLookup,
+		outputParserTargets,
+		dataTableColumns,
+	);

 	// Build prompt and call LLM
 	const userPrompt = buildPinDataUserPrompt(workflow, contexts, {
@@ -90,23 +103,54 @@ export async function generatePinData(options: GeneratePinDataOptions): Promise<
 		cache: true,
 	});

-	const result = await agent.generate(userPrompt, {
-		providerOptions: { anthropic: { maxTokens: 16_384 } },
-		abortSignal: AbortSignal.timeout(PIN_DATA_LLM_TIMEOUT_MS),
-	});
+	const generateOnce = async (prompt: string): Promise<PinData> => {
+		const result = await agent.generate(prompt, {
+			providerOptions: { anthropic: { maxTokens: 16_384 } },
+			abortSignal: AbortSignal.timeout(PIN_DATA_LLM_TIMEOUT_MS),
+		});

-	const responseText = extractText(result);
-	const pinData = parsePinDataResponse(responseText, expectedNodeNames);
+		const responseText = extractText(result);
+		const pinData = parsePinDataResponse(responseText, expectedNodeNames);

-	const missing = expectedNodeNames.filter((name) => !(name in pinData));
-	if (missing.length > 0) {
+		const missing = expectedNodeNames.filter((name) => !(name in pinData));
+		if (missing.length > 0) {
+			throw new OperationalError(
+				`Pin data generation returned no data for node(s): ${missing.join(', ')}`,
+			);
+		}
+
+		// Envelope repair for parser-target roots; the shared helper derives the
+		// envelope key from each root's with-parser `__schema__` variant
+		// (`output` for agent and chainLlm ≥1.9, always `output` for extractors).
+		return repairStructuredOutput(pinData, workflow, contexts);
+	};
+
+	const pinData = await generateOnce(userPrompt);
+
+	// Field-name drift (e.g. `invoice_amount` where the declared schema says
+	// `total_amount`) silently breaks correctly-built downstream expressions.
+	// Regenerate once with explicit corrections rather than renaming keys in
+	// place — a deterministic rename could fabricate scenario-relevant data.
+	const violations = collectPinFieldViolations(pinData, contexts);
+	if (violations.length === 0) return pinData;
+
+	const retryPrompt = `${userPrompt}\n\n## Correction required\n\n${buildFieldViolationRetryMessage(violations)}`;
+	const retried = await generateOnce(retryPrompt);
+
+	const remaining = collectPinFieldViolations(retried, contexts);
+	if (remaining.length > 0) {
+		const summary = remaining
+			.map(
+				(v) =>
+					`${v.nodeName} (unknown: ${v.unknownKeys.join(', ') || '-'}; missing: ${v.missingKeys.join(', ') || '-'}; declared: ${v.declaredKeys.join(', ')})`,
+			)
+			.join('; ');
+		// Fail loud: a drifted fixture served silently would poison failure
+		// attribution — an unpinnable scenario must surface as a harness fault.
 		throw new OperationalError(
-			`Pin data generation returned no data for node(s): ${missing.join(', ')}`,
+			`Pin data generation drifted from declared field names after retry: ${summary}`,
 		);
 	}

-	// Envelope repair for parser-target roots; the shared helper derives the
-	// envelope key from each root's with-parser `__schema__` variant
-	// (`output` for both agent and chainLlm).
-	return repairStructuredOutput(pinData, workflow, contexts);
+	return retried;
 }
@@ -94,7 +94,12 @@ const PROTOCOL_BINARY_SUB_NODE_TYPES = new Set([
 * verification runs, so scenario outcomes become a coin flip on build-phase leftovers. */
 const DATA_TABLE_READ_OPERATIONS = new Set(['get', 'rowExists', 'rowNotExists']);

-function isDataTableRead(node: INode): boolean {
+/** Of the read operations, only `get` emits stored rows — `rowExists`/`rowNotExists`
+ *  return the input item passed straight through, so the table's column contract
+ *  does not describe their output. */
+const DATA_TABLE_ROW_EMITTING_OPERATIONS = new Set(['get']);
+
+export function isDataTableRead(node: INode): boolean {
 	if (node.type !== 'n8n-nodes-base.dataTable') return false;
 	const params = node.parameters as { resource?: string; operation?: string } | undefined;
 	// Node defaults: resource 'row', operation 'insert' (a write) — only pin explicit reads.
@@ -104,6 +109,15 @@ function isDataTableRead(node: INode): boolean {
 	);
 }

+/** True for Data Table reads whose output IS stored rows — the only reads a real
+ *  column contract applies to. Still pinned like any other read; they just get
+ *  prompt-only generation instead of enforced column names. */
+export function emitsDataTableRows(node: INode): boolean {
+	if (!isDataTableRead(node)) return false;
+	const params = node.parameters as { operation?: string } | undefined;
+	return DATA_TABLE_ROW_EMITTING_OPERATIONS.has(params?.operation ?? 'insert');
+}
+
 /** Returns nodes that need pin data — AI roots (unless in `exclusionSet`), bypass-protocol nodes, and Data Table reads. */
 export function identifyNodesForPinData(
 	workflow: IWorkflowBase,