github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-07-29 03:55:15 +02:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity Graph

refactor(ai-builder): Reduce code complexity below maintainability thresholds (no-changelog) (#32626)

Browse Source
This commit is contained in:
Albert Alises 2026-06-19 14:22:04 +02:00 committed by GitHub
parent 30b2127731
commit 21994c465f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 2060 additions and 1468 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

415
packages/@n8n/instance-ai/src/agent/sanitize-mcp-schemas.ts
View File

@ -266,6 +266,23 @@ function createUnsupportedTypeError(
});
}
function isUnsupportedZodType(schema: z.ZodTypeAny): boolean {
return (
schema instanceof z.ZodMap ||
schema instanceof z.ZodSet ||
schema instanceof z.ZodPromise ||
schema instanceof z.ZodFunction ||
schema instanceof z.ZodIntersection ||
schema instanceof z.ZodTuple ||
schema instanceof z.ZodNaN ||
schema instanceof z.ZodBigInt ||
schema instanceof z.ZodUndefined ||
schema instanceof z.ZodNever ||
schema instanceof z.ZodVoid ||
schema instanceof z.ZodSymbol
);
}
function isSupportedLeafSchema(schema: z.ZodTypeAny): boolean {
return (
schema instanceof z.ZodString ||
@ -280,6 +297,212 @@ function isSupportedLeafSchema(schema: z.ZodTypeAny): boolean {
);
}
type SanitizeChild = (child: z.ZodTypeAny, path: string) => z.ZodTypeAny;
type DiscriminatedFieldEntry = { action: string; description?: string; type: z.ZodTypeAny };
/** Collect per-action and per-field metadata from a discriminated union's variants. */
function collectDiscriminatedUnionMeta(
variants: Array<z.ZodObject<z.ZodRawShape>>,
discriminator: string,
): {
actionMeta: Array<{ value: string; description?: string }>;
fieldMeta: Map<string, DiscriminatedFieldEntry[]>;
} {
const actionMeta: Array<{ value: string; description?: string }> = [];
const fieldMeta = new Map<string, DiscriminatedFieldEntry[]>();
for (const variant of variants) {
let actionValue = '';
for (const [key, value] of Object.entries(variant.shape)) {
if (key === discriminator && value instanceof z.ZodLiteral) {
actionValue = String(value.value);
actionMeta.push({ value: actionValue, description: value.description });
}
}
for (const [key, value] of Object.entries(variant.shape)) {
if (key === discriminator) continue;
if (!fieldMeta.has(key)) fieldMeta.set(key, []);
fieldMeta
.get(key)!
.push({ action: actionValue, description: value.description, type: value });
}
}
return { actionMeta, fieldMeta };
}
/**
* Detect enum value conflicts across variants sharing a field. Only the first
* variant's type is kept, so differing enum values elsewhere would be silently lost.
*/
function assertNoEnumConflict(fieldName: string, entries: DiscriminatedFieldEntry[]): void {
const unwrapOptional = (t: z.ZodTypeAny): z.ZodTypeAny =>
t instanceof z.ZodOptional ? unwrapOptional(t.unwrap() as z.ZodTypeAny) : t;
const enumEntries = entries.filter((e) => unwrapOptional(e.type) instanceof z.ZodEnum);
if (enumEntries.length <= 1) return;
const valueSets = enumEntries.map((e) => {
const raw = unwrapOptional(e.type);
return (raw as z.ZodEnum<[string, ...string[]]>).options.slice().sort().join(',');
});
if (new Set(valueSets).size <= 1) return;
const conflictDetails = enumEntries
.map((e) => {
const raw = unwrapOptional(e.type);
const vals = (raw as z.ZodEnum<[string, ...string[]]>).options;
return ` Action "${e.action}": [${vals.join(', ')}]`;
})
.join('\n');
throw new Error(
`Enum conflict for field "${fieldName}" in discriminated union:\n` +
`${conflictDetails}\n` +
'Harmonize enum values across all actions that share this field.',
);
}
/** Build a single flattened, optional field for a merged discriminated union, merging descriptions. */
function buildMergedDiscriminatedField(
fieldName: string,
entries: DiscriminatedFieldEntry[],
actionCount: number,
context: SanitizeContext,
sanitizeChild: SanitizeChild,
): z.ZodTypeAny {
const sanitizedField = sanitizeChild(entries[0].type, `${context.path}.${fieldName}`).optional();
if (context.strict && entries.length > 1) {
assertNoEnumConflict(fieldName, entries);
}
const withDesc = entries.filter((e): e is typeof e & { description: string } => !!e.description);
const uniqueDescs = new Set(withDesc.map((d) => d.description));
if (uniqueDescs.size > 1) {
if (context.strict) {
const conflictDetails = withDesc
.map((d) => ` Action "${d.action}": "${d.description}"`)
.join('\n');
throw new Error(
`Description conflict for field "${fieldName}" in discriminated union:\n` +
`${conflictDetails}\n` +
'Harmonize to a single description across all actions that share this field.',
);
}
// Non-strict: combine with action context for external MCP tools
const combined = withDesc.map((d) => `For "${d.action}": ${d.description}`).join('. ');
return sanitizedField.describe(combined);
}
if (entries.length < actionCount) {
// Field appears in a subset of variants. Annotate with an action hint so
// the model binds the (now-flattened-optional) field to the right actions
// and stops cross-mixing fields between sibling actions.
const actionList = entries.map((e) => `"${e.action}"`).join(', ');
const baseDesc = withDesc[0]?.description;
const merged = baseDesc ? `For ${actionList}: ${baseDesc}` : `Only for ${actionList}`;
return sanitizedField.describe(merged);
}
return sanitizedField;
}
/**
* Flatten a ZodDiscriminatedUnion to a single z.object: the discriminator becomes
* an enum with per-action descriptions, and variant-specific fields become optional
* with merged descriptions. Anthropic rejects top-level unions (no type=object).
*/
function sanitizeDiscriminatedUnion(
schema: z.ZodDiscriminatedUnion<string, Array<z.ZodObject<z.ZodRawShape>>>,
context: SanitizeContext,
sanitizeChild: SanitizeChild,
): z.ZodTypeAny {
const discriminator = schema.discriminator;
const variants = [...schema.options.values()] as Array<z.ZodObject<z.ZodRawShape>>;
if (variants.length > context.maxUnionOptions) {
throw createLimitError(
context,
`MCP schema discriminated union exceeds maximum option count of ${context.maxUnionOptions}`,
'unionOptions',
context.maxUnionOptions,
variants.length,
);
}
const { actionMeta, fieldMeta } = collectDiscriminatedUnionMeta(variants, discriminator);
const mergedPropertyCount = fieldMeta.size + (actionMeta.length > 0 ? 1 : 0);
if (mergedPropertyCount > context.maxObjectProperties) {
throw createLimitError(
context,
`MCP schema object exceeds maximum property count of ${context.maxObjectProperties}`,
'objectProperties',
context.maxObjectProperties,
mergedPropertyCount,
);
}
const mergedShape: z.ZodRawShape = {};
// Build discriminator enum with per-action descriptions
if (actionMeta.length > 0) {
const enumValues = actionMeta.map((a) => a.value);
const actionDescParts = actionMeta.map((a) =>
a.description ? `"${a.value}": ${a.description}` : `"${a.value}"`,
);
mergedShape[discriminator] = z
.enum(enumValues as [string, ...string[]])
.describe(actionDescParts.join(' | '));
}
for (const [fieldName, entries] of fieldMeta) {
mergedShape[fieldName] = buildMergedDiscriminatedField(
fieldName,
entries,
actionMeta.length,
context,
sanitizeChild,
);
}
return z.object(mergedShape);
}
/** Strip ZodNull members from a union, making the result optional if null was present. */
function sanitizeUnion(
schema: z.ZodUnion<[z.ZodTypeAny, ...z.ZodTypeAny[]]>,
context: SanitizeContext,
sanitizeChild: SanitizeChild,
): z.ZodTypeAny {
const options = schema.options as z.ZodTypeAny[];
if (options.length > context.maxUnionOptions) {
throw createLimitError(
context,
`MCP schema union exceeds maximum option count of ${context.maxUnionOptions}`,
'unionOptions',
context.maxUnionOptions,
options.length,
);
}
const nonNull = options.filter((o) => !(o instanceof z.ZodNull));
const hadNull = nonNull.length < options.length;
const sanitized = nonNull.map((o, index) => sanitizeChild(o, `${context.path}.union[${index}]`));
if (sanitized.length === 0) {
// All options were null — degenerate case
return z.string().optional();
}
if (sanitized.length === 1) {
return hadNull ? sanitized[0].optional() : sanitized[0];
}
const union = z.union(sanitized as [z.ZodTypeAny, z.ZodTypeAny, ...z.ZodTypeAny[]]);
return hadNull ? union.optional() : union;
}
function sanitizeZodTypeInner(schema: z.ZodTypeAny, context: SanitizeContext): z.ZodTypeAny {
if (context.depth > context.maxDepth) {
throw createLimitError(
@ -321,179 +544,22 @@ function sanitizeZodTypeInner(schema: z.ZodTypeAny, context: SanitizeContext): z
).optional();
}
// ZodDiscriminatedUnion — flatten to a single z.object
// (discriminator becomes an enum with per-action descriptions,
// variant-specific fields become optional with merged descriptions).
// Anthropic rejects top-level unions because they produce schemas without type=object.
// ZodDiscriminatedUnion — flatten to a single z.object (see helper).
if (schema instanceof z.ZodDiscriminatedUnion) {
const disc = schema as z.ZodDiscriminatedUnion<string, Array<z.ZodObject<z.ZodRawShape>>>;
const discriminator = disc.discriminator;
const variants = [...disc.options.values()] as Array<z.ZodObject<z.ZodRawShape>>;
if (variants.length > context.maxUnionOptions) {
throw createLimitError(
context,
`MCP schema discriminated union exceeds maximum option count of ${context.maxUnionOptions}`,
'unionOptions',
context.maxUnionOptions,
variants.length,
);
}
// Phase 1: Collect metadata from all variants
const actionMeta: Array<{ value: string; description?: string }> = [];
const fieldMeta = new Map<
string,
Array<{ action: string; description?: string; type: z.ZodTypeAny }>
>();
for (const variant of variants) {
let actionValue = '';
for (const [key, value] of Object.entries(variant.shape)) {
if (key === discriminator && value instanceof z.ZodLiteral) {
actionValue = String(value.value);
actionMeta.push({ value: actionValue, description: value.description });
}
}
for (const [key, value] of Object.entries(variant.shape)) {
if (key === discriminator) continue;
if (!fieldMeta.has(key)) fieldMeta.set(key, []);
fieldMeta.get(key)!.push({
action: actionValue,
description: value.description,
type: value,
});
}
}
const mergedPropertyCount = fieldMeta.size + (actionMeta.length > 0 ? 1 : 0);
if (mergedPropertyCount > context.maxObjectProperties) {
throw createLimitError(
context,
`MCP schema object exceeds maximum property count of ${context.maxObjectProperties}`,
'objectProperties',
context.maxObjectProperties,
mergedPropertyCount,
);
}
// Phase 2: Build the merged shape
const mergedShape: z.ZodRawShape = {};
// Build discriminator enum with per-action descriptions
if (actionMeta.length > 0) {
const enumValues = actionMeta.map((a) => a.value);
const actionDescParts = actionMeta.map((a) =>
a.description ? `"${a.value}": ${a.description}` : `"${a.value}"`,
);
mergedShape[discriminator] = z
.enum(enumValues as [string, ...string[]])
.describe(actionDescParts.join(' | '));
}
// Build each field with properly merged descriptions
for (const [fieldName, entries] of fieldMeta) {
const sanitizedField = sanitizeChild(
entries[0].type,
`${context.path}.${fieldName}`,
).optional();
// Detect enum value conflicts across variants.
// Only the first variant's type is used (entries[0].type), so differing
// enum values in other variants would be silently lost.
if (context.strict && entries.length > 1) {
const unwrapOptional = (t: z.ZodTypeAny): z.ZodTypeAny =>
t instanceof z.ZodOptional ? unwrapOptional(t.unwrap() as z.ZodTypeAny) : t;
const enumEntries = entries.filter((e) => {
return unwrapOptional(e.type) instanceof z.ZodEnum;
});
if (enumEntries.length > 1) {
const valueSets = enumEntries.map((e) => {
const raw = unwrapOptional(e.type);
return (raw as z.ZodEnum<[string, ...string[]]>).options.slice().sort().join(',');
});
const uniqueValues = new Set(valueSets);
if (uniqueValues.size > 1) {
const conflictDetails = enumEntries
.map((e) => {
const raw = unwrapOptional(e.type);
const vals = (raw as z.ZodEnum<[string, ...string[]]>).options;
return ` Action "${e.action}": [${vals.join(', ')}]`;
})
.join('\n');
throw new Error(
`Enum conflict for field "${fieldName}" in discriminated union:\n` +
`${conflictDetails}\n` +
'Harmonize enum values across all actions that share this field.',
);
}
}
}
const withDesc = entries.filter(
(e): e is typeof e & { description: string } => !!e.description,
);
const uniqueDescs = new Set(withDesc.map((d) => d.description));
if (uniqueDescs.size > 1) {
if (context.strict) {
const conflictDetails = withDesc
.map((d) => ` Action "${d.action}": "${d.description}"`)
.join('\n');
throw new Error(
`Description conflict for field "${fieldName}" in discriminated union:\n` +
`${conflictDetails}\n` +
'Harmonize to a single description across all actions that share this field.',
);
}
// Non-strict: combine with action context for external MCP tools
const combined = withDesc.map((d) => `For "${d.action}": ${d.description}`).join('. ');
mergedShape[fieldName] = sanitizedField.describe(combined);
} else if (entries.length < actionMeta.length) {
// Field appears in a subset of variants. Annotate with an action hint so
// the model binds the (now-flattened-optional) field to the right actions
// and stops cross-mixing fields between sibling actions.
const actionList = entries.map((e) => `"${e.action}"`).join(', ');
const baseDesc = withDesc[0]?.description;
const merged = baseDesc ? `For ${actionList}: ${baseDesc}` : `Only for ${actionList}`;
mergedShape[fieldName] = sanitizedField.describe(merged);
} else {
mergedShape[fieldName] = sanitizedField;
}
}
return z.object(mergedShape);
return sanitizeDiscriminatedUnion(
schema as z.ZodDiscriminatedUnion<string, Array<z.ZodObject<z.ZodRawShape>>>,
context,
sanitizeChild,
);
}
// ZodUnion — strip ZodNull members, make result optional if null was present
if (schema instanceof z.ZodUnion) {
const options = (schema as z.ZodUnion<[z.ZodTypeAny, ...z.ZodTypeAny[]]>)
.options as z.ZodTypeAny[];
if (options.length > context.maxUnionOptions) {
throw createLimitError(
context,
`MCP schema union exceeds maximum option count of ${context.maxUnionOptions}`,
'unionOptions',
context.maxUnionOptions,
options.length,
);
}
const nonNull = options.filter((o) => !(o instanceof z.ZodNull));
const hadNull = nonNull.length < options.length;
const sanitized = nonNull.map((o, index) =>
sanitizeChild(o, `${context.path}.union[${index}]`),
return sanitizeUnion(
schema as z.ZodUnion<[z.ZodTypeAny, ...z.ZodTypeAny[]]>,
context,
sanitizeChild,
);
if (sanitized.length === 0) {
// All options were null — degenerate case
return z.string().optional();
}
if (sanitized.length === 1) {
return hadNull ? sanitized[0].optional() : sanitized[0];
}
const union = z.union(sanitized as [z.ZodTypeAny, z.ZodTypeAny, ...z.ZodTypeAny[]]);
return hadNull ? union.optional() : union;
}
// ZodObject — recurse into shape
@ -594,20 +660,7 @@ function sanitizeZodTypeInner(schema: z.ZodTypeAny, context: SanitizeContext): z
);
}
if (
schema instanceof z.ZodMap ||
schema instanceof z.ZodSet ||
schema instanceof z.ZodPromise ||
schema instanceof z.ZodFunction ||
schema instanceof z.ZodIntersection ||
schema instanceof z.ZodTuple ||
schema instanceof z.ZodNaN ||
schema instanceof z.ZodBigInt ||
schema instanceof z.ZodUndefined ||
schema instanceof z.ZodNever ||
schema instanceof z.ZodVoid ||
schema instanceof z.ZodSymbol
) {
if (isUnsupportedZodType(schema)) {
throw createUnsupportedTypeError(context, schema);
}

349
packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts
View File

@ -190,6 +190,231 @@ export function normalizeStreamSource(result: unknown): ResumableStreamSource {
throw new Error('Unsupported agent stream result');
}
function buildCancelledResult(
agentRunId: string,
text: Promise<string> | undefined,
workSummaryAccumulator: WorkSummaryAccumulator,
usageAccumulator: UsageAccumulator,
): ExecuteResumableStreamResult {
return {
status: 'cancelled',
agentRunId,
text,
workSummary: workSummaryAccumulator.toSummary(),
usage: usageAccumulator.hasUsage() ? usageAccumulator.toUsage() : undefined,
};
}
/**
* Record a parsed suspension. The first suspension is kept (and, in auto mode,
* its confirmation awaited); any further distinct suspension before resume is
* logged and deferred.
*/
function recordSuspension(
parsedSuspension: SuspensionInfo,
current: SuspensionInfo | undefined,
control: ResumableStreamControl,
context: ResumableStreamContext,
): { suspension: SuspensionInfo; pendingConfirmation?: Promise<Record<string, unknown>> } {
if (current) {
if (!isSameSuspension(parsedSuspension, current)) {
context.logger.warn('Additional HITL suspension encountered before resume; deferring', {
threadId: context.threadId,
runId: context.runId,
activeRequestId: current.requestId,
deferredRequestId: parsedSuspension.requestId,
activeToolCallId: current.toolCallId,
deferredToolCallId: parsedSuspension.toolCallId,
});
}
return { suspension: current };
}
if (control.mode === 'auto') {
control.onSuspension?.(parsedSuspension);
return {
suspension: parsedSuspension,
pendingConfirmation: control.waitForConfirmation(parsedSuspension.requestId),
};
}
return { suspension: parsedSuspension };
}
/**
* Publish redacted events, holding back the primary confirmation-request event in
* manual mode and de-duplicating it. Returns the updated confirmation-tracking state.
*/
function publishRedactedEvents(
events: InstanceAiEvent[],
args: {
suspension: SuspensionInfo | undefined;
confirmationEvent: ConfirmationRequestEvent | undefined;
confirmationEventPublished: boolean;
control: ResumableStreamControl;
context: ResumableStreamContext;
workSummaryAccumulator: WorkSummaryAccumulator;
},
): { confirmationEvent?: ConfirmationRequestEvent; confirmationEventPublished: boolean } {
const { suspension, control, context, workSummaryAccumulator } = args;
let confirmationEvent = args.confirmationEvent;
let confirmationEventPublished = args.confirmationEventPublished;
for (const event of events) {
workSummaryAccumulator.observe(event);
let shouldPublishEvent = true;
if (event.type === 'confirmation-request') {
const isPrimarySuspension =
suspension !== undefined &&
event.payload.requestId === suspension.requestId &&
event.payload.toolCallId === suspension.toolCallId;
if (!isPrimarySuspension || confirmationEventPublished || confirmationEvent) {
shouldPublishEvent = false;
}
if (shouldPublishEvent && control.mode === 'manual') {
confirmationEvent = event;
shouldPublishEvent = false;
}
if (shouldPublishEvent) {
confirmationEventPublished = true;
}
}
if (shouldPublishEvent) {
context.eventBus.publish(context.threadId, event);
}
}
return { confirmationEvent, confirmationEventPublished };
}
interface StreamPassResult {
cancelled: boolean;
suspension?: SuspensionInfo;
hasError: boolean;
error?: unknown;
pendingConfirmation?: Promise<Record<string, unknown>>;
confirmationEvent?: ConfirmationRequestEvent;
drainedCorrectionsForResume: string[];
currentResponseId: string | undefined;
nativeStepIndex: number;
}
/**
* Consume one stream until it ends (or is cancelled), publishing redacted events,
* accumulating usage/work, and capturing the first suspension. Returns the pass
* outcome plus the updated response-id / step counters for the next pass.
*/
async function consumeStreamPass(args: {
activeStream: AsyncIterable<unknown>;
activeAgentRunId: string;
options: ExecuteResumableStreamOptions;
workSummaryAccumulator: WorkSummaryAccumulator;
usageAccumulator: UsageAccumulator;
outputRedactor: OutputRedactor;
currentResponseId: string | undefined;
nativeStepIndex: number;
}): Promise<StreamPassResult> {
const {
activeStream,
activeAgentRunId,
options,
workSummaryAccumulator,
usageAccumulator,
outputRedactor,
} = args;
let currentResponseId = args.currentResponseId;
let nativeStepIndex = args.nativeStepIndex;
let suspension: SuspensionInfo | undefined;
let hasError = false;
let error: unknown;
let pendingConfirmation: Promise<Record<string, unknown>> | undefined;
let confirmationEvent: ConfirmationRequestEvent | undefined;
let confirmationEventPublished = false;
const drainedCorrectionsForResume: string[] = [];
for await (const chunk of activeStream) {
if (options.context.signal.aborted) {
return {
cancelled: true,
hasError,
drainedCorrectionsForResume,
currentResponseId,
nativeStepIndex,
};
}
options.context.onActivity?.();
usageAccumulator.observe(chunk);
if (isRecord(chunk) && chunk.type === 'start-step') {
nativeStepIndex += 1;
const responseRunId = activeAgentRunId || options.context.runId;
currentResponseId = `${responseRunId}:step:${nativeStepIndex}`;
}
const parsedSuspension = parseSuspension(chunk);
if (parsedSuspension) {
const recorded = recordSuspension(
parsedSuspension,
suspension,
options.control,
options.context,
);
suspension = recorded.suspension;
if (recorded.pendingConfirmation) pendingConfirmation = recorded.pendingConfirmation;
}
if (isErrorChunk(chunk)) {
hasError = true;
error = chunk.error;
}
const mappedEvent = mapAgentChunkToEvent(
options.context.runId,
options.context.agentId,
chunk,
currentResponseId,
);
// Scan/redact secrets & PII before events reach the user. Buffered
// delta text is released here at structural boundaries, so this may
// expand into several events (or none, while text is held back).
const events = mappedEvent ? outputRedactor.processEvent(mappedEvent) : [];
const published = publishRedactedEvents(events, {
suspension,
confirmationEvent,
confirmationEventPublished,
control: options.control,
context: options.context,
workSummaryAccumulator,
});
confirmationEvent = published.confirmationEvent;
confirmationEventPublished = published.confirmationEventPublished;
if (options.control.mode === 'auto' && options.control.drainCorrections) {
const corrections = options.control.drainCorrections();
publishCorrections(options.context, corrections);
drainedCorrectionsForResume.push(...corrections);
}
}
return {
cancelled: false,
suspension,
hasError,
error,
pendingConfirmation,
confirmationEvent,
drainedCorrectionsForResume,
currentResponseId,
nativeStepIndex,
};
}
export async function executeResumableStream(
options: ExecuteResumableStreamOptions,
): Promise<ExecuteResumableStreamResult> {
@ -210,121 +435,33 @@ export async function executeResumableStream(
let nativeStepIndex = 0;
while (true) {
let suspension: SuspensionInfo | undefined;
let hasError = false;
let error: unknown;
let pendingConfirmation: Promise<Record<string, unknown>> | undefined;
let confirmationEvent: ConfirmationRequestEvent | undefined;
let confirmationEventPublished = false;
const drainedCorrectionsForResume: string[] = [];
for await (const chunk of activeStream) {
if (options.context.signal.aborted) {
return {
status: 'cancelled',
agentRunId: activeAgentRunId,
text,
workSummary: workSummaryAccumulator.toSummary(),
usage: usageAccumulator.hasUsage() ? usageAccumulator.toUsage() : undefined,
};
}
const pass = await consumeStreamPass({
activeStream,
activeAgentRunId,
options,
workSummaryAccumulator,
usageAccumulator,
outputRedactor,
currentResponseId,
nativeStepIndex,
});
currentResponseId = pass.currentResponseId;
nativeStepIndex = pass.nativeStepIndex;
options.context.onActivity?.();
usageAccumulator.observe(chunk);
if (isRecord(chunk) && chunk.type === 'start-step') {
nativeStepIndex += 1;
const responseRunId = activeAgentRunId || options.context.runId;
currentResponseId = `${responseRunId}:step:${nativeStepIndex}`;
}
const parsedSuspension = parseSuspension(chunk);
if (parsedSuspension) {
if (!suspension) {
suspension = parsedSuspension;
if (options.control.mode === 'auto') {
options.control.onSuspension?.(parsedSuspension);
pendingConfirmation = options.control.waitForConfirmation(parsedSuspension.requestId);
}
} else if (!isSameSuspension(parsedSuspension, suspension)) {
options.context.logger.warn(
'Additional HITL suspension encountered before resume; deferring',
{
threadId: options.context.threadId,
runId: options.context.runId,
activeRequestId: suspension.requestId,
deferredRequestId: parsedSuspension.requestId,
activeToolCallId: suspension.toolCallId,
deferredToolCallId: parsedSuspension.toolCallId,
},
);
}
}
if (isErrorChunk(chunk)) {
hasError = true;
error = chunk.error;
}
const mappedEvent = mapAgentChunkToEvent(
options.context.runId,
options.context.agentId,
chunk,
currentResponseId,
);
// Scan/redact secrets & PII before events reach the user. Buffered
// delta text is released here at structural boundaries, so this may
// expand into several events (or none, while text is held back).
const events = mappedEvent ? outputRedactor.processEvent(mappedEvent) : [];
for (const event of events) {
workSummaryAccumulator.observe(event);
let shouldPublishEvent = true;
if (event.type === 'confirmation-request') {
const isPrimarySuspension =
suspension !== undefined &&
event.payload.requestId === suspension.requestId &&
event.payload.toolCallId === suspension.toolCallId;
if (!isPrimarySuspension || confirmationEventPublished || confirmationEvent) {
shouldPublishEvent = false;
}
if (shouldPublishEvent && options.control.mode === 'manual') {
confirmationEvent = event;
shouldPublishEvent = false;
}
if (shouldPublishEvent) {
confirmationEventPublished = true;
}
}
if (shouldPublishEvent) {
options.context.eventBus.publish(options.context.threadId, event);
}
}
if (options.control.mode === 'auto' && options.control.drainCorrections) {
const corrections = options.control.drainCorrections();
publishCorrections(options.context, corrections);
drainedCorrectionsForResume.push(...corrections);
}
if (pass.cancelled) {
return buildCancelledResult(activeAgentRunId, text, workSummaryAccumulator, usageAccumulator);
}
const { suspension, hasError, error, pendingConfirmation, confirmationEvent } = pass;
const { drainedCorrectionsForResume } = pass;
for (const flushed of outputRedactor.flush()) {
workSummaryAccumulator.observe(flushed);
options.context.eventBus.publish(options.context.threadId, flushed);
}
if (options.context.signal.aborted) {
return {
status: 'cancelled',
agentRunId: activeAgentRunId,
text,
workSummary: workSummaryAccumulator.toSummary(),
usage: usageAccumulator.hasUsage() ? usageAccumulator.toUsage() : undefined,
};
return buildCancelledResult(activeAgentRunId, text, workSummaryAccumulator, usageAccumulator);
}
if (!suspension) {

533
packages/@n8n/instance-ai/src/stream/map-chunk.ts
View File

@ -7,15 +7,7 @@ import {
gatewayConfirmationRequiredPayloadSchema,
webSearchMetaSchema,
} from '@n8n/api-types';
import type {
InstanceAiCredentialRequest,
InstanceAiEvent,
InstanceAiWorkflowSetupNode,
PlannedTaskArg,
TaskList,
GatewayConfirmationRequiredPayload,
WebSearchMeta,
} from '@n8n/api-types';
import type { InstanceAiEvent } from '@n8n/api-types';
import { z } from 'zod';
const questionItemSchema = z.object({
@ -165,6 +157,239 @@ function extractErrorInfo(error: unknown): ErrorInfo {
return { content: 'Unknown error' };
}
type EventBase = { runId: string; agentId: string; responseId?: string };
type ConfirmationInputType =
| 'approval'
| 'text'
| 'questions'
| 'plan-review'
| 'resource-decision'
| 'continue';
/** A non-empty string, or undefined for anything else (matches the legacy `value ? value : undefined` gate). */
function presentString(value: unknown): string | undefined {
return typeof value === 'string' && value ? value : undefined;
}
function parseSchemaArray<T>(value: unknown, schema: z.ZodType<T>): T[] | undefined {
if (!Array.isArray(value)) return undefined;
const parsed = value
.map((item) => schema.safeParse(item))
.filter((r) => r.success)
.map((r) => r.data);
return parsed.length > 0 ? parsed : undefined;
}
function parseSchemaRecord<T>(value: unknown, schema: z.ZodType<T>): T | undefined {
if (!isRecord(value)) return undefined;
const parsed = schema.safeParse(value);
return parsed.success ? parsed.data : undefined;
}
function parseSeverity(value: unknown): 'destructive' | 'warning' | 'info' {
const raw = typeof value === 'string' ? value : '';
const valid = ['destructive', 'warning', 'info'] as const;
return (valid as readonly string[]).includes(raw) ? (raw as (typeof valid)[number]) : 'warning';
}
function parseInputType(value: unknown): ConfirmationInputType | undefined {
const raw = typeof value === 'string' ? value : undefined;
const valid = [
'approval',
'text',
'questions',
'plan-review',
'resource-decision',
'continue',
] as const;
return (valid as readonly string[]).includes(raw ?? '')
? (raw as (typeof valid)[number])
: undefined;
}
function parseDomainAccess(value: unknown): { url: string; host: string } | undefined {
const raw = isRecord(value) ? value : undefined;
return raw && typeof raw.url === 'string' && typeof raw.host === 'string'
? { url: raw.url, host: raw.host }
: undefined;
}
function parseCredentialFlow(value: unknown): { stage: 'generic' | 'finalize' } | undefined {
const raw = isRecord(value) ? value : undefined;
const validStages = new Set<'generic' | 'finalize'>(['generic', 'finalize']);
const stage = raw && typeof raw.stage === 'string' ? raw.stage : undefined;
return stage !== undefined && validStages.has(stage as 'generic' | 'finalize')
? { stage: stage as 'generic' | 'finalize' }
: undefined;
}
function mapToolCallChunk(
chunk: Extract<StreamChunk, { type: 'tool-call' }>,
base: EventBase,
): InstanceAiEvent {
return {
type: 'tool-call',
...base,
payload: {
toolCallId: chunk.toolCallId,
toolName: chunk.toolName,
args: isRecord(chunk.input) ? chunk.input : {},
},
};
}
function mapToolResultChunk(
chunk: Extract<StreamChunk, { type: 'tool-result' }>,
base: EventBase,
): InstanceAiEvent {
if (chunk.isError === true) {
return {
type: 'tool-error',
...base,
payload: { toolCallId: chunk.toolCallId, error: extractToolErrorText(chunk.output) },
};
}
return {
type: 'tool-result',
...base,
payload: { toolCallId: chunk.toolCallId, result: chunk.output },
};
}
function mapSuspendedChunk(
chunk: Extract<StreamChunk, { type: 'tool-call-suspended' }>,
base: EventBase,
): InstanceAiEvent | null {
const suspendPayload = isRecord(chunk.suspendPayload) ? chunk.suspendPayload : {};
const toolCallId = typeof chunk.toolCallId === 'string' ? chunk.toolCallId : '';
const requestId =
typeof suspendPayload.requestId === 'string' && suspendPayload.requestId
? suspendPayload.requestId
: toolCallId;
if (!requestId || !toolCallId) return null;
const credentialRequests = parseSchemaArray(
suspendPayload.credentialRequests,
credentialRequestSchema,
);
const projectId = presentString(suspendPayload.projectId);
const inputType = parseInputType(suspendPayload.inputType);
const questions = parseSchemaArray(suspendPayload.questions, questionItemSchema);
const introMessage = presentString(suspendPayload.introMessage);
const tasks = parseSchemaRecord(suspendPayload.tasks, taskListSchema);
const planItems = parseSchemaArray(suspendPayload.planItems, plannedTaskArgSchema);
const domainAccess = parseDomainAccess(suspendPayload.domainAccess);
const webSearch = parseSchemaRecord(suspendPayload.webSearch, webSearchMetaSchema);
const credentialFlow = parseCredentialFlow(suspendPayload.credentialFlow);
const setupRequests = parseSchemaArray(suspendPayload.setupRequests, workflowSetupNodeSchema);
const workflowId = presentString(suspendPayload.workflowId);
const resourceDecision = parseSchemaRecord(
suspendPayload.resourceDecision,
gatewayConfirmationRequiredPayloadSchema,
);
return {
type: 'confirmation-request',
...base,
payload: {
requestId,
toolCallId,
toolName: typeof chunk.toolName === 'string' ? chunk.toolName : '',
args: isRecord(chunk.input) ? chunk.input : {},
severity: parseSeverity(suspendPayload.severity),
message:
typeof suspendPayload.message === 'string'
? suspendPayload.message
: 'Confirmation required',
...(credentialRequests ? { credentialRequests } : {}),
...(projectId ? { projectId } : {}),
...(inputType ? { inputType } : {}),
...(domainAccess ? { domainAccess } : {}),
...(webSearch ? { webSearch } : {}),
...(credentialFlow ? { credentialFlow } : {}),
...(setupRequests ? { setupRequests } : {}),
...(workflowId ? { workflowId } : {}),
...(questions ? { questions } : {}),
...(introMessage ? { introMessage } : {}),
...(tasks ? { tasks } : {}),
...(planItems ? { planItems } : {}),
...(resourceDecision ? { resourceDecision } : {}),
},
};
}
function toolCallFromContent(part: unknown, base: EventBase): InstanceAiEvent | null {
if (!isRecord(part) || part.type !== 'tool-call') return null;
return {
type: 'tool-call',
...base,
payload: {
toolCallId: typeof part.toolCallId === 'string' ? part.toolCallId : '',
toolName: typeof part.toolName === 'string' ? part.toolName : '',
args: isRecord(part.input) ? part.input : {},
},
};
}
function toolResultFromContent(part: unknown, base: EventBase): InstanceAiEvent | null {
if (!isRecord(part) || part.type !== 'tool-result') return null;
const toolCallId = typeof part.toolCallId === 'string' ? part.toolCallId : '';
if (part.isError === true) {
return {
type: 'tool-error',
...base,
payload: { toolCallId, error: extractToolErrorText(part.result) },
};
}
return {
type: 'tool-result',
...base,
payload: { toolCallId, result: part.result },
};
}
function mapMessageChunk(
chunk: Extract<StreamChunk, { type: 'message' }>,
base: EventBase,
): InstanceAiEvent | null {
const message = getRecordProperty(chunk, 'message');
if (message?.role !== 'tool') return null;
const content = getArrayProperty(message, 'content');
if (!content) return null;
const toolCall = toolCallFromContent(
content.find((part) => isRecord(part) && part.type === 'tool-call'),
base,
);
if (toolCall) return toolCall;
return toolResultFromContent(
content.find((part) => isRecord(part) && part.type === 'tool-result'),
base,
);
}
function mapErrorChunk(
chunk: Extract<StreamChunk, { type: 'error' }>,
base: EventBase,
): InstanceAiEvent {
const errorInfo = extractErrorInfo(chunk.error);
return {
type: 'error',
...base,
payload: {
content: errorInfo.content,
...(errorInfo.statusCode !== undefined ? { statusCode: errorInfo.statusCode } : {}),
...(errorInfo.provider ? { provider: errorInfo.provider } : {}),
...(errorInfo.technicalDetails ? { technicalDetails: errorInfo.technicalDetails } : {}),
},
};
}
export function mapAgentChunkToEvent(
runId: string,
agentId: string,
@ -173,278 +398,24 @@ export function mapAgentChunkToEvent(
): InstanceAiEvent | null {
if (!isAgentStreamChunk(chunk)) return null;
const base = { runId, agentId, ...(responseId ? { responseId } : {}) };
const base: EventBase = { runId, agentId, ...(responseId ? { responseId } : {}) };
if (chunk.type === 'text-delta') {
return {
type: 'text-delta',
...base,
payload: { text: chunk.delta },
};
switch (chunk.type) {
case 'text-delta':
return { type: 'text-delta', ...base, payload: { text: chunk.delta } };
case 'reasoning-delta':
return { type: 'reasoning-delta', ...base, payload: { text: chunk.delta } };
case 'tool-call':
return mapToolCallChunk(chunk, base);
case 'tool-result':
return mapToolResultChunk(chunk, base);
case 'tool-call-suspended':
return mapSuspendedChunk(chunk, base);
case 'message':
return mapMessageChunk(chunk, base);
case 'error':
return mapErrorChunk(chunk, base);
default:
return null;
}
if (chunk.type === 'reasoning-delta') {
return {
type: 'reasoning-delta',
...base,
payload: { text: chunk.delta },
};
}
if (chunk.type === 'tool-call') {
return {
type: 'tool-call',
...base,
payload: {
toolCallId: chunk.toolCallId,
toolName: chunk.toolName,
args: isRecord(chunk.input) ? chunk.input : {},
},
};
}
if (chunk.type === 'tool-result') {
if (chunk.isError === true) {
return {
type: 'tool-error',
...base,
payload: {
toolCallId: chunk.toolCallId,
error: extractToolErrorText(chunk.output),
},
};
}
return {
type: 'tool-result',
...base,
payload: {
toolCallId: chunk.toolCallId,
result: chunk.output,
},
};
}
if (chunk.type === 'tool-call-suspended') {
const suspendPayload = isRecord(chunk.suspendPayload) ? chunk.suspendPayload : {};
const toolCallId = typeof chunk.toolCallId === 'string' ? chunk.toolCallId : '';
const requestId =
typeof suspendPayload.requestId === 'string' && suspendPayload.requestId
? suspendPayload.requestId
: toolCallId;
if (!requestId || !toolCallId) return null;
const rawSeverity = typeof suspendPayload.severity === 'string' ? suspendPayload.severity : '';
const validSeverities = ['destructive', 'warning', 'info'] as const;
const severity = (validSeverities as readonly string[]).includes(rawSeverity)
? (rawSeverity as (typeof validSeverities)[number])
: 'warning';
let credentialRequests: InstanceAiCredentialRequest[] | undefined;
if (Array.isArray(suspendPayload.credentialRequests)) {
const parsed = suspendPayload.credentialRequests
.map((item) => credentialRequestSchema.safeParse(item))
.filter((r) => r.success)
.map((r) => r.data);
if (parsed.length > 0) {
credentialRequests = parsed;
}
}
const projectId =
typeof suspendPayload.projectId === 'string' ? suspendPayload.projectId : undefined;
const rawInputType =
typeof suspendPayload.inputType === 'string' ? suspendPayload.inputType : undefined;
const validInputTypes = [
'approval',
'text',
'questions',
'plan-review',
'resource-decision',
'continue',
] as const;
const inputType = (validInputTypes as readonly string[]).includes(rawInputType ?? '')
? (rawInputType as (typeof validInputTypes)[number])
: undefined;
let questions: Array<z.infer<typeof questionItemSchema>> | undefined;
if (Array.isArray(suspendPayload.questions)) {
const parsed = suspendPayload.questions
.map((item) => questionItemSchema.safeParse(item))
.filter((r) => r.success)
.map((r) => r.data);
if (parsed.length > 0) {
questions = parsed;
}
}
const introMessage =
typeof suspendPayload.introMessage === 'string' ? suspendPayload.introMessage : undefined;
let tasks: TaskList | undefined;
if (isRecord(suspendPayload.tasks)) {
const parsed = taskListSchema.safeParse(suspendPayload.tasks);
if (parsed.success) {
tasks = parsed.data;
}
}
let planItems: PlannedTaskArg[] | undefined;
if (Array.isArray(suspendPayload.planItems)) {
const parsed = suspendPayload.planItems
.map((item) => plannedTaskArgSchema.safeParse(item))
.filter((r) => r.success)
.map((r) => r.data);
if (parsed.length > 0) {
planItems = parsed;
}
}
const rawDomainAccess = isRecord(suspendPayload.domainAccess)
? suspendPayload.domainAccess
: undefined;
const domainAccess =
rawDomainAccess &&
typeof rawDomainAccess.url === 'string' &&
typeof rawDomainAccess.host === 'string'
? { url: rawDomainAccess.url, host: rawDomainAccess.host }
: undefined;
let webSearch: WebSearchMeta | undefined;
if (isRecord(suspendPayload.webSearch)) {
const parsed = webSearchMetaSchema.safeParse(suspendPayload.webSearch);
if (parsed.success) {
webSearch = parsed.data;
}
}
const rawCredentialFlow = isRecord(suspendPayload.credentialFlow)
? suspendPayload.credentialFlow
: undefined;
const validStages = new Set<'generic' | 'finalize'>(['generic', 'finalize']);
const rawStage =
rawCredentialFlow && typeof rawCredentialFlow.stage === 'string'
? rawCredentialFlow.stage
: undefined;
const credentialFlow =
rawStage !== undefined && validStages.has(rawStage as 'generic' | 'finalize')
? { stage: rawStage as 'generic' | 'finalize' }
: undefined;
let setupRequests: InstanceAiWorkflowSetupNode[] | undefined;
if (Array.isArray(suspendPayload.setupRequests)) {
const parsed = suspendPayload.setupRequests
.map((item) => workflowSetupNodeSchema.safeParse(item))
.filter((r) => r.success)
.map((r) => r.data);
if (parsed.length > 0) {
setupRequests = parsed;
}
}
const workflowId =
typeof suspendPayload.workflowId === 'string' ? suspendPayload.workflowId : undefined;
let resourceDecision: GatewayConfirmationRequiredPayload | undefined;
if (isRecord(suspendPayload.resourceDecision)) {
const parsed = gatewayConfirmationRequiredPayloadSchema.safeParse(
suspendPayload.resourceDecision,
);
if (parsed.success) {
resourceDecision = parsed.data;
}
}
return {
type: 'confirmation-request',
...base,
payload: {
requestId,
toolCallId,
toolName: typeof chunk.toolName === 'string' ? chunk.toolName : '',
args: isRecord(chunk.input) ? chunk.input : {},
severity,
message:
typeof suspendPayload.message === 'string'
? suspendPayload.message
: 'Confirmation required',
...(credentialRequests ? { credentialRequests } : {}),
...(projectId ? { projectId } : {}),
...(inputType ? { inputType } : {}),
...(domainAccess ? { domainAccess } : {}),
...(webSearch ? { webSearch } : {}),
...(credentialFlow ? { credentialFlow } : {}),
...(setupRequests ? { setupRequests } : {}),
...(workflowId ? { workflowId } : {}),
...(questions ? { questions } : {}),
...(introMessage ? { introMessage } : {}),
...(tasks ? { tasks } : {}),
...(planItems ? { planItems } : {}),
...(resourceDecision ? { resourceDecision } : {}),
},
};
}
if (chunk.type === 'message') {
const message = getRecordProperty(chunk, 'message');
if (message?.role !== 'tool') return null;
const content = getArrayProperty(message, 'content');
if (!content) return null;
const toolCall = content.find((part) => isRecord(part) && part.type === 'tool-call');
if (isRecord(toolCall) && toolCall.type === 'tool-call') {
return {
type: 'tool-call',
...base,
payload: {
toolCallId: typeof toolCall.toolCallId === 'string' ? toolCall.toolCallId : '',
toolName: typeof toolCall.toolName === 'string' ? toolCall.toolName : '',
args: isRecord(toolCall.input) ? toolCall.input : {},
},
};
}
const toolResult = content.find((part) => isRecord(part) && part.type === 'tool-result');
if (isRecord(toolResult) && toolResult.type === 'tool-result') {
if (toolResult.isError === true) {
return {
type: 'tool-error',
...base,
payload: {
toolCallId: typeof toolResult.toolCallId === 'string' ? toolResult.toolCallId : '',
error: extractToolErrorText(toolResult.result),
},
};
}
return {
type: 'tool-result',
...base,
payload: {
toolCallId: typeof toolResult.toolCallId === 'string' ? toolResult.toolCallId : '',
result: toolResult.result,
},
};
}
}
if (chunk.type === 'error') {
const errorInfo = extractErrorInfo(chunk.error);
return {
type: 'error',
...base,
payload: {
content: errorInfo.content,
...(errorInfo.statusCode !== undefined ? { statusCode: errorInfo.statusCode } : {}),
...(errorInfo.provider ? { provider: errorInfo.provider } : {}),
...(errorInfo.technicalDetails ? { technicalDetails: errorInfo.technicalDetails } : {}),
},
};
}
return null;
}

23
packages/@n8n/instance-ai/src/tools/evals/column-ref-utils.ts
View File

@ -286,21 +286,24 @@ export function extractNamedRefMatches(text: string): NamedRefMatch[] {
* Returns names of nodes connected into `agentNodeName` via any `ai_*`
* connection type — i.e. tools, memory, output parsers, etc.
*/
export function connectionSlotTargetsNode(slot: unknown, targetNodeName: string): boolean {
if (!Array.isArray(slot)) return false;
for (const inner of slot) {
if (!Array.isArray(inner)) continue;
for (const conn of inner) {
if (isRecord(conn) && conn.node === targetNodeName) return true;
}
}
return false;
}
export function findAgentSubComponents(workflow: WorkflowJSON, agentNodeName: string): string[] {
const subs = new Set<string>();
for (const [sourceName, byType] of Object.entries(workflow.connections ?? {})) {
if (!isRecord(byType)) continue;
for (const [connType, slot] of Object.entries(byType)) {
if (!connType.startsWith('ai_')) continue;
if (!Array.isArray(slot)) continue;
for (const inner of slot) {
if (!Array.isArray(inner)) continue;
for (const conn of inner) {
if (isRecord(conn) && conn.node === agentNodeName) {
subs.add(sourceName);
break;
}
}
if (connType.startsWith('ai_') && connectionSlotTargetsNode(slot, agentNodeName)) {
subs.add(sourceName);
}
}
}

24
packages/@n8n/instance-ai/src/tools/evals/describe-metric-for-workflow.service.ts
View File

@ -1,6 +1,6 @@
import type { WorkflowJSON } from '@n8n/workflow-sdk';
import { isRecord } from './column-ref-utils';
import { isRecord, connectionSlotTargetsNode } from './column-ref-utils';
import type { MetricId } from './metric-catalog';
type RecommendedMetricId = Exclude<MetricId, 'helpfulness'>;
@ -19,15 +19,8 @@ function findSourceNodesByConnectionType(
const connections = workflow.connections ?? {};
for (const [sourceName, byType] of Object.entries(connections)) {
if (!isRecord(byType)) continue;
const slot = byType[connectionType];
if (!Array.isArray(slot)) continue;
for (const inner of slot) {
if (!Array.isArray(inner)) continue;
for (const conn of inner) {
if (isRecord(conn) && conn.node === targetNodeName) {
sources.push(sourceName);
}
}
if (connectionSlotTargetsNode(byType[connectionType], targetNodeName)) {
sources.push(sourceName);
}
}
return sources;
@ -52,15 +45,8 @@ function getAgentRetrieverNames(workflow: WorkflowJSON, agentNodeName: string):
for (const [sourceName, byType] of Object.entries(connections)) {
if (!isRecord(byType)) continue;
for (const [connType, slot] of Object.entries(byType)) {
if (!connType.startsWith('ai_')) continue;
if (!Array.isArray(slot)) continue;
for (const inner of slot) {
if (!Array.isArray(inner)) continue;
for (const conn of inner) {
if (isRecord(conn) && conn.node === agentNodeName) {
candidates.add(sourceName);
}
}
if (connType.startsWith('ai_') && connectionSlotTargetsNode(slot, agentNodeName)) {
candidates.add(sourceName);
}
}
}

613
packages/@n8n/instance-ai/src/tools/orchestration/verify-built-workflow.tool.ts
View File

@ -197,6 +197,21 @@ const remediationOutputSchema = z
})
.optional();
const CREDENTIAL_FAILURE_KEYWORDS = [
'credential',
'unauthorized',
'forbidden',
'401',
'403',
'free tier',
'quota',
];
const TRANSIENT_FAILURE_KEYWORDS = ['429', 'rate limit', '502', 'bad gateway', 'timed out'];
function messageMatchesAny(normalized: string, keywords: readonly string[]): boolean {
return keywords.some((keyword) => normalized.includes(keyword));
}
function classifyVerificationFailure(
error: string | undefined,
status: string | undefined,
@ -231,15 +246,7 @@ function classifyVerificationFailure(
const mockedCredentialTypeCount = buildOutcome.mockedCredentialTypes?.length ?? 0;
const mockedNodeCount = buildOutcome.mockedNodeNames?.length ?? 0;
const hasMockedCredentialContext = Boolean(mockedCredentialTypeCount > 0 || mockedNodeCount > 0);
if (
normalized.includes('credential') ||
normalized.includes('unauthorized') ||
normalized.includes('forbidden') ||
normalized.includes('401') ||
normalized.includes('403') ||
normalized.includes('free tier') ||
normalized.includes('quota')
) {
if (messageMatchesAny(normalized, CREDENTIAL_FAILURE_KEYWORDS)) {
return createRemediation({
category: 'needs_setup',
shouldEdit: false,
@ -252,13 +259,7 @@ function classifyVerificationFailure(
});
}
if (
normalized.includes('429') ||
normalized.includes('rate limit') ||
normalized.includes('502') ||
normalized.includes('bad gateway') ||
normalized.includes('timed out')
) {
if (messageMatchesAny(normalized, TRANSIENT_FAILURE_KEYWORDS)) {
return createRemediation({
category: 'blocked',
shouldEdit: false,
@ -277,6 +278,324 @@ function classifyVerificationFailure(
});
}
const verifyBuiltWorkflowOutputSchema = z.object({
executionId: z.string().optional(),
success: z.boolean(),
status: z.enum(['running', 'success', 'error', 'waiting', 'unknown']).optional(),
nodesExecuted: z.array(z.string()).optional(),
nodePreviews: z
.array(
z.object({
nodeName: z.string(),
itemCount: z.number().optional(),
preview: z.string(),
truncated: z.boolean(),
chars: z.number(),
simulated: z.boolean().optional(),
}),
)
.optional(),
simulatedNodes: z.array(z.object({ nodeName: z.string(), reason: z.string() })).optional(),
simulationNote: z.string().optional(),
lastNodeExecuted: z.string().optional(),
nodesNotReached: z.array(z.string()).optional(),
coverageNote: z.string().optional(),
data: z.record(z.unknown()).optional(),
error: z.string().optional(),
remediation: remediationOutputSchema,
guidance: z.string().optional(),
});
type VerifyBuiltWorkflowOutput = z.infer<typeof verifyBuiltWorkflowOutputSchema>;
type VerifyInput = z.infer<typeof verifyBuiltWorkflowInputSchema>;
type WorkflowTaskService = NonNullable<OrchestrationContext['workflowTaskService']>;
type ExecutionRunResult = Awaited<
ReturnType<NonNullable<OrchestrationContext['domainContext']>['executionService']['run']>
>;
/** Names that actually ran, falling back to data keys for hosts that don't report executedNodeNames. */
function namesOrDataKeys(
reachedNames: Set<string>,
data: Record<string, unknown> | undefined,
): string[] | undefined {
if (reachedNames.size > 0) return [...reachedNames];
return data ? Object.keys(data) : undefined;
}
/**
* Validate the verify request and load the build outcome. Returns an early result
* to short-circuit the handler, or the resolved build outcome + prior loop state.
*/
async function resolveVerifyPreconditions(
input: VerifyInput,
context: OrchestrationContext,
): Promise<
| { result: VerifyBuiltWorkflowOutput }
| {
buildOutcome: WorkflowBuildOutcome;
workflowId: string;
stateBefore: Awaited<ReturnType<WorkflowTaskService['getWorkflowLoopState']>> | undefined;
workflowTaskService: WorkflowTaskService;
domainContext: NonNullable<OrchestrationContext['domainContext']>;
}
> {
if (!context.workflowTaskService || !context.domainContext) {
const remediation = createRemediation({
category: 'blocked',
shouldEdit: false,
reason: 'verification_support_unavailable',
guidance: 'Verification support is not available. Stop code edits and explain the blocker.',
});
return {
result: { success: false, error: 'Verification support not available.', remediation },
};
}
const stateBefore = await context.workflowTaskService.getWorkflowLoopState(input.workItemId);
const terminalRemediation =
stateBefore?.lastRemediation && !stateBefore.lastRemediation.shouldEdit
? terminalRemediationFromState(stateBefore, context.runId)
: undefined;
if (terminalRemediation) {
return {
result: {
success: false,
error: terminalRemediation.guidance,
remediation: terminalRemediation,
guidance: terminalRemediation.guidance,
},
};
}
const buildOutcome = await context.workflowTaskService.getBuildOutcome(input.workItemId);
if (!buildOutcome) {
const remediation = createRemediation({
category: 'blocked',
shouldEdit: false,
reason: 'missing_build_outcome',
guidance: `No build outcome found for work item ${input.workItemId}. Stop code edits and explain the blocker.`,
});
return {
result: {
success: false,
error: `No build outcome found for work item ${input.workItemId}.`,
remediation,
guidance: remediation.guidance,
},
};
}
if (!buildOutcome.workflowId) {
return {
result: {
success: false,
error: `Build outcome ${input.workItemId} does not include a workflow ID.`,
},
};
}
if (buildOutcome.workflowId !== input.workflowId) {
return {
result: {
success: false,
error:
`Build outcome ${input.workItemId} belongs to workflow ${buildOutcome.workflowId}, ` +
`but verification was requested for workflow ${input.workflowId}.`,
},
};
}
return {
buildOutcome,
workflowId: buildOutcome.workflowId,
stateBefore,
workflowTaskService: context.workflowTaskService,
domainContext: context.domainContext,
};
}
/**
* Handle the no-simulation-plan case: refuse to run (no safeguards for destructive
* nodes), persist the terminal verdict best-effort, and return a blocked result.
*/
async function handleMissingSimulationPlan(
input: VerifyInput,
context: OrchestrationContext,
workflowTaskService: WorkflowTaskService,
workflowId: string,
): Promise<VerifyBuiltWorkflowOutput> {
const guidance =
'Verification was not run because the build outcome has no simulation plan. ' +
'Rebuild or resubmit the workflow so destructive nodes can be classified before verification.';
const remediation = createRemediation({
category: 'blocked',
shouldEdit: false,
reason: 'missing_simulation_plan',
guidance,
});
context.logger.warn(
'verify-built-workflow: build outcome has no simulation plan — refusing to run without simulation safeguards',
{ workItemId: input.workItemId, workflowId },
);
try {
await workflowTaskService.updateBuildOutcome(input.workItemId, {
remediation,
verification: {
attempted: true,
success: false,
status: 'unknown',
failureSignature: 'missing_simulation_plan',
evidence: { errorMessage: guidance },
verifiedAt: new Date().toISOString(),
},
});
} catch {
// intentional: verification record persistence is advisory
}
try {
await workflowTaskService.reportVerificationVerdict({
workItemId: input.workItemId,
runId: context.runId,
workflowId,
verdict: 'failed_terminal',
failureSignature: 'missing_simulation_plan',
diagnosis: guidance,
remediation,
summary: guidance,
});
} catch (error) {
context.logger.warn('verify-built-workflow: failed to persist terminal verdict', {
workItemId: input.workItemId,
workflowId,
error: error instanceof Error ? error.message : String(error),
});
}
return { success: false, status: 'unknown', error: guidance, remediation, guidance };
}
/** Persist a structured verification record onto the build outcome (best-effort). */
async function persistVerificationRecord(
input: VerifyInput,
workflowTaskService: WorkflowTaskService,
args: {
success: boolean;
result: ExecutionRunResult;
reachedNames: Set<string>;
nodesNotReached: string[];
},
): Promise<void> {
const { success, result, reachedNames, nodesNotReached } = args;
try {
const executedForEvidence = namesOrDataKeys(reachedNames, result.data);
await workflowTaskService.updateBuildOutcome(input.workItemId, {
verification: {
attempted: true,
success,
executionId: result.executionId || undefined,
status: result.status,
failureSignature: success ? undefined : result.error,
evidence: {
nodesExecuted:
executedForEvidence && executedForEvidence.length > 0 ? executedForEvidence : undefined,
nodesNotReached: nodesNotReached.length > 0 ? nodesNotReached : undefined,
producedOutputRows: countProducedOutputRows(result.data),
errorMessage: success ? undefined : result.error,
},
verifiedAt: new Date().toISOString(),
},
});
} catch {
// intentional: verification record persistence is advisory
}
}
/** Report a terminal (non-editable) remediation verdict and emit guard telemetry (best-effort). */
async function reportTerminalRemediation(
input: VerifyInput,
context: OrchestrationContext,
workflowTaskService: WorkflowTaskService,
args: { remediation: RemediationMetadata; workflowId: string; executionId: string | undefined },
): Promise<void> {
const { remediation, workflowId, executionId } = args;
try {
await workflowTaskService.reportVerificationVerdict({
workItemId: input.workItemId,
runId: context.runId,
workflowId,
executionId,
verdict: remediation.category === 'needs_setup' ? 'needs_user_input' : 'failed_terminal',
failureSignature: remediation.reason,
diagnosis: remediation.guidance,
remediation,
summary: remediation.guidance,
});
} catch (error) {
context.logger.warn('verify-built-workflow: failed to persist terminal verdict', {
workItemId: input.workItemId,
workflowId,
error: error instanceof Error ? error.message : String(error),
});
}
try {
context.trackTelemetry?.('Builder remediation guard fired', {
thread_id: context.threadId,
run_id: context.runId,
work_item_id: input.workItemId,
workflow_id: workflowId,
category: remediation.category,
attempt_count: remediation.attemptCount,
reason: remediation.reason,
});
} catch (error) {
context.logger.warn('verify-built-workflow: failed to emit remediation telemetry', {
workItemId: input.workItemId,
workflowId,
error: error instanceof Error ? error.message : String(error),
});
}
}
function buildSimulationNote(
reachedSimulatedNodes: Array<{ nodeName: string; reason: string }>,
planMissing: boolean,
): string | undefined {
if (reachedSimulatedNodes.length > 0) {
return (
`Simulated ${reachedSimulatedNodes.length} node(s) during verification — no real external writes happened: ` +
reachedSimulatedNodes.map((n) => `${n.nodeName} (${n.reason})`).join('; ') +
'. Relay this to the user when presenting the result.'
);
}
if (planMissing) {
return (
'No simulation plan was available for this verification run — nodes were NOT ' +
'simulated and may have performed real external writes (sent messages, created or ' +
'modified records). Relay this to the user when presenting the result.'
);
}
return undefined;
}
function buildCoverageNote(
nodesNotReached: string[],
result: ExecutionRunResult,
success: boolean,
): string | undefined {
if (nodesNotReached.length === 0) return undefined;
const ending = result.lastNodeExecuted
? `. Execution ended at "${result.lastNodeExecuted}"${success ? ' because it produced no output items (empty item lists stop downstream nodes)' : ''}.`
: '.';
const guidance = success
? ' This usually means a lookup or query returned nothing. Seed matching test data and re-run verification, or tell the user the unreached part needs a manual test. Do NOT report the workflow as fully verified.'
: '';
return (
`Partial coverage: ${nodesNotReached.length} node(s) were never reached and remain UNVERIFIED: ` +
nodesNotReached.join(', ') +
ending +
guidance
);
}
export function createVerifyBuiltWorkflowTool(context: OrchestrationContext) {
return new Tool('verify-built-workflow')
.description(
@ -290,94 +609,12 @@ export function createVerifyBuiltWorkflowTool(context: OrchestrationContext) {
'look like an expression bug but are not — do not patch the workflow, re-run verify with the correct shape.',
)
.input(verifyBuiltWorkflowInputSchema)
.output(
z.object({
executionId: z.string().optional(),
success: z.boolean(),
status: z.enum(['running', 'success', 'error', 'waiting', 'unknown']).optional(),
nodesExecuted: z.array(z.string()).optional(),
nodePreviews: z
.array(
z.object({
nodeName: z.string(),
itemCount: z.number().optional(),
preview: z.string(),
truncated: z.boolean(),
chars: z.number(),
simulated: z.boolean().optional(),
}),
)
.optional(),
simulatedNodes: z.array(z.object({ nodeName: z.string(), reason: z.string() })).optional(),
simulationNote: z.string().optional(),
lastNodeExecuted: z.string().optional(),
nodesNotReached: z.array(z.string()).optional(),
coverageNote: z.string().optional(),
data: z.record(z.unknown()).optional(),
error: z.string().optional(),
remediation: remediationOutputSchema,
guidance: z.string().optional(),
}),
)
.handler(async (input: z.infer<typeof verifyBuiltWorkflowInputSchema>) => {
if (!context.workflowTaskService || !context.domainContext) {
const remediation = createRemediation({
category: 'blocked',
shouldEdit: false,
reason: 'verification_support_unavailable',
guidance:
'Verification support is not available. Stop code edits and explain the blocker.',
});
return { success: false, error: 'Verification support not available.', remediation };
}
const stateBefore = await context.workflowTaskService.getWorkflowLoopState(input.workItemId);
const terminalRemediation =
stateBefore?.lastRemediation && !stateBefore.lastRemediation.shouldEdit
? terminalRemediationFromState(stateBefore, context.runId)
: undefined;
if (terminalRemediation) {
return {
success: false,
error: terminalRemediation.guidance,
remediation: terminalRemediation,
guidance: terminalRemediation.guidance,
};
}
const buildOutcome = await context.workflowTaskService.getBuildOutcome(input.workItemId);
if (!buildOutcome) {
const remediation = createRemediation({
category: 'blocked',
shouldEdit: false,
reason: 'missing_build_outcome',
guidance: `No build outcome found for work item ${input.workItemId}. Stop code edits and explain the blocker.`,
});
return {
success: false,
error: `No build outcome found for work item ${input.workItemId}.`,
remediation,
guidance: remediation.guidance,
};
}
if (!buildOutcome.workflowId) {
return {
success: false,
error: `Build outcome ${input.workItemId} does not include a workflow ID.`,
};
}
if (buildOutcome.workflowId !== input.workflowId) {
return {
success: false,
error:
`Build outcome ${input.workItemId} belongs to workflow ${buildOutcome.workflowId}, ` +
`but verification was requested for workflow ${input.workflowId}.`,
};
}
const workflowId = buildOutcome.workflowId;
.output(verifyBuiltWorkflowOutputSchema)
.handler(async (input: VerifyInput) => {
const preconditions = await resolveVerifyPreconditions(input, context);
if ('result' in preconditions) return preconditions.result;
const { buildOutcome, workflowId, stateBefore, workflowTaskService, domainContext } =
preconditions;
// Destructive nodes (including dataTable writes) are simulated via the
// build outcome's node simulation plan, so verification creates no
@ -389,59 +626,7 @@ export function createVerifyBuiltWorkflowTool(context: OrchestrationContext) {
// silently executing everything for real.
const planMissing = buildOutcome.nodeSimulationPlan === undefined;
if (planMissing) {
const guidance =
'Verification was not run because the build outcome has no simulation plan. ' +
'Rebuild or resubmit the workflow so destructive nodes can be classified before verification.';
const remediation = createRemediation({
category: 'blocked',
shouldEdit: false,
reason: 'missing_simulation_plan',
guidance,
});
context.logger.warn(
'verify-built-workflow: build outcome has no simulation plan — refusing to run without simulation safeguards',
{ workItemId: input.workItemId, workflowId: buildOutcome.workflowId },
);
try {
await context.workflowTaskService.updateBuildOutcome(input.workItemId, {
remediation,
verification: {
attempted: true,
success: false,
status: 'unknown',
failureSignature: 'missing_simulation_plan',
evidence: { errorMessage: guidance },
verifiedAt: new Date().toISOString(),
},
});
} catch {
// intentional: verification record persistence is advisory
}
try {
await context.workflowTaskService.reportVerificationVerdict({
workItemId: input.workItemId,
runId: context.runId,
workflowId,
verdict: 'failed_terminal',
failureSignature: 'missing_simulation_plan',
diagnosis: guidance,
remediation,
summary: guidance,
});
} catch (error) {
context.logger.warn('verify-built-workflow: failed to persist terminal verdict', {
workItemId: input.workItemId,
workflowId,
error: error instanceof Error ? error.message : String(error),
});
}
return {
success: false,
status: 'unknown',
error: guidance,
remediation,
guidance,
};
return await handleMissingSimulationPlan(input, context, workflowTaskService, workflowId);
}
const { pinData: verificationPinData, simulatedNodes } =
buildVerificationPinData(buildOutcome);
@ -450,7 +635,7 @@ export function createVerifyBuiltWorkflowTool(context: OrchestrationContext) {
? Object.fromEntries(simulatedNodes.map((n) => [n.nodeName, { reason: n.reason }]))
: undefined;
const result = await context.domainContext.executionService.run(workflowId, input.inputData, {
const result = await domainContext.executionService.run(workflowId, input.inputData, {
timeout: input.timeout,
pinData: verificationPinData,
simulation: simulationMap,
@ -500,108 +685,28 @@ export function createVerifyBuiltWorkflowTool(context: OrchestrationContext) {
: undefined;
const remediation = budgetRemediation ?? failureRemediation;
// Persist a structured verification record onto the build outcome so the
// checkpoint follow-up turn can reuse it instead of re-running verify.
// Best-effort: swallow storage errors so they don't mask the verify result.
try {
const executedForEvidence =
reachedNames.size > 0
? [...reachedNames]
: result.data
? Object.keys(result.data)
: undefined;
await context.workflowTaskService.updateBuildOutcome(input.workItemId, {
verification: {
attempted: true,
success,
executionId: result.executionId || undefined,
status: result.status,
failureSignature: success ? undefined : result.error,
evidence: {
nodesExecuted:
executedForEvidence && executedForEvidence.length > 0
? executedForEvidence
: undefined,
nodesNotReached: nodesNotReached.length > 0 ? nodesNotReached : undefined,
producedOutputRows: countProducedOutputRows(result.data),
errorMessage: success ? undefined : result.error,
},
verifiedAt: new Date().toISOString(),
},
});
} catch {
// intentional: verification record persistence is advisory
}
// Persist a structured verification record (best-effort) so the checkpoint
// follow-up turn can reuse it instead of re-running verify.
await persistVerificationRecord(input, workflowTaskService, {
success,
result,
reachedNames,
nodesNotReached,
});
if (remediation && !remediation.shouldEdit) {
try {
await context.workflowTaskService.reportVerificationVerdict({
workItemId: input.workItemId,
runId: context.runId,
workflowId,
executionId: result.executionId || undefined,
verdict:
remediation.category === 'needs_setup' ? 'needs_user_input' : 'failed_terminal',
failureSignature: remediation.reason,
diagnosis: remediation.guidance,
remediation,
summary: remediation.guidance,
});
} catch (error) {
context.logger.warn('verify-built-workflow: failed to persist terminal verdict', {
workItemId: input.workItemId,
workflowId,
error: error instanceof Error ? error.message : String(error),
});
}
try {
context.trackTelemetry?.('Builder remediation guard fired', {
thread_id: context.threadId,
run_id: context.runId,
work_item_id: input.workItemId,
workflow_id: workflowId,
category: remediation.category,
attempt_count: remediation.attemptCount,
reason: remediation.reason,
});
} catch (error) {
context.logger.warn('verify-built-workflow: failed to emit remediation telemetry', {
workItemId: input.workItemId,
workflowId,
error: error instanceof Error ? error.message : String(error),
});
}
await reportTerminalRemediation(input, context, workflowTaskService, {
remediation,
workflowId,
executionId: result.executionId || undefined,
});
}
const maxDataChars = input.maxDataChars ?? DEFAULT_NODE_PREVIEW_CHARS;
const nodesExecuted =
reachedNames.size > 0
? [...reachedNames]
: result.data
? Object.keys(result.data)
: undefined;
const nodesExecuted = namesOrDataKeys(reachedNames, result.data);
const simulatedNames = new Set(reachedSimulatedNodes.map((n) => n.nodeName));
const simulationNote =
reachedSimulatedNodes.length > 0
? `Simulated ${reachedSimulatedNodes.length} node(s) during verification — no real external writes happened: ` +
reachedSimulatedNodes.map((n) => `${n.nodeName} (${n.reason})`).join('; ') +
'. Relay this to the user when presenting the result.'
: planMissing
? 'No simulation plan was available for this verification run — nodes were NOT ' +
'simulated and may have performed real external writes (sent messages, created or ' +
'modified records). Relay this to the user when presenting the result.'
: undefined;
const coverageNote =
nodesNotReached.length > 0
? `Partial coverage: ${nodesNotReached.length} node(s) were never reached and remain UNVERIFIED: ` +
nodesNotReached.join(', ') +
(result.lastNodeExecuted
? `. Execution ended at "${result.lastNodeExecuted}"${success ? ' because it produced no output items (empty item lists stop downstream nodes)' : ''}.`
: '.') +
(success
? ' This usually means a lookup or query returned nothing. Seed matching test data and re-run verification, or tell the user the unreached part needs a manual test. Do NOT report the workflow as fully verified.'
: '')
: undefined;
const simulationNote = buildSimulationNote(reachedSimulatedNodes, planMissing);
const coverageNote = buildCoverageNote(nodesNotReached, result, success);
return {
executionId: result.executionId || undefined,
success,

330
packages/@n8n/instance-ai/src/tools/workflows.tool.ts
View File

@ -485,11 +485,188 @@ async function handleUnarchive(
return { success: true };
}
type SetupState = { currentRequestId: string | null; preTestSnapshot: WorkflowJSON | null };
type SetupResumeData = NonNullable<WorkflowToolContext['resumeData']>;
/** Run a single trigger node and map the execution status to a setup trigger-test result. */
async function runTriggerTest(
context: InstanceAiContext,
workflowId: string,
triggerNodeName: string,
): Promise<{ status: 'success' | 'error' | 'listening'; error?: string }> {
try {
const result = await context.executionService.run(workflowId, undefined, {
timeout: 30_000,
triggerNodeName,
});
if (result.status === 'success') return { status: 'success' };
if (result.status === 'waiting') return { status: 'listening' };
return { status: 'error', error: result.error ?? 'Trigger test failed' };
} catch (error) {
return {
status: 'error',
error: error instanceof Error ? error.message : 'Trigger test failed',
};
}
}
/** Collect nodes whose applied credential failed its test, so they move from completed to failed. */
function collectCredentialTestFailures(
remainingRequests: Awaited<ReturnType<typeof analyzeWorkflow>>,
credentials: Record<string, Record<string, string>> | undefined,
): Array<{ nodeName: string; error: string }> {
const failures: Array<{ nodeName: string; error: string }> = [];
for (const req of remainingRequests) {
if (
req.credentialTestResult &&
!req.credentialTestResult.success &&
req.credentialType &&
credentials?.[req.node.name]?.[req.credentialType]
) {
failures.push({
nodeName: req.node.name,
error: `Credential test failed for ${req.credentialType}: ${req.credentialTestResult.message ?? 'Invalid credentials'}`,
});
}
}
return failures;
}
/** Setup state 3: persist setup, run the trigger, and re-suspend with the refreshed requests. */
async function handleSetupTestTrigger(
context: InstanceAiContext,
input: Extract<Input, { action: 'setup' }>,
ctx: WorkflowToolContext,
state: SetupState,
resumeData: SetupResumeData,
testTriggerNode: string,
) {
state.preTestSnapshot ??= await context.workflowService.getAsWorkflowJSON(input.workflowId);
const preTestApply = await applyNodeChanges(
context,
input.workflowId,
resumeData.credentials,
resumeData.nodeParameters,
);
const applyFailures = preTestApply.failed;
if (applyFailures.length > 0) {
return {
success: false,
error: `Failed to apply setup before trigger test: ${applyFailures.map((f) => `${f.nodeName}: ${f.error}`).join('; ')}`,
failedNodes: applyFailures,
};
}
const triggerTestResult = await runTriggerTest(context, input.workflowId, testTriggerNode);
const refreshedRequests = await analyzeWorkflow(context, input.workflowId, {
[testTriggerNode]: triggerTestResult,
});
// Generate a new requestId so the frontend doesn't filter it
// as already-resolved from the previous suspend cycle
state.currentRequestId = nanoid();
return await ctx.suspend({
requestId: state.currentRequestId,
message: 'Configure credentials for your workflow',
severity: 'info' as const,
setupRequests: refreshedRequests,
workflowId: input.workflowId,
...(input.projectId ? { projectId: input.projectId } : {}),
});
}
/** Setup state 4: apply credentials and parameters atomically and report the outcome. */
async function handleSetupApply(
context: InstanceAiContext,
input: Extract<Input, { action: 'setup' }>,
state: SetupState,
resumeData: SetupResumeData,
) {
try {
state.preTestSnapshot = null;
const applyResult = await applyNodeChanges(
context,
input.workflowId,
resumeData.credentials,
resumeData.nodeParameters,
);
const failedNodes = applyResult.failed.length > 0 ? applyResult.failed : undefined;
// Fetch updated workflow to include in response so the frontend can refresh the canvas
const updatedWorkflow = await context.workflowService.getAsWorkflowJSON(input.workflowId);
const updatedNodes = updatedWorkflow.nodes.map((node) => ({
id: node.id,
name: node.name,
type: node.type,
typeVersion: node.typeVersion,
position: node.position,
parameters: node.parameters as Record<string, unknown> | undefined,
credentials: node.credentials,
disabled: node.disabled,
}));
const updatedConnections = updatedWorkflow.connections as Record<string, unknown>;
// Re-analyze to determine if any nodes still need setup.
// Filter by needsAction to distinguish "render this card" from
// "this still requires user intervention".
const remainingRequests = await analyzeWorkflow(context, input.workflowId);
const pendingRequests = remainingRequests.filter((r) => r.needsAction);
const completedNodes = buildCompletedReport(resumeData.credentials, resumeData.nodeParameters);
// Detect credentials that were applied but failed testing.
const credTestFailures = collectCredentialTestFailures(
remainingRequests,
resumeData.credentials,
);
const credFailedNodeNames = new Set(credTestFailures.map((f) => f.nodeName));
const validCompletedNodes = completedNodes.filter((n) => !credFailedNodeNames.has(n.nodeName));
const allFailedNodes = [...(failedNodes ?? []), ...credTestFailures];
const mergedFailedNodes = allFailedNodes.length > 0 ? allFailedNodes : undefined;
if (pendingRequests.length > 0) {
const skippedNodes = pendingRequests.map((r) => ({
nodeName: r.node.name,
credentialType: r.credentialType,
}));
return {
success: true,
partial: true,
reason: `Applied setup for ${String(validCompletedNodes.length)} node(s), ${String(pendingRequests.length)} node(s) still need configuration.`,
completedNodes: validCompletedNodes,
skippedNodes,
failedNodes: mergedFailedNodes,
updatedNodes,
updatedConnections,
};
}
return {
success: true,
completedNodes: validCompletedNodes,
failedNodes: mergedFailedNodes,
updatedNodes,
updatedConnections,
};
} catch (error) {
return {
success: false,
error: `Workflow apply failed: ${error instanceof Error ? error.message : 'Unknown error'}`,
};
}
}
async function handleSetup(
context: InstanceAiContext,
input: Extract<Input, { action: 'setup' }>,
ctx: WorkflowToolContext,
state: { currentRequestId: string | null; preTestSnapshot: WorkflowJSON | null },
state: SetupState,
) {
// `setup` mutates workflow nodes via applyNodeChanges (credentials and
// parameters are workflow-record fields), so it's gated under
@ -535,155 +712,18 @@ async function handleSetup(
// State 3: Test trigger — persist changes, run, re-suspend with result
if (resumeData.action === 'test-trigger' && resumeData.testTriggerNode) {
state.preTestSnapshot ??= await context.workflowService.getAsWorkflowJSON(input.workflowId);
const preTestApply = await applyNodeChanges(
return await handleSetupTestTrigger(
context,
input.workflowId,
resumeData.credentials,
resumeData.nodeParameters,
input,
ctx,
state,
resumeData,
resumeData.testTriggerNode,
);
const applyFailures = preTestApply.failed;
if (applyFailures.length > 0) {
return {
success: false,
error: `Failed to apply setup before trigger test: ${applyFailures.map((f) => `${f.nodeName}: ${f.error}`).join('; ')}`,
failedNodes: applyFailures,
};
}
let triggerTestResult: {
status: 'success' | 'error' | 'listening';
error?: string;
};
try {
const result = await context.executionService.run(input.workflowId, undefined, {
timeout: 30_000,
triggerNodeName: resumeData.testTriggerNode,
});
if (result.status === 'success') {
triggerTestResult = { status: 'success' };
} else if (result.status === 'waiting') {
triggerTestResult = { status: 'listening' as const };
} else {
triggerTestResult = {
status: 'error',
error: result.error ?? 'Trigger test failed',
};
}
} catch (error) {
triggerTestResult = {
status: 'error',
error: error instanceof Error ? error.message : 'Trigger test failed',
};
}
const refreshedRequests = await analyzeWorkflow(context, input.workflowId, {
[resumeData.testTriggerNode]: triggerTestResult,
});
// Generate a new requestId so the frontend doesn't filter it
// as already-resolved from the previous suspend cycle
state.currentRequestId = nanoid();
return await ctx.suspend({
requestId: state.currentRequestId,
message: 'Configure credentials for your workflow',
severity: 'info' as const,
setupRequests: refreshedRequests,
workflowId: input.workflowId,
...(input.projectId ? { projectId: input.projectId } : {}),
});
}
// State 4: Apply — save credentials and parameters atomically
try {
state.preTestSnapshot = null;
const applyResult = await applyNodeChanges(
context,
input.workflowId,
resumeData.credentials,
resumeData.nodeParameters,
);
const failedNodes = applyResult.failed.length > 0 ? applyResult.failed : undefined;
// Fetch updated workflow to include in response so the frontend can refresh the canvas
const updatedWorkflow = await context.workflowService.getAsWorkflowJSON(input.workflowId);
const updatedNodes = updatedWorkflow.nodes.map((node) => ({
id: node.id,
name: node.name,
type: node.type,
typeVersion: node.typeVersion,
position: node.position,
parameters: node.parameters as Record<string, unknown> | undefined,
credentials: node.credentials,
disabled: node.disabled,
}));
const updatedConnections = updatedWorkflow.connections as Record<string, unknown>;
// Re-analyze to determine if any nodes still need setup.
// Filter by needsAction to distinguish "render this card" from
// "this still requires user intervention".
const remainingRequests = await analyzeWorkflow(context, input.workflowId);
const pendingRequests = remainingRequests.filter((r) => r.needsAction);
const completedNodes = buildCompletedReport(resumeData.credentials, resumeData.nodeParameters);
// Detect credentials that were applied but failed testing.
// Move them from completedNodes to failedNodes so the LLM knows
// the credential is invalid rather than seeing it in both lists.
const credTestFailures: Array<{ nodeName: string; error: string }> = [];
for (const req of remainingRequests) {
if (
req.credentialTestResult &&
!req.credentialTestResult.success &&
req.credentialType &&
resumeData.credentials?.[req.node.name]?.[req.credentialType]
) {
credTestFailures.push({
nodeName: req.node.name,
error: `Credential test failed for ${req.credentialType}: ${req.credentialTestResult.message ?? 'Invalid credentials'}`,
});
}
}
const credFailedNodeNames = new Set(credTestFailures.map((f) => f.nodeName));
const validCompletedNodes = completedNodes.filter((n) => !credFailedNodeNames.has(n.nodeName));
const allFailedNodes = [...(failedNodes ?? []), ...credTestFailures];
const mergedFailedNodes = allFailedNodes.length > 0 ? allFailedNodes : undefined;
if (pendingRequests.length > 0) {
const skippedNodes = pendingRequests.map((r) => ({
nodeName: r.node.name,
credentialType: r.credentialType,
}));
return {
success: true,
partial: true,
reason: `Applied setup for ${String(validCompletedNodes.length)} node(s), ${String(pendingRequests.length)} node(s) still need configuration.`,
completedNodes: validCompletedNodes,
skippedNodes,
failedNodes: mergedFailedNodes,
updatedNodes,
updatedConnections,
};
}
return {
success: true,
completedNodes: validCompletedNodes,
failedNodes: mergedFailedNodes,
updatedNodes,
updatedConnections,
};
} catch (error) {
return {
success: false,
error: `Workflow apply failed: ${error instanceof Error ? error.message : 'Unknown error'}`,
};
}
return await handleSetupApply(context, input, state, resumeData);
}
async function handleValidate(

699
packages/@n8n/instance-ai/src/tools/workflows/setup-workflow.service.ts
View File

@ -154,37 +154,15 @@ export async function stripStaleCredentialsFromWorkflow(
);
}
/**
* Build setup request(s) from a single WorkflowJSON node.
* Detects credential types, auto-selects the most recent credential,
* tests testable credentials, determines trigger eligibility, and
* computes parameter issues with editable parameter definitions.
*/
export async function buildSetupRequests(
type NodeDescription = Awaited<ReturnType<InstanceAiContext['nodeService']['getDescription']>>;
/** Compute parameter issues from the node service, then add placeholder-value issues. */
async function computeParameterIssues(
context: InstanceAiContext,
node: NodeJSON,
triggerTestResult?: { status: 'success' | 'error' | 'listening'; error?: string },
cache?: CredentialCache,
workflowId?: string,
): Promise<SetupRequest[]> {
if (!node.name) return [];
if (node.disabled) return [];
const typeVersion = node.typeVersion ?? 1;
const parameters = (node.parameters as Record<string, unknown>) ?? {};
const nodeDesc = await context.nodeService
.getDescription(node.type, typeVersion)
.catch(() => undefined);
const isTrigger = nodeDesc?.group?.includes('trigger') ?? false;
const isTestable =
isTrigger &&
((nodeDesc?.webhooks !== undefined && nodeDesc.webhooks.length > 0) ||
nodeDesc?.polling === true ||
nodeDesc?.triggerPanel !== undefined);
// Compute parameter issues
parameters: Record<string, unknown>,
typeVersion: number,
): Promise<Record<string, string[]>> {
let parameterIssues: Record<string, string[]> = {};
if (context.nodeService.getParameterIssues) {
parameterIssues = await context.nodeService
@ -204,33 +182,52 @@ export async function buildSetupRequests(
}
}
}
return parameterIssues;
}
// Build editable parameter definitions for parameters that have issues
let editableParameters: SetupRequest['editableParameters'];
if (Object.keys(parameterIssues).length > 0 && nodeDesc?.properties) {
editableParameters = [];
for (const paramName of Object.keys(parameterIssues)) {
const prop = nodeDesc.properties.find((p) => p.name === paramName);
if (!prop) continue;
editableParameters.push({
name: prop.name,
displayName: prop.displayName,
type: prop.type,
...(prop.required !== undefined ? { required: prop.required } : {}),
...(prop.default !== undefined ? { default: prop.default } : {}),
...(prop.options
? {
options: prop.options as SetupRequest['editableParameters'] extends Array<infer T>
? T extends { options?: infer O }
? O
: never
: never,
}
: {}),
});
}
/** Build editable parameter definitions for the parameters that have issues. */
function buildEditableParameters(
parameterIssues: Record<string, string[]>,
nodeDesc: NodeDescription | undefined,
): SetupRequest['editableParameters'] {
if (Object.keys(parameterIssues).length === 0 || !nodeDesc?.properties) return undefined;
const editableParameters: NonNullable<SetupRequest['editableParameters']> = [];
for (const paramName of Object.keys(parameterIssues)) {
const prop = nodeDesc.properties.find((p) => p.name === paramName);
if (!prop) continue;
editableParameters.push({
name: prop.name,
displayName: prop.displayName,
type: prop.type,
...(prop.required !== undefined ? { required: prop.required } : {}),
...(prop.default !== undefined ? { default: prop.default } : {}),
...(prop.options
? {
options: prop.options as SetupRequest['editableParameters'] extends Array<infer T>
? T extends { options?: infer O }
? O
: never
: never,
}
: {}),
});
}
return editableParameters;
}
/**
* Resolve the credential types valid for a node: dynamic resolver first, then
* the description's static credentials filtered by displayOptions, then the
* dynamic types implied by `authentication: generic/predefinedCredentialType`.
*/
async function resolveCredentialTypes(
context: InstanceAiContext,
node: NodeJSON,
parameters: Record<string, unknown>,
typeVersion: number,
nodeDesc: NodeDescription | undefined,
): Promise<string[]> {
let credentialTypes: string[] = [];
if (context.nodeService.getNodeCredentialTypes) {
credentialTypes = await context.nodeService
@ -280,145 +277,284 @@ export async function buildSetupRequests(
}
}
return credentialTypes;
}
interface CredentialState {
existingCredentials: Array<{ id: string; name: string }>;
isAutoApplied: boolean;
credentialTestResult?: { success: boolean; message?: string };
}
/**
* For a single credential type, list existing credentials (cached + workflow-scoped),
* decide whether to auto-apply the sole candidate, and test the resolved credential.
*/
async function resolveCredentialState(
context: InstanceAiContext,
node: NodeJSON,
credentialType: string,
cache: CredentialCache | undefined,
workflowId: string | undefined,
): Promise<CredentialState> {
// Use cache to avoid duplicate fetches for the same credential type across nodes.
// Scope to the workflow so we list only credentials the save path will accept —
// the editor's credential picker uses the same scoping. The cache key includes
// workflowId so a cache shared across workflows stays correct.
const cacheKey = listCacheKey(workflowId, credentialType);
let listPromise = cache?.lists.get(cacheKey);
if (!listPromise) {
listPromise = context.credentialService
.list({ type: credentialType, ...(workflowId ? { workflowId } : {}) })
.then((creds) => creds.map((c) => ({ id: c.id, name: c.name })));
cache?.lists.set(cacheKey, listPromise);
}
const sortedCreds = await listPromise;
const existingCredentials = sortedCreds.map((c) => ({ id: c.id, name: c.name }));
const existingOnNode = node.credentials?.[credentialType];
// Only auto-apply when there is exactly one candidate. With multiple
// candidates, picking the first is a silent guess — surface the list
// so the setup wizard can prompt the user to choose.
const isAutoApplied = !existingOnNode?.id && existingCredentials.length === 1;
const credToTest = existingOnNode?.id ?? (isAutoApplied ? existingCredentials[0]?.id : undefined);
if (!credToTest) return { existingCredentials, isAutoApplied };
let testabilityPromise = cache?.testability.get(credentialType);
if (!testabilityPromise) {
testabilityPromise = context.credentialService.isTestable
? context.credentialService.isTestable(credentialType).catch(() => true)
: Promise.resolve(true);
cache?.testability.set(credentialType, testabilityPromise);
}
const canTest = await testabilityPromise;
if (!canTest) return { existingCredentials, isAutoApplied };
let testPromise = cache?.tests.get(credToTest);
if (!testPromise) {
testPromise = context.credentialService.test(credToTest).catch((testError) => ({
success: false,
message: testError instanceof Error ? testError.message : 'Test failed',
}));
cache?.tests.set(credToTest, testPromise);
}
const credentialTestResult = await testPromise;
return { existingCredentials, isAutoApplied, credentialTestResult };
}
type RequestNodeCredentials = NonNullable<SetupRequest['node']['credentials']>;
/** Build the optional `credentials` slice of a setup request's node, merging an auto-applied credential. */
function buildRequestCredentials(
nodeCredentials: Record<string, { id: string; name?: string }> | undefined,
isAutoApplied: boolean,
credentialType: string | undefined,
existingCredentials: Array<{ id: string; name: string }>,
): { credentials?: RequestNodeCredentials } {
const autoCredential =
isAutoApplied && credentialType && existingCredentials.length > 0
? { [credentialType]: { id: existingCredentials[0].id, name: existingCredentials[0].name } }
: undefined;
if (nodeCredentials && Object.keys(nodeCredentials).length > 0) {
return {
credentials: (autoCredential
? { ...nodeCredentials, ...autoCredential }
: nodeCredentials) as RequestNodeCredentials,
};
}
return autoCredential ? { credentials: autoCredential as RequestNodeCredentials } : {};
}
/**
* Build setup request(s) from a single WorkflowJSON node.
* Detects credential types, auto-selects the most recent credential,
* tests testable credentials, determines trigger eligibility, and
* computes parameter issues with editable parameter definitions.
*/
/** Resolve credential state for a type and auto-apply the sole candidate onto nodeCredentials. */
async function resolveAppliedCredentialState(
context: InstanceAiContext,
node: NodeJSON,
credentialType: string | undefined,
cache: CredentialCache | undefined,
workflowId: string | undefined,
nodeCredentials: Record<string, { id: string; name?: string }> | undefined,
): Promise<CredentialState> {
if (!credentialType) {
return { existingCredentials: [], isAutoApplied: false };
}
const state = await resolveCredentialState(context, node, credentialType, cache, workflowId);
if (state.isAutoApplied && nodeCredentials) {
nodeCredentials[credentialType] = {
id: state.existingCredentials[0].id,
name: state.existingCredentials[0].name,
};
}
return state;
}
interface NodeSetupContext {
nodeName: string;
isTrigger: boolean;
isTestable: boolean;
hasParamIssues: boolean;
parameterIssues: Record<string, string[]>;
editableParameters: SetupRequest['editableParameters'];
triggerTestResult?: { status: 'success' | 'error' | 'listening'; error?: string };
nodeId: string;
nodePosition: [number, number];
typeVersion: number;
parameters: Record<string, unknown>;
}
/**
* Build a single setup request for one (optional) credential type: resolve and
* auto-apply credentials, decide whether user action is still needed, and assemble
* the request. Returns null when the request carries nothing actionable.
*/
async function buildRequestForCredentialType(
context: InstanceAiContext,
node: NodeJSON,
credentialType: string | undefined,
cache: CredentialCache | undefined,
workflowId: string | undefined,
nodeCtx: NodeSetupContext,
): Promise<SetupRequest | null> {
const nodeCredentials = node.credentials
? Object.fromEntries(
Object.entries(node.credentials)
.filter(([, v]) => v.id !== undefined)
.map(([k, v]) => [k, { id: v.id!, name: v.name }]),
)
: undefined;
const { existingCredentials, isAutoApplied, credentialTestResult } =
await resolveAppliedCredentialState(
context,
node,
credentialType,
cache,
workflowId,
nodeCredentials,
);
const { isTrigger, isTestable, hasParamIssues } = nodeCtx;
if (!credentialType && !isTrigger && !hasParamIssues) return null;
if (!credentialType && isTrigger && !isTestable && !hasParamIssues) return null;
// Determine whether this request still needs user intervention.
// A credential request needs action if no credential is set or the test failed.
// A parameter request needs action if issues remain.
// A trigger-only request (no credential, no param issues) never blocks apply.
let needsAction = false;
if (credentialType) {
const existingOnNode = node.credentials?.[credentialType];
const hasValidCredential =
existingOnNode?.id !== undefined &&
(credentialTestResult === undefined || credentialTestResult.success);
needsAction = !hasValidCredential;
}
if (hasParamIssues) {
needsAction = true;
}
return {
node: {
name: nodeCtx.nodeName,
type: node.type,
typeVersion: nodeCtx.typeVersion,
parameters: nodeCtx.parameters,
position: nodeCtx.nodePosition,
id: nodeCtx.nodeId,
...buildRequestCredentials(
nodeCredentials,
isAutoApplied,
credentialType,
existingCredentials,
),
},
...(credentialType ? { credentialType } : {}),
...(existingCredentials.length > 0 ? { existingCredentials } : {}),
isTrigger,
...(isTestable ? { isTestable } : {}),
...(isAutoApplied ? { isAutoApplied } : {}),
...(credentialTestResult ? { credentialTestResult } : {}),
...(nodeCtx.triggerTestResult ? { triggerTestResult: nodeCtx.triggerTestResult } : {}),
...(hasParamIssues ? { parameterIssues: nodeCtx.parameterIssues } : {}),
...(nodeCtx.editableParameters && nodeCtx.editableParameters.length > 0
? { editableParameters: nodeCtx.editableParameters }
: {}),
needsAction,
};
}
export async function buildSetupRequests(
context: InstanceAiContext,
node: NodeJSON,
triggerTestResult?: { status: 'success' | 'error' | 'listening'; error?: string },
cache?: CredentialCache,
workflowId?: string,
): Promise<SetupRequest[]> {
if (!node.name) return [];
if (node.disabled) return [];
const typeVersion = node.typeVersion ?? 1;
const parameters = (node.parameters as Record<string, unknown>) ?? {};
const nodeDesc = await context.nodeService
.getDescription(node.type, typeVersion)
.catch(() => undefined);
const isTrigger = nodeDesc?.group?.includes('trigger') ?? false;
const isTestable =
isTrigger &&
((nodeDesc?.webhooks !== undefined && nodeDesc.webhooks.length > 0) ||
nodeDesc?.polling === true ||
nodeDesc?.triggerPanel !== undefined);
const parameterIssues = await computeParameterIssues(context, node, parameters, typeVersion);
const editableParameters = buildEditableParameters(parameterIssues, nodeDesc);
const credentialTypes = await resolveCredentialTypes(
context,
node,
parameters,
typeVersion,
nodeDesc,
);
const nodeId = node.id ?? nanoid();
const nodePosition: [number, number] = node.position ?? [0, 0];
const hasParamIssues = Object.keys(parameterIssues).length > 0;
const requests: SetupRequest[] = [];
const processedCredTypes = credentialTypes.length > 0 ? credentialTypes : [undefined];
const nodeCtx: NodeSetupContext = {
nodeName: node.name,
isTrigger,
isTestable,
hasParamIssues,
parameterIssues,
editableParameters,
triggerTestResult,
nodeId,
nodePosition,
typeVersion,
parameters,
};
for (const credentialType of processedCredTypes) {
let existingCredentials: Array<{ id: string; name: string }> = [];
let isAutoApplied = false;
let credentialTestResult: { success: boolean; message?: string } | undefined;
const nodeCredentials = node.credentials
? Object.fromEntries(
Object.entries(node.credentials)
.filter(([, v]) => v.id !== undefined)
.map(([k, v]) => [k, { id: v.id!, name: v.name }]),
)
: undefined;
if (credentialType) {
// Use cache to avoid duplicate fetches for the same credential type across nodes.
// Scope to the workflow so we list only credentials the save path will accept —
// the editor's credential picker uses the same scoping. The cache key includes
// workflowId so a cache shared across workflows stays correct.
const cacheKey = listCacheKey(workflowId, credentialType);
let listPromise = cache?.lists.get(cacheKey);
if (!listPromise) {
listPromise = context.credentialService
.list({ type: credentialType, ...(workflowId ? { workflowId } : {}) })
.then((creds) => creds.map((c) => ({ id: c.id, name: c.name })));
cache?.lists.set(cacheKey, listPromise);
}
const sortedCreds = await listPromise;
existingCredentials = sortedCreds.map((c) => ({ id: c.id, name: c.name }));
const existingOnNode = node.credentials?.[credentialType];
// Only auto-apply when there is exactly one candidate. With multiple
// candidates, picking the first is a silent guess — surface the list
// so the setup wizard can prompt the user to choose.
if (!existingOnNode?.id && existingCredentials.length === 1) {
isAutoApplied = true;
if (nodeCredentials) {
nodeCredentials[credentialType] = {
id: existingCredentials[0].id,
name: existingCredentials[0].name,
};
}
}
const credToTest =
existingOnNode?.id ?? (isAutoApplied ? existingCredentials[0]?.id : undefined);
if (credToTest) {
let testabilityPromise = cache?.testability.get(credentialType);
if (!testabilityPromise) {
testabilityPromise = context.credentialService.isTestable
? context.credentialService.isTestable(credentialType).catch(() => true)
: Promise.resolve(true);
cache?.testability.set(credentialType, testabilityPromise);
}
const canTest = await testabilityPromise;
if (canTest) {
let testPromise = cache?.tests.get(credToTest);
if (!testPromise) {
testPromise = context.credentialService.test(credToTest).catch((testError) => ({
success: false,
message: testError instanceof Error ? testError.message : 'Test failed',
}));
cache?.tests.set(credToTest, testPromise);
}
credentialTestResult = await testPromise;
}
}
}
if (!credentialType && !isTrigger && !hasParamIssues) continue;
if (!credentialType && isTrigger && !isTestable && !hasParamIssues) continue;
// Determine whether this request still needs user intervention.
// A credential request needs action if no credential is set or the test failed.
// A parameter request needs action if issues remain.
// A trigger-only request (no credential, no param issues) never blocks apply.
let needsAction = false;
if (credentialType) {
const existingOnNode = node.credentials?.[credentialType];
const hasValidCredential =
existingOnNode?.id !== undefined &&
(credentialTestResult === undefined || credentialTestResult.success);
needsAction = !hasValidCredential;
}
if (hasParamIssues) {
needsAction = true;
}
const request: SetupRequest = {
node: {
name: node.name,
type: node.type,
typeVersion,
parameters,
position: nodePosition,
id: nodeId,
...(nodeCredentials && Object.keys(nodeCredentials).length > 0
? {
credentials:
isAutoApplied && credentialType && existingCredentials.length > 0
? {
...nodeCredentials,
[credentialType]: {
id: existingCredentials[0].id,
name: existingCredentials[0].name,
},
}
: nodeCredentials,
}
: isAutoApplied && credentialType && existingCredentials.length > 0
? {
credentials: {
[credentialType]: {
id: existingCredentials[0].id,
name: existingCredentials[0].name,
},
},
}
: {}),
},
...(credentialType ? { credentialType } : {}),
...(existingCredentials.length > 0 ? { existingCredentials } : {}),
isTrigger,
...(isTestable ? { isTestable } : {}),
...(isAutoApplied ? { isAutoApplied } : {}),
...(credentialTestResult ? { credentialTestResult } : {}),
...(triggerTestResult ? { triggerTestResult } : {}),
...(hasParamIssues ? { parameterIssues } : {}),
...(editableParameters && editableParameters.length > 0 ? { editableParameters } : {}),
needsAction,
};
requests.push(request);
const request = await buildRequestForCredentialType(
context,
node,
credentialType,
cache,
workflowId,
nodeCtx,
);
if (request) requests.push(request);
}
return requests;
@ -433,41 +569,75 @@ export async function buildSetupRequests(
* Algorithm: DFS from each trigger (sorted left-to-right by X position),
* following outgoing connections. Nodes not reachable from any trigger go last.
*/
export function sortByExecutionOrder(
requests: SetupRequest[],
connections: Record<string, unknown>,
function addConnectionEdge(
mainOutgoing: Map<string, string[]>,
nonMainIncoming: Map<string, string[]>,
sourceName: string,
connType: string,
conn: unknown,
): void {
// Build main outgoing adjacency (source -> destinations via 'main' outputs)
if (typeof conn !== 'object' || conn === null || !('node' in conn)) return;
const destName = (conn as { node: string }).node;
if (connType === 'main') {
const existing = mainOutgoing.get(sourceName) ?? [];
if (!existing.includes(destName)) existing.push(destName);
mainOutgoing.set(sourceName, existing);
} else {
// Non-main connection: source is an AI sub-node of destination
const existing = nonMainIncoming.get(destName) ?? [];
if (!existing.includes(sourceName)) existing.push(sourceName);
nonMainIncoming.set(destName, existing);
}
}
function addSourceConnections(
sourceName: string,
nodeConns: Record<string, unknown>,
mainOutgoing: Map<string, string[]>,
nonMainIncoming: Map<string, string[]>,
): void {
for (const [connType, outputs] of Object.entries(nodeConns)) {
if (!Array.isArray(outputs)) continue;
for (const slot of outputs) {
if (!Array.isArray(slot)) continue;
for (const conn of slot) {
addConnectionEdge(mainOutgoing, nonMainIncoming, sourceName, connType, conn);
}
}
}
}
/**
* Build adjacency maps from workflow connections: main outgoing (source →
* destinations) and non-main incoming (destination → AI sub-node sources).
*/
function buildConnectionAdjacency(connections: Record<string, unknown>): {
mainOutgoing: Map<string, string[]>;
nonMainIncoming: Map<string, string[]>;
} {
const mainOutgoing = new Map<string, string[]>();
// Build non-main incoming adjacency (destination -> sources via non-main inputs)
// Non-main connections represent AI sub-nodes (tools, memory, models) attached to agent nodes
const nonMainIncoming = new Map<string, string[]>();
for (const [sourceName, nodeConns] of Object.entries(connections)) {
if (typeof nodeConns !== 'object' || nodeConns === null) continue;
for (const [connType, outputs] of Object.entries(nodeConns as Record<string, unknown>)) {
if (!Array.isArray(outputs)) continue;
for (const slot of outputs) {
if (!Array.isArray(slot)) continue;
for (const conn of slot) {
if (typeof conn !== 'object' || conn === null || !('node' in conn)) continue;
const destName = (conn as { node: string }).node;
if (connType === 'main') {
const existing = mainOutgoing.get(sourceName) ?? [];
if (!existing.includes(destName)) existing.push(destName);
mainOutgoing.set(sourceName, existing);
} else {
// Non-main connection: source is an AI sub-node of destination
const existing = nonMainIncoming.get(destName) ?? [];
if (!existing.includes(sourceName)) existing.push(sourceName);
nonMainIncoming.set(destName, existing);
}
}
}
}
addSourceConnections(
sourceName,
nodeConns as Record<string, unknown>,
mainOutgoing,
nonMainIncoming,
);
}
return { mainOutgoing, nonMainIncoming };
}
export function sortByExecutionOrder(
requests: SetupRequest[],
connections: Record<string, unknown>,
): void {
const { mainOutgoing, nonMainIncoming } = buildConnectionAdjacency(connections);
const triggerRequests = requests
.filter((r) => r.isTrigger)
.sort((a, b) => a.node.position[0] - b.node.position[0]);
@ -622,6 +792,63 @@ export async function applyNodeParameters(
return result;
}
/** Resolve and apply each credential in credsMap onto the node; returns whether all succeeded. */
async function applyCredentialsToNode(
context: InstanceAiContext,
node: NodeJSON,
nodeName: string,
credsMap: Record<string, string>,
result: ApplyResult,
): Promise<boolean> {
let nodeSucceeded = true;
for (const [credType, credId] of Object.entries(credsMap)) {
try {
const cred = await context.credentialService.get(credId);
if (cred) {
node.credentials = {
...node.credentials,
[credType]: { id: cred.id, name: cred.name },
};
} else {
nodeSucceeded = false;
result.failed.push({
nodeName,
error: `Credential ${credId} (type: ${credType}) not found — it may have been deleted`,
});
}
} catch (error) {
nodeSucceeded = false;
result.failed.push({
nodeName,
error: `Failed to resolve credential ${credId} (type: ${credType}): ${error instanceof Error ? error.message : 'Unknown error'}`,
});
}
}
return nodeSucceeded;
}
/** Merge params into the node's parameters; returns whether it succeeded. */
function applyParametersToNode(
node: NodeJSON,
nodeName: string,
params: Record<string, unknown>,
result: ApplyResult,
): boolean {
try {
node.parameters = {
...(node.parameters ?? {}),
...params,
} as IDataObject;
return true;
} catch (error) {
result.failed.push({
nodeName,
error: `Failed to merge parameters: ${error instanceof Error ? error.message : 'Unknown error'}`,
});
return false;
}
}
/**
* Atomically apply both credentials and parameters to a workflow in a single
* load-mutate-save cycle, avoiding partial-success overwrite windows.
@ -638,52 +865,16 @@ export async function applyNodeChanges(
for (const node of workflowJson.nodes) {
if (!node.name) continue;
const nodeName = node.name;
// Apply credentials
const credsMap = nodeCredentials?.[node.name];
if (credsMap) {
let nodeSucceeded = true;
for (const [credType, credId] of Object.entries(credsMap)) {
try {
const cred = await context.credentialService.get(credId);
if (cred) {
node.credentials = {
...node.credentials,
[credType]: { id: cred.id, name: cred.name },
};
} else {
nodeSucceeded = false;
result.failed.push({
nodeName: node.name,
error: `Credential ${credId} (type: ${credType}) not found — it may have been deleted`,
});
}
} catch (error) {
nodeSucceeded = false;
result.failed.push({
nodeName: node.name,
error: `Failed to resolve credential ${credId} (type: ${credType}): ${error instanceof Error ? error.message : 'Unknown error'}`,
});
}
}
if (nodeSucceeded) appliedNodes.add(node.name);
const credsMap = nodeCredentials?.[nodeName];
if (credsMap && (await applyCredentialsToNode(context, node, nodeName, credsMap, result))) {
appliedNodes.add(nodeName);
}
// Apply parameters
const params = nodeParameters?.[node.name];
if (params) {
try {
node.parameters = {
...(node.parameters ?? {}),
...params,
} as IDataObject;
appliedNodes.add(node.name);
} catch (error) {
result.failed.push({
nodeName: node.name,
error: `Failed to merge parameters: ${error instanceof Error ? error.message : 'Unknown error'}`,
});
}
const params = nodeParameters?.[nodeName];
if (params && applyParametersToNode(node, nodeName, params, result)) {
appliedNodes.add(nodeName);
}
// Drop credential entries that are no longer valid for the node's current

271
packages/@n8n/instance-ai/src/tools/workflows/validate-workflow.service.ts
View File

@ -127,18 +127,17 @@ function doNotExistIssue(
* already references — the editor uses this to suppress the "doesn't exist"
* issue in shared-workflow proxy-auth scenarios. Empty Set in Phase 1.
*/
async function computeCredentialIssues(
type CredentialDescription = NonNullable<NodeDescription['credentials']>[number];
/** HTTP Request generic-auth and proxy-auth special cases (credential type lives in parameters). */
async function computeHttpAuthCredentialIssue(
context: InstanceAiContext,
cache: CredentialLookupCache,
node: NodeJSON,
nodeDesc: NodeDescription,
usedCredentialIds: Set<string>,
): Promise<INodeIssues | null> {
if (node.disabled) return null;
if (!nodeDesc.credentials || nodeDesc.credentials.length === 0) return null;
const parameters = (node.parameters ?? {}) as Record<string, unknown>;
const typeVersion = node.typeVersion ?? 1;
const authentication = parameters.authentication;
const genericAuthType = parameters.genericAuthType;
const nodeCredentialType = parameters.nodeCredentialType;
@ -153,14 +152,14 @@ async function computeCredentialIssues(
return { credentials: notSetIssue(genericAuthType, nodeDesc.displayName) };
}
// HTTP Request v2 proxy-auth predefined-credential paths.
if (
const isPredefinedProxyAuth =
hasProxyAuth(node) &&
authentication === 'predefinedCredentialType' &&
typeof nodeCredentialType === 'string' &&
nodeCredentialType !== '' &&
node.credentials !== undefined
) {
nodeCredentialType !== '';
// HTTP Request v2 proxy-auth predefined-credential paths.
if (isPredefinedProxyAuth && node.credentials !== undefined) {
const stored = await listCredentialsByType(context, cache, nodeCredentialType);
const selectedId = node.credentials[nodeCredentialType]?.id;
const isCredentialUsedInWorkflow =
@ -171,87 +170,121 @@ async function computeCredentialIssues(
}
}
if (
hasProxyAuth(node) &&
authentication === 'predefinedCredentialType' &&
typeof nodeCredentialType === 'string' &&
nodeCredentialType !== '' &&
selectedCredsAreUnusable(node, nodeCredentialType)
) {
if (isPredefinedProxyAuth && selectedCredsAreUnusable(node, nodeCredentialType)) {
return { credentials: notSetIssue(nodeCredentialType, nodeDesc.displayName) };
}
return null;
}
/** Evaluate one credential-type entry of a node description; returns its issue(s) or null. */
async function evaluateCredentialEntry(
context: InstanceAiContext,
cache: CredentialLookupCache,
node: NodeJSON,
nodeDesc: NodeDescription,
credentialTypeDescription: CredentialDescription,
parameters: Record<string, unknown>,
typeVersion: number,
usedCredentialIds: Set<string>,
): Promise<INodeIssueObjectProperty | null> {
if (credentialTypeDescription.displayOptions) {
const visible = matchesDisplayOptions(
{ parameters, nodeVersion: typeVersion },
credentialTypeDescription.displayOptions as DisplayOptions,
);
if (!visible) return null;
}
const credName = credentialTypeDescription.name;
// Without a credential type display name registry on the backend, fall
// back to the type name itself — same fallback the frontend uses when
// `getCredentialTypeByName` returns undefined.
const credentialTypeDisplayName = credName;
const selected = node.credentials?.[credName];
if (!selected) {
return credentialTypeDescription.required ? notSetIssue(credName, nodeDesc.displayName) : null;
}
// AI-gateway-managed credentials have no real DB record — treat as configured.
if (
typeof selected === 'object' &&
'__aiGatewayManaged' in selected &&
(selected as { __aiGatewayManaged?: boolean }).__aiGatewayManaged === true
) {
return null;
}
const selectedRef =
typeof selected === 'string'
? { id: null, name: selected }
: { id: selected.id ?? null, name: selected.name };
const userCredentials = await listCredentialsByType(context, cache, credName);
if (selectedRef.id) {
const idMatch = userCredentials.find((c) => c.id === selectedRef.id);
if (idMatch) return null;
}
const nameMatches = userCredentials.filter((c) => c.name === selectedRef.name);
if (nameMatches.length > 1) {
return notIdentifiedIssue(credName, selectedRef.name, credentialTypeDisplayName);
}
if (nameMatches.length === 0) {
// Frontend suppresses this when the credential ID is in `usedCredentials`
// or when the user has global `credential:read` scope. Phase 1 ports the
// usedCredentials half (passed in as a Set) and conservatively assumes
// the user lacks the global scope — matches the most common case where
// the agent is verifying a workflow owned by the user.
const isUsedInWorkflow =
typeof selectedRef.id === 'string' && usedCredentialIds.has(selectedRef.id);
if (!isUsedInWorkflow) {
return doNotExistIssue(credName, selectedRef.name, credentialTypeDisplayName);
}
}
return null;
}
async function computeCredentialIssues(
context: InstanceAiContext,
cache: CredentialLookupCache,
node: NodeJSON,
nodeDesc: NodeDescription,
usedCredentialIds: Set<string>,
): Promise<INodeIssues | null> {
if (node.disabled) return null;
if (!nodeDesc.credentials || nodeDesc.credentials.length === 0) return null;
const httpAuthIssue = await computeHttpAuthCredentialIssue(
context,
cache,
node,
nodeDesc,
usedCredentialIds,
);
if (httpAuthIssue) return httpAuthIssue;
const parameters = (node.parameters ?? {}) as Record<string, unknown>;
const typeVersion = node.typeVersion ?? 1;
const foundIssues: INodeIssueObjectProperty = {};
for (const credentialTypeDescription of nodeDesc.credentials) {
if (credentialTypeDescription.displayOptions) {
const visible = matchesDisplayOptions(
{ parameters, nodeVersion: typeVersion },
credentialTypeDescription.displayOptions as DisplayOptions,
);
if (!visible) continue;
}
const credName = credentialTypeDescription.name;
// Without a credential type display name registry on the backend, fall
// back to the type name itself — same fallback the frontend uses when
// `getCredentialTypeByName` returns undefined.
const credentialTypeDisplayName = credName;
const selected = node.credentials?.[credName];
if (!selected) {
if (credentialTypeDescription.required) {
Object.assign(foundIssues, notSetIssue(credName, nodeDesc.displayName));
}
continue;
}
// AI-gateway-managed credentials have no real DB record — treat as configured.
if (
typeof selected === 'object' &&
'__aiGatewayManaged' in selected &&
(selected as { __aiGatewayManaged?: boolean }).__aiGatewayManaged === true
) {
continue;
}
const selectedRef =
typeof selected === 'string'
? { id: null, name: selected }
: { id: selected.id ?? null, name: selected.name };
const userCredentials = await listCredentialsByType(context, cache, credName);
if (selectedRef.id) {
const idMatch = userCredentials.find((c) => c.id === selectedRef.id);
if (idMatch) continue;
}
const nameMatches = userCredentials.filter((c) => c.name === selectedRef.name);
if (nameMatches.length > 1) {
Object.assign(
foundIssues,
notIdentifiedIssue(credName, selectedRef.name, credentialTypeDisplayName),
);
continue;
}
if (nameMatches.length === 0) {
// Frontend suppresses this when the credential ID is in `usedCredentials`
// or when the user has global `credential:read` scope. Phase 1 ports the
// usedCredentials half (passed in as a Set) and conservatively assumes
// the user lacks the global scope — matches the most common case where
// the agent is verifying a workflow owned by the user.
const isUsedInWorkflow =
typeof selectedRef.id === 'string' && usedCredentialIds.has(selectedRef.id);
if (!isUsedInWorkflow) {
Object.assign(
foundIssues,
doNotExistIssue(credName, selectedRef.name, credentialTypeDisplayName),
);
}
}
const entry = await evaluateCredentialEntry(
context,
cache,
node,
nodeDesc,
credentialTypeDescription,
parameters,
typeVersion,
usedCredentialIds,
);
if (entry) Object.assign(foundIssues, entry);
}
if (Object.keys(foundIssues).length === 0) return null;
@ -318,6 +351,40 @@ function getFirstExecutionError(taskData: ITaskData[] | undefined): string | nul
return null;
}
async function computeNodeParameterIssues(
context: InstanceAiContext,
node: NodeJSON,
typeVersion: number,
parameters: Record<string, unknown>,
): Promise<Record<string, string[]> | null> {
if (!context.nodeService.getParameterIssues) return null;
const parameterIssues = await context.nodeService
.getParameterIssues(node.type, typeVersion, parameters)
.catch(() => ({}) as Record<string, string[]>);
return Object.keys(parameterIssues).length > 0 ? parameterIssues : null;
}
function computeExecutionIssue(
issues: INodeIssues | null,
nodeName: string,
taskData: ITaskData[] | undefined,
executionErrors: Record<string, string>,
allowSendingParameterValues: boolean,
): INodeIssues | null {
const errorMessage = getFirstExecutionError(taskData);
if (errorMessage === null) return issues;
const next = issues ?? {};
next.execution = true;
// Only retain the message text when the instance permits sending
// parameter-derived strings to the LLM. Error messages can embed
// parameter values (URLs, headers, payloads), so skip the detail in
// the summary when restricted; the `execution: true` flag still flows.
if (allowSendingParameterValues) {
executionErrors[nodeName] = errorMessage;
}
return next;
}
async function computeNodeIssues(
context: InstanceAiContext,
cache: CredentialLookupCache,
@ -347,11 +414,14 @@ async function computeNodeIssues(
let issues: INodeIssues | null = null;
if (!ignoreIssues.has('parameters') && context.nodeService.getParameterIssues) {
const parameterIssues = await context.nodeService
.getParameterIssues(node.type, typeVersion, parameters)
.catch(() => ({}) as Record<string, string[]>);
if (Object.keys(parameterIssues).length > 0) {
if (!ignoreIssues.has('parameters')) {
const parameterIssues = await computeNodeParameterIssues(
context,
node,
typeVersion,
parameters,
);
if (parameterIssues) {
issues = { parameters: parameterIssues };
}
}
@ -390,18 +460,13 @@ async function computeNodeIssues(
}
if (!ignoreIssues.has('execution') && latestRunData) {
const errorMessage = getFirstExecutionError(latestRunData[node.name]);
if (errorMessage !== null) {
issues = issues ?? {};
issues.execution = true;
// Only retain the message text when the instance permits sending
// parameter-derived strings to the LLM. Error messages can embed
// parameter values (URLs, headers, payloads), so skip the detail in
// the summary when restricted; the `execution: true` flag still flows.
if (allowSendingParameterValues) {
executionErrors[node.name] = errorMessage;
}
}
issues = computeExecutionIssue(
issues,
node.name,
latestRunData[node.name],
executionErrors,
allowSendingParameterValues,
);
}
return issues;

271
packages/@n8n/instance-ai/src/tracing/trace-payloads.ts
View File

@ -308,6 +308,33 @@ function stringifyToolPayload(value: unknown): string {
}
}
function extractAssistantToolCall(
part: Record<string, unknown>,
): Record<string, unknown> | undefined {
const toolCallId =
typeof part.toolCallId === 'string'
? part.toolCallId
: typeof part.id === 'string'
? part.id
: undefined;
const toolName =
typeof part.toolName === 'string'
? part.toolName
: typeof part.name === 'string'
? part.name
: undefined;
if (!toolCallId || !toolName) return undefined;
return {
id: toolCallId,
type: 'function',
function: {
name: toolName,
arguments: stringifyToolPayload(readToolCallPayload(part)),
},
};
}
function normalizeAssistantMessageForLangSmith(message: Record<string, unknown>): unknown {
const content = message.content;
if (!Array.isArray(content)) {
@ -327,28 +354,8 @@ function normalizeAssistantMessageForLangSmith(message: Record<string, unknown>)
if (part.type !== 'tool-call') continue;
const toolCallId =
typeof part.toolCallId === 'string'
? part.toolCallId
: typeof part.id === 'string'
? part.id
: undefined;
const toolName =
typeof part.toolName === 'string'
? part.toolName
: typeof part.name === 'string'
? part.name
: undefined;
if (!toolCallId || !toolName) continue;
toolCalls.push({
id: toolCallId,
type: 'function',
function: {
name: toolName,
arguments: stringifyToolPayload(readToolCallPayload(part)),
},
});
const toolCall = extractAssistantToolCall(part);
if (toolCall) toolCalls.push(toolCall);
}
if (toolCalls.length === 0) {
@ -603,6 +610,56 @@ function calculateInputTokenAccounting(
};
}
function readCacheReadTokens(
attributes: Record<string, unknown>,
providerAnthropicUsage: Record<string, unknown>,
inputDetails: unknown,
): number {
return (
firstNumberAttribute(attributes, [
'ai.usage.inputTokenDetails.cacheReadTokens',
'ai.usage.cachedInputTokens',
'ai.usage.cacheReadInputTokens',
]) ??
firstNumberAttribute(providerAnthropicUsage, ['cache_read_input_tokens']) ??
readTokenDetail(inputDetails, [
'cache_read',
'cache_read_tokens',
'cache_read_input_tokens',
'cached_tokens',
]) ??
0
);
}
function readCacheCreationTokens(
attributes: Record<string, unknown>,
providerAnthropicUsage: Record<string, unknown>,
inputDetails: unknown,
): number {
return (
firstNumberAttribute(attributes, [
'ai.usage.inputTokenDetails.cacheCreationTokens',
'ai.usage.cacheCreationInputTokens',
'ai.usage.inputTokenDetails.cacheWriteTokens',
'ai.usage.cacheWriteInputTokens',
]) ??
firstNumberAttribute(providerAnthropicUsage, [
'cache_creation_input_tokens',
'cache_write_input_tokens',
]) ??
readTokenDetail(inputDetails, [
'cache_creation',
'cache_creation_tokens',
'cache_creation_input_tokens',
'cache_write',
'cache_write_tokens',
'cache_write_input_tokens',
]) ??
0
);
}
function buildLangSmithUsageMetadata(
attributes: Record<string, unknown>,
): Record<string, unknown> | undefined {
@ -624,40 +681,12 @@ function buildLangSmithUsageMetadata(
const providerAnthropicUsage = readProviderAnthropicUsage(attributes);
const inputDetails = attributes[GEN_AI_USAGE_INPUT_TOKEN_DETAILS];
const cacheReadTokens =
firstNumberAttribute(attributes, [
'ai.usage.inputTokenDetails.cacheReadTokens',
'ai.usage.cachedInputTokens',
'ai.usage.cacheReadInputTokens',
]) ??
firstNumberAttribute(providerAnthropicUsage, ['cache_read_input_tokens']) ??
readTokenDetail(inputDetails, [
'cache_read',
'cache_read_tokens',
'cache_read_input_tokens',
'cached_tokens',
]) ??
0;
const cacheCreationTokens =
firstNumberAttribute(attributes, [
'ai.usage.inputTokenDetails.cacheCreationTokens',
'ai.usage.cacheCreationInputTokens',
'ai.usage.inputTokenDetails.cacheWriteTokens',
'ai.usage.cacheWriteInputTokens',
]) ??
firstNumberAttribute(providerAnthropicUsage, [
'cache_creation_input_tokens',
'cache_write_input_tokens',
]) ??
readTokenDetail(inputDetails, [
'cache_creation',
'cache_creation_tokens',
'cache_creation_input_tokens',
'cache_write',
'cache_write_tokens',
'cache_write_input_tokens',
]) ??
0;
const cacheReadTokens = readCacheReadTokens(attributes, providerAnthropicUsage, inputDetails);
const cacheCreationTokens = readCacheCreationTokens(
attributes,
providerAnthropicUsage,
inputDetails,
);
const { totalInputTokens } = calculateInputTokenAccounting(
inputTokens,
@ -683,6 +712,51 @@ function buildLangSmithUsageMetadata(
};
}
function applyLangSmithUsageAttributes(
attributes: Record<string, unknown>,
usageMetadata: Record<string, unknown>,
tokens: {
inputTokens: number | undefined;
totalInputTokens: number;
outputTokens: number;
regularInputTokens: number;
cacheReadTokens: number;
cacheCreationTokens: number;
},
): void {
const {
inputTokens,
totalInputTokens,
outputTokens,
regularInputTokens,
cacheReadTokens,
cacheCreationTokens,
} = tokens;
attributes[GEN_AI_USAGE_INPUT_TOKENS] = totalInputTokens;
attributes[GEN_AI_USAGE_OUTPUT_TOKENS] = outputTokens;
attributes[GEN_AI_USAGE_TOTAL_TOKENS] = totalInputTokens + outputTokens;
attributes['ai.usage.inputTokens'] = totalInputTokens;
attributes[LANGSMITH_USAGE_METADATA] = JSON.stringify(usageMetadata);
if (inputTokens !== undefined) {
attributes['langsmith.metadata.original_input_tokens'] = inputTokens;
attributes['langsmith.metadata.anthropic_original_input_tokens'] = inputTokens;
}
attributes['langsmith.metadata.total_input_tokens'] = totalInputTokens;
attributes['langsmith.metadata.regular_input_tokens'] = regularInputTokens;
attributes['langsmith.metadata.cache_read_input_tokens'] = cacheReadTokens;
attributes['langsmith.metadata.cache_creation_input_tokens'] = cacheCreationTokens;
attributes['langsmith.metadata.anthropic_total_input_tokens'] = totalInputTokens;
attributes['langsmith.metadata.anthropic_regular_input_tokens'] = regularInputTokens;
attributes['langsmith.metadata.anthropic_cache_read_input_tokens'] = cacheReadTokens;
attributes['langsmith.metadata.anthropic_cache_creation_input_tokens'] = cacheCreationTokens;
attributes[GEN_AI_USAGE_INPUT_TOKEN_DETAILS] = JSON.stringify({
cache_read: cacheReadTokens,
cache_creation: cacheCreationTokens,
regular: regularInputTokens,
...(inputTokens !== undefined ? { original_input_tokens: inputTokens } : {}),
});
}
function normalizeUsageForLangSmith(attributes: Record<string, unknown>): void {
const usageMetadata = buildLangSmithUsageMetadata(attributes);
if (!usageMetadata) {
@ -710,28 +784,13 @@ function normalizeUsageForLangSmith(attributes: Record<string, unknown>): void {
cacheCreationTokens,
);
attributes[GEN_AI_USAGE_INPUT_TOKENS] = totalInputTokens;
attributes[GEN_AI_USAGE_OUTPUT_TOKENS] = outputTokens;
attributes[GEN_AI_USAGE_TOTAL_TOKENS] = totalInputTokens + outputTokens;
attributes['ai.usage.inputTokens'] = totalInputTokens;
attributes[LANGSMITH_USAGE_METADATA] = JSON.stringify(usageMetadata);
if (inputTokens !== undefined) {
attributes['langsmith.metadata.original_input_tokens'] = inputTokens;
attributes['langsmith.metadata.anthropic_original_input_tokens'] = inputTokens;
}
attributes['langsmith.metadata.total_input_tokens'] = totalInputTokens;
attributes['langsmith.metadata.regular_input_tokens'] = regularInputTokens;
attributes['langsmith.metadata.cache_read_input_tokens'] = cacheReadTokens;
attributes['langsmith.metadata.cache_creation_input_tokens'] = cacheCreationTokens;
attributes['langsmith.metadata.anthropic_total_input_tokens'] = totalInputTokens;
attributes['langsmith.metadata.anthropic_regular_input_tokens'] = regularInputTokens;
attributes['langsmith.metadata.anthropic_cache_read_input_tokens'] = cacheReadTokens;
attributes['langsmith.metadata.anthropic_cache_creation_input_tokens'] = cacheCreationTokens;
attributes[GEN_AI_USAGE_INPUT_TOKEN_DETAILS] = JSON.stringify({
cache_read: cacheReadTokens,
cache_creation: cacheCreationTokens,
regular: regularInputTokens,
...(inputTokens !== undefined ? { original_input_tokens: inputTokens } : {}),
applyLangSmithUsageAttributes(attributes, usageMetadata, {
inputTokens,
totalInputTokens,
outputTokens,
regularInputTokens,
cacheReadTokens,
cacheCreationTokens,
});
}
@ -1172,40 +1231,26 @@ function summarizeRuntimeSkillRegistry(
};
}
const NOT_SCALAR = Symbol('not-scalar');
/** Sanitize scalar/leaf values; returns NOT_SCALAR for arrays/objects the caller must recurse into. */
function sanitizeTraceScalar(value: unknown): unknown {
if (value === null || value === undefined) return value;
if (typeof value === 'string') return truncateString(value);
if (typeof value === 'number' || typeof value === 'boolean') return value;
if (typeof value === 'bigint') return value.toString();
if (typeof value === 'function') return `[function ${value.name || 'anonymous'}]`;
if (value instanceof Date) return value.toISOString();
if (value instanceof Error) return { name: value.name, message: truncateString(value.message) };
if (value instanceof Uint8Array) return `[binary ${value.byteLength} bytes]`;
if (typeof value === 'symbol') return value.toString();
return NOT_SCALAR;
}
export function sanitizeTraceValue(value: unknown, depth = 0): unknown {
if (value === null || value === undefined) {
return value;
}
if (typeof value === 'string') {
return truncateString(value);
}
if (typeof value === 'number' || typeof value === 'boolean') {
return value;
}
if (typeof value === 'bigint') {
return value.toString();
}
if (typeof value === 'function') {
return `[function ${value.name || 'anonymous'}]`;
}
if (value instanceof Date) {
return value.toISOString();
}
if (value instanceof Error) {
return {
name: value.name,
message: truncateString(value.message),
};
}
if (value instanceof Uint8Array) {
return `[binary ${value.byteLength} bytes]`;
const scalar = sanitizeTraceScalar(value);
if (scalar !== NOT_SCALAR) {
return scalar;
}
if (Array.isArray(value)) {
@ -1234,10 +1279,6 @@ export function sanitizeTraceValue(value: unknown, depth = 0): unknown {
return sanitized;
}
if (typeof value === 'symbol') {
return value.toString();
}
return truncateString(Object.prototype.toString.call(value));
}