From 21994c465f54af783b5dec8d2de50c275c291237 Mon Sep 17 00:00:00 2001 From: Albert Alises Date: Fri, 19 Jun 2026 14:22:04 +0200 Subject: [PATCH] refactor(ai-builder): Reduce code complexity below maintainability thresholds (no-changelog) (#32626) --- .../src/agent/sanitize-mcp-schemas.ts | 415 ++++++----- .../src/runtime/resumable-stream-executor.ts | 349 ++++++--- .../@n8n/instance-ai/src/stream/map-chunk.ts | 533 +++++++------ .../src/tools/evals/column-ref-utils.ts | 23 +- .../describe-metric-for-workflow.service.ts | 24 +- .../verify-built-workflow.tool.ts | 613 ++++++++------- .../instance-ai/src/tools/workflows.tool.ts | 330 +++++---- .../tools/workflows/setup-workflow.service.ts | 699 +++++++++++------- .../workflows/validate-workflow.service.ts | 271 ++++--- .../instance-ai/src/tracing/trace-payloads.ts | 271 ++++--- 10 files changed, 2060 insertions(+), 1468 deletions(-) diff --git a/packages/@n8n/instance-ai/src/agent/sanitize-mcp-schemas.ts b/packages/@n8n/instance-ai/src/agent/sanitize-mcp-schemas.ts index 27e3199162c..5190a35e981 100644 --- a/packages/@n8n/instance-ai/src/agent/sanitize-mcp-schemas.ts +++ b/packages/@n8n/instance-ai/src/agent/sanitize-mcp-schemas.ts @@ -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>, + discriminator: string, +): { + actionMeta: Array<{ value: string; description?: string }>; + fieldMeta: Map; +} { + const actionMeta: Array<{ value: string; description?: string }> = []; + const fieldMeta = new Map(); + + 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>>, + context: SanitizeContext, + sanitizeChild: SanitizeChild, +): z.ZodTypeAny { + const discriminator = schema.discriminator; + const variants = [...schema.options.values()] as Array>; + 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>>; - const discriminator = disc.discriminator; - const variants = [...disc.options.values()] as Array>; - 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>>, + 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); } diff --git a/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts b/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts index 7925e3dd4b3..b47d1e35e6d 100644 --- a/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts +++ b/packages/@n8n/instance-ai/src/runtime/resumable-stream-executor.ts @@ -190,6 +190,231 @@ export function normalizeStreamSource(result: unknown): ResumableStreamSource { throw new Error('Unsupported agent stream result'); } +function buildCancelledResult( + agentRunId: string, + text: Promise | 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> } { + 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>; + 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; + activeAgentRunId: string; + options: ExecuteResumableStreamOptions; + workSummaryAccumulator: WorkSummaryAccumulator; + usageAccumulator: UsageAccumulator; + outputRedactor: OutputRedactor; + currentResponseId: string | undefined; + nativeStepIndex: number; +}): Promise { + 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> | 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 { @@ -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> | 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) { diff --git a/packages/@n8n/instance-ai/src/stream/map-chunk.ts b/packages/@n8n/instance-ai/src/stream/map-chunk.ts index 79bf97edff3..bfeffa54659 100644 --- a/packages/@n8n/instance-ai/src/stream/map-chunk.ts +++ b/packages/@n8n/instance-ai/src/stream/map-chunk.ts @@ -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(value: unknown, schema: z.ZodType): 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(value: unknown, schema: z.ZodType): 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, + 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, + 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, + 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, + 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, + 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> | 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; } diff --git a/packages/@n8n/instance-ai/src/tools/evals/column-ref-utils.ts b/packages/@n8n/instance-ai/src/tools/evals/column-ref-utils.ts index 397e339bf5d..c51a0a9e0be 100644 --- a/packages/@n8n/instance-ai/src/tools/evals/column-ref-utils.ts +++ b/packages/@n8n/instance-ai/src/tools/evals/column-ref-utils.ts @@ -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(); 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); } } } diff --git a/packages/@n8n/instance-ai/src/tools/evals/describe-metric-for-workflow.service.ts b/packages/@n8n/instance-ai/src/tools/evals/describe-metric-for-workflow.service.ts index 9e616542f3b..891b61df6db 100644 --- a/packages/@n8n/instance-ai/src/tools/evals/describe-metric-for-workflow.service.ts +++ b/packages/@n8n/instance-ai/src/tools/evals/describe-metric-for-workflow.service.ts @@ -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; @@ -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); } } } diff --git a/packages/@n8n/instance-ai/src/tools/orchestration/verify-built-workflow.tool.ts b/packages/@n8n/instance-ai/src/tools/orchestration/verify-built-workflow.tool.ts index 78f7bb435a3..620c7f8131d 100644 --- a/packages/@n8n/instance-ai/src/tools/orchestration/verify-built-workflow.tool.ts +++ b/packages/@n8n/instance-ai/src/tools/orchestration/verify-built-workflow.tool.ts @@ -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; +type VerifyInput = z.infer; +type WorkflowTaskService = NonNullable; +type ExecutionRunResult = Awaited< + ReturnType['executionService']['run']> +>; + +/** Names that actually ran, falling back to data keys for hosts that don't report executedNodeNames. */ +function namesOrDataKeys( + reachedNames: Set, + data: Record | 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> | undefined; + workflowTaskService: WorkflowTaskService; + domainContext: NonNullable; + } +> { + 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 { + 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; + nodesNotReached: string[]; + }, +): Promise { + 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 { + 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) => { - 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, diff --git a/packages/@n8n/instance-ai/src/tools/workflows.tool.ts b/packages/@n8n/instance-ai/src/tools/workflows.tool.ts index 7493f66e492..3768bd8574a 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows.tool.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows.tool.ts @@ -485,11 +485,188 @@ async function handleUnarchive( return { success: true }; } +type SetupState = { currentRequestId: string | null; preTestSnapshot: WorkflowJSON | null }; +type SetupResumeData = NonNullable; + +/** 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>, + credentials: Record> | 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, + 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, + 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 | undefined, + credentials: node.credentials, + disabled: node.disabled, + })); + const updatedConnections = updatedWorkflow.connections as Record; + + // 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, 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 | undefined, - credentials: node.credentials, - disabled: node.disabled, - })); - const updatedConnections = updatedWorkflow.connections as Record; - - // 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( diff --git a/packages/@n8n/instance-ai/src/tools/workflows/setup-workflow.service.ts b/packages/@n8n/instance-ai/src/tools/workflows/setup-workflow.service.ts index b367b86b643..99daf4227a0 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/setup-workflow.service.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/setup-workflow.service.ts @@ -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>; + +/** 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 { - if (!node.name) return []; - if (node.disabled) return []; - - const typeVersion = node.typeVersion ?? 1; - const parameters = (node.parameters as Record) ?? {}; - - 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, + typeVersion: number, +): Promise> { let parameterIssues: Record = {}; 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 - ? T extends { options?: infer O } - ? O - : never - : never, - } - : {}), - }); - } +/** Build editable parameter definitions for the parameters that have issues. */ +function buildEditableParameters( + parameterIssues: Record, + nodeDesc: NodeDescription | undefined, +): SetupRequest['editableParameters'] { + if (Object.keys(parameterIssues).length === 0 || !nodeDesc?.properties) return undefined; + + const editableParameters: NonNullable = []; + 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 + ? 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, + typeVersion: number, + nodeDesc: NodeDescription | undefined, +): Promise { 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 { + // 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; + +/** Build the optional `credentials` slice of a setup request's node, merging an auto-applied credential. */ +function buildRequestCredentials( + nodeCredentials: Record | 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 | undefined, +): Promise { + 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; + editableParameters: SetupRequest['editableParameters']; + triggerTestResult?: { status: 'success' | 'error' | 'listening'; error?: string }; + nodeId: string; + nodePosition: [number, number]; + typeVersion: number; + parameters: Record; +} + +/** + * 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 { + 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 { + if (!node.name) return []; + if (node.disabled) return []; + + const typeVersion = node.typeVersion ?? 1; + const parameters = (node.parameters as Record) ?? {}; + + 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, +function addConnectionEdge( + mainOutgoing: Map, + nonMainIncoming: Map, + 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, + mainOutgoing: Map, + nonMainIncoming: Map, +): 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): { + mainOutgoing: Map; + nonMainIncoming: Map; +} { const mainOutgoing = new Map(); - // 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(); 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)) { - 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, + mainOutgoing, + nonMainIncoming, + ); } + return { mainOutgoing, nonMainIncoming }; +} + +export function sortByExecutionOrder( + requests: SetupRequest[], + connections: Record, +): 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, + result: ApplyResult, +): Promise { + 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, + 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 diff --git a/packages/@n8n/instance-ai/src/tools/workflows/validate-workflow.service.ts b/packages/@n8n/instance-ai/src/tools/workflows/validate-workflow.service.ts index 7c4fab91adb..25720684ac8 100644 --- a/packages/@n8n/instance-ai/src/tools/workflows/validate-workflow.service.ts +++ b/packages/@n8n/instance-ai/src/tools/workflows/validate-workflow.service.ts @@ -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[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, ): Promise { - if (node.disabled) return null; - if (!nodeDesc.credentials || nodeDesc.credentials.length === 0) return null; - const parameters = (node.parameters ?? {}) as Record; - 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, + typeVersion: number, + usedCredentialIds: Set, +): Promise { + 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, +): Promise { + 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; + 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, +): Promise | null> { + if (!context.nodeService.getParameterIssues) return null; + const parameterIssues = await context.nodeService + .getParameterIssues(node.type, typeVersion, parameters) + .catch(() => ({}) as Record); + return Object.keys(parameterIssues).length > 0 ? parameterIssues : null; +} + +function computeExecutionIssue( + issues: INodeIssues | null, + nodeName: string, + taskData: ITaskData[] | undefined, + executionErrors: Record, + 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); - 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; diff --git a/packages/@n8n/instance-ai/src/tracing/trace-payloads.ts b/packages/@n8n/instance-ai/src/tracing/trace-payloads.ts index 6734024a6bc..b93693128a8 100644 --- a/packages/@n8n/instance-ai/src/tracing/trace-payloads.ts +++ b/packages/@n8n/instance-ai/src/tracing/trace-payloads.ts @@ -308,6 +308,33 @@ function stringifyToolPayload(value: unknown): string { } } +function extractAssistantToolCall( + part: Record, +): Record | 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): unknown { const content = message.content; if (!Array.isArray(content)) { @@ -327,28 +354,8 @@ function normalizeAssistantMessageForLangSmith(message: Record) 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, + providerAnthropicUsage: Record, + 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, + providerAnthropicUsage: Record, + 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, ): Record | 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, + usageMetadata: Record, + 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): void { const usageMetadata = buildLangSmithUsageMetadata(attributes); if (!usageMetadata) { @@ -710,28 +784,13 @@ function normalizeUsageForLangSmith(attributes: Record): 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)); }