diff --git a/packages/@n8n/api-types/src/index.ts b/packages/@n8n/api-types/src/index.ts index decf7343b4e..fc21b2bbf55 100644 --- a/packages/@n8n/api-types/src/index.ts +++ b/packages/@n8n/api-types/src/index.ts @@ -358,6 +358,7 @@ export { export { buildRunWorkflowSessionGrantKey, + buildExecuteNodeSessionGrantKey, buildRunStepSessionGrantKey, buildUpdateWorkflowSessionGrantKey, buildCredentialDestinationGrantKey, @@ -366,6 +367,7 @@ export { parseSetupSkipGrants, buildFetchUrlGrantKey, FETCH_URL_ALLOW_ALL_GRANT_KEY, + NODE_RESOURCE_GRANT_FALLBACK_KEYS, WEB_SEARCH_GRANT_KEY, parseDomainAccessGrants, instanceAiEventTypeSchema, diff --git a/packages/@n8n/api-types/src/schemas/__tests__/instance-ai.schema.test.ts b/packages/@n8n/api-types/src/schemas/__tests__/instance-ai.schema.test.ts index 1dc5418a3ca..072951c3bd5 100644 --- a/packages/@n8n/api-types/src/schemas/__tests__/instance-ai.schema.test.ts +++ b/packages/@n8n/api-types/src/schemas/__tests__/instance-ai.schema.test.ts @@ -12,6 +12,7 @@ import { applyBranchReadOnlyOverrides, buildCredentialDestinationGrantKey, buildDataTablesSessionGrantKey, + buildExecuteNodeSessionGrantKey, buildUpdateWorkflowSessionGrantKey, buildSetupSkipGrantKey, parseSetupSkipGrants, @@ -708,6 +709,45 @@ describe('data-tables session grant keys', () => { }); }); +describe('node execution session grant keys', () => { + it('scopes the key by resource and operation', () => { + expect( + buildExecuteNodeSessionGrantKey('n8n-nodes-base.slack', { + resource: 'message', + operation: 'post', + }), + ).toBe('nodes:execute:n8n-nodes-base.slack:message:post'); + }); + + it('falls back to the first scoping parameter a node without resource/operation declares', () => { + expect( + buildExecuteNodeSessionGrantKey('n8n-nodes-base.httpRequest', { + method: 'POST', + url: 'https://example.com/v4/sheets', + }), + ).toBe('nodes:execute:n8n-nodes-base.httpRequest:https://example.com/v4/sheets'); + // `mode` precedes `url` in the fallback order, so a node declaring both is scoped by mode. + expect( + buildExecuteNodeSessionGrantKey('n8n-nodes-base.set', { mode: 'manual', url: 'https://x' }), + ).toBe('nodes:execute:n8n-nodes-base.set:manual'); + }); + + it('refuses a key a fallback parameter would push past the column width', () => { + expect( + buildExecuteNodeSessionGrantKey('n8n-nodes-base.graphql', { query: 'q'.repeat(600) }), + ).toBeNull(); + }); + + it('keeps the key at the node type when nothing scopes the call', () => { + expect(buildExecuteNodeSessionGrantKey('n8n-nodes-base.filter', { conditions: {} })).toBe( + 'nodes:execute:n8n-nodes-base.filter', + ); + expect(buildExecuteNodeSessionGrantKey('n8n-nodes-base.filter')).toBe( + 'nodes:execute:n8n-nodes-base.filter', + ); + }); +}); + describe('workflow update session grant keys', () => { it('builds per-workflow keys matching the frontend always-allow format', () => { expect(buildUpdateWorkflowSessionGrantKey('wf-1')).toBe('workflows:update:wf-1'); diff --git a/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts b/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts index c83bc0ce745..6271748ac4c 100644 --- a/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts +++ b/packages/@n8n/api-types/src/schemas/instance-ai.schema.ts @@ -56,6 +56,43 @@ export function buildRunWorkflowSessionGrantKey(workflowId: string): string { return `executions:run:${workflowId}`; } +/** Used to determine resource for nodes that have no resource and operation, like http request node*/ +export const NODE_RESOURCE_GRANT_FALLBACK_KEYS = ['mode', 'url', 'query', 'command', 'action']; + +/** Width of `instance_ai_thread_grants.grantKey`. */ +const MAX_GRANT_KEY_LENGTH = 512; + +/** + * Builds the thread-level grant key for `nodes(action="execute")`. Scoped per + * node type + resource + operation (the split the generated node TS types use). + * Extracts the discriminators from the parameters itself so backend and + * frontend cannot build diverging keys. + * + * Null when a fallback discriminator (a long URL, GraphQL query or command) pushes the key + * past the column width: the call then needs approval every time. Truncating instead would + * let one approved URL stand for every other URL sharing its prefix. + */ +export function buildExecuteNodeSessionGrantKey( + nodeType: string, + parameters?: Record, +): string | null { + const resourceParts = [ + nodeType, + typeof parameters?.resource === 'string' ? parameters.resource : '', + typeof parameters?.operation === 'string' ? parameters.operation : '', + ].filter(Boolean); + if (resourceParts.length === 1) { + for (const name of NODE_RESOURCE_GRANT_FALLBACK_KEYS) { + if (typeof parameters?.[name] === 'string' && parameters?.[name]) { + resourceParts.push(parameters?.[name]); + break; + } + } + } + const key = `nodes:execute:${resourceParts.join(':')}`; + return key.length <= MAX_GRANT_KEY_LENGTH ? key : null; +} + /** * Builds the thread-level "always allow" grant key for running one node of a * workflow ("execute step"). diff --git a/packages/@n8n/db/src/repositories/__tests__/workflow-create-with-owner.repository.test.ts b/packages/@n8n/db/src/repositories/__tests__/workflow-create-with-owner.repository.test.ts new file mode 100644 index 00000000000..e7f0b2b8223 --- /dev/null +++ b/packages/@n8n/db/src/repositories/__tests__/workflow-create-with-owner.repository.test.ts @@ -0,0 +1,67 @@ +import { GlobalConfig } from '@n8n/config'; +import { mock } from 'vitest-mock-extended'; + +import { WorkflowEntity } from '../../entities'; +import { SharedWorkflow } from '../../entities/shared-workflow'; +import type { OperationContext, TransactionRunner } from '../../services/transaction'; +import { mockEntityManager } from '../../utils/test-utils/mock-entity-manager'; +import { mockInstance } from '../../utils/test-utils/mock-instance'; +import { FolderRepository } from '../folder.repository'; +import { SharedWorkflowRepository } from '../shared-workflow.repository'; +import { WorkflowHistoryRepository } from '../workflow-history.repository'; +import { WorkflowRepository } from '../workflow.repository'; + +describe('WorkflowRepository.createWorkflowWithOwner', () => { + const entityManager = mockEntityManager(WorkflowEntity); + const transactionRunner = mock(); + const workflowRepository = new WorkflowRepository( + entityManager.connection, + mockInstance(GlobalConfig, { database: { type: 'postgresdb' } }), + mockInstance(FolderRepository), + mockInstance(SharedWorkflowRepository), + mockInstance(WorkflowHistoryRepository), + transactionRunner, + ); + + beforeEach(() => { + vi.resetAllMocks(); + transactionRunner.run.mockImplementation( + async (ctx: OperationContext, fn: (ctx: OperationContext) => Promise) => + await fn(ctx), + ); + entityManager.create.mockImplementation((_target, entityLike) => entityLike as never); + entityManager.save.mockImplementation(async (entity: unknown) => { + if (entity instanceof WorkflowEntity) entity.id = 'wf-1'; + return await Promise.resolve(entity); + }); + }); + + it('saves the workflow and its owner share in one transaction', async () => { + const workflow = new WorkflowEntity(); + + const saved = await workflowRepository.createWorkflowWithOwner(workflow, 'project-1'); + + expect(saved).toBe(workflow); + expect(transactionRunner.run).toHaveBeenCalledTimes(1); + expect(entityManager.save).toHaveBeenNthCalledWith(1, workflow); + expect(entityManager.create).toHaveBeenCalledWith(SharedWorkflow, { + role: 'workflow:owner', + projectId: 'project-1', + workflowId: 'wf-1', + }); + expect(entityManager.save).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ workflowId: 'wf-1' }), + ); + }); + + it('does not create the share when saving the workflow fails', async () => { + entityManager.save.mockRejectedValueOnce(new Error('boom')); + + await expect( + workflowRepository.createWorkflowWithOwner(new WorkflowEntity(), 'project-1'), + ).rejects.toThrow('boom'); + + expect(entityManager.create).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/@n8n/db/src/repositories/workflow.repository.ts b/packages/@n8n/db/src/repositories/workflow.repository.ts index bcf8c9ab478..c087cdb4b95 100644 --- a/packages/@n8n/db/src/repositories/workflow.repository.ts +++ b/packages/@n8n/db/src/repositories/workflow.repository.ts @@ -308,6 +308,28 @@ export class WorkflowRepository extends BaseRepository { await this.managerFor(ctx).update(WorkflowEntity, id, content); } + /** + * Creates the workflow together with its `workflow:owner` share in one transaction. + * + * Deliberately outside the `workflowSave` clearance the other writes here assert: the only + * caller is standalone node execution, whose row is archived, single-node and deleted after + * the run. The content is still policed where it matters — `PolicyLifecycleHandler` enforces + * `workflowStart` on `workflowExecuteBefore`, which every execution path reaches. + */ + async createWorkflowWithOwner( + workflow: WorkflowEntity, + projectId: string, + ctx: OperationContext = {}, + ): Promise { + return await this.runInTransaction(ctx, async (em) => { + const saved = await em.save(workflow); + await em.save( + em.create(SharedWorkflow, { role: 'workflow:owner', projectId, workflowId: saved.id }), + ); + return saved; + }); + } + /** * Persists a new workflow, gated on a clearance for its content. * diff --git a/packages/@n8n/instance-ai/docs/tools.md b/packages/@n8n/instance-ai/docs/tools.md index bd60aa4da24..bad5a45db80 100644 --- a/packages/@n8n/instance-ai/docs/tools.md +++ b/packages/@n8n/instance-ai/docs/tools.md @@ -53,7 +53,7 @@ keep their explicit pattern. `like` matches case; `ilike` ignores case. | `workspace` | 8 | | `executions` | 8 | | `credentials` | 6 | -| `nodes` | 6 | +| `nodes` | 7 | | `mcp-servers` | 4 | | `conversation-history` | 2 | | `task-control` | 3 | @@ -828,13 +828,13 @@ Test whether a credential is valid and can connect to its service. --- -## `nodes` (6 actions) +## `nodes` (7 actions) -The full domain surface has six actions. The orchestrator receives all six +The full domain surface has seven actions. The orchestrator receives all seven actions in the current registry. The tool also defines a restricted `type-definition` and `explore-resources` surface, but the orchestrator registry does not currently select it. Specialized agents that resolve the full domain -tool can also receive all six actions. +tool can also receive all seven actions. ### `nodes(action="list")` @@ -910,6 +910,70 @@ discriminator values like spreadsheet IDs, calendar names, etc. **Returns**: `{ results, paginationToken?, builderHint?, error? }`. +### `nodes(action="execute")` + +Execute a single node standalone — real credentials, caller-supplied parameters +and input items — and return its real output items. The node runs through the +regular execution engine (an archived temporary workflow is created for the run +and deleted afterwards), so queue-mode worker dispatch applies. Intended for +learning a node's exact output shape before wiring downstream expressions, or +testing one node in isolation. + +Before the approval prompt, `config` is validated against the generated +workflow-sdk node schema (`validateNodeConfig`) - a malformed config returns +field-level errors immediately. One exception: a missing discriminator (e.g. +`resource`/`operation`) does not block — n8n falls back to the node's defaults +at runtime, so the node can still run and cause side effects. + +**Approval mirrors `executions(action="run")`** — executing one node is +equivalent to running a one-node workflow, so the same `runWorkflow` admin +policy applies (`blocked` denies; `always_allow` skips the prompt — a +standalone node request is always agent-authored, the analog of an AI-created +workflow). A *scoped* `always_allow` — the checkpoint follow-up override, which +names the workflow IDs it covers — does not skip the prompt: a standalone node +run has no workflow ID to match. Under the default `require_approval`, the tool suspends with +severity `warning`; "Always allow" persists a session grant scoped by node +type + resource + operation (`nodes:execute:::`) — +the same split the generated node TS types use, so a future per-operation +destructiveness policy plugs in without changing the key format. Later +executions of the same operation skip the prompt for the session. + +A node that declares neither discriminator (HTTP Request, Set, Merge, Filter, …) +is scoped by the first of `mode`, `url`, `query`, `command` or `action` it +declares, and by node type alone when it declares none. "Always allow" is not +offered when that value is long enough to push the key past the grant column +width — one approved URL must not stand for every URL sharing its prefix. + +The request envelope mirrors a workflow-sdk node (`{ type, version, config }`), +so the agent can pass a node it is building verbatim: + +| Field | Type | Required | Description | +|-------|------|----------|-------------| +| `type` | string | yes | Full node type name, e.g. `n8n-nodes-base.slack` | +| `version` | number | yes | Node type version | +| `config.parameters` | object | yes | Same shape as workflow-sdk `NodeConfig.parameters` | +| `config.credentials` | object | no | Resolved credential references `{ id, name }` by credential type; n8n Connect managed credentials use `{ id: null, name, __aiGatewayManaged: true }` | +| `input` | array | no | Input items `{ json }` (defaults to one empty item) | +| `timeoutMs` | number | no | Max execution time, capped at 60s | + +**Returns**: `{ status: 'success', output }` or +`{ status: 'error', error: { message, description?, nodeErrorType? } }`. The +output is the serialized items inside an `` boundary, the same envelope the workflow-execution +path puts on node output — the items come from whatever service the node +called. Binary output is reduced to metadata (`fileName`, `mimeType`, +`fileSize`). Output is size-capped (a `truncated` field reports shown vs total +items). +When `N8N_AI_ALLOW_SENDING_PARAMETER_VALUES` is disabled, output items and +upstream error details are suppressed, mirroring `executions(action="run")`. +Wait states are not supported — a node that starts waiting (e.g. Wait, +send-and-wait operations) returns an error. + +Limitations: the node really runs (side effects happen); expressions +referencing other nodes cannot resolve; trigger/webhook-only nodes are +rejected; credentials must be resolved references — the SDK's +placeholder/new-credential forms have no stored row and cannot execute. + --- ## `data-tables` (11 actions) diff --git a/packages/@n8n/instance-ai/skills/one-off-operations/SKILL.md b/packages/@n8n/instance-ai/skills/one-off-operations/SKILL.md index d7bd7a8911c..75a3e7af21e 100644 --- a/packages/@n8n/instance-ai/skills/one-off-operations/SKILL.md +++ b/packages/@n8n/instance-ai/skills/one-off-operations/SKILL.md @@ -10,6 +10,7 @@ description: >- for automations the user will run again — that is the normal build + post-build-flow path. recommended_tools: + - nodes - build-workflow - workflows - executions @@ -50,11 +51,51 @@ treat the request as reusable** and follow the normal build flow — a reusable workflow that runs once is harmless; a one-off flow applied to an automation skips verification the user would have wanted. -A one-off that touches external systems is still workflow-anchored (you cannot -write to external services directly) — the intent changes the *post-build -flow*, not the anchor. +## Single-node one-offs: prefer direct node execution -## The one-off flow +When the entire effect is **one node operation**, skip the workflow: execute +the node directly with `nodes(action="execute")`. It runs a single node with +real credentials through the regular execution engine and returns its real +output items — no workflow to build, set up, verify, or clean up afterwards. + +Direct execution is sufficient when ALL of these hold: + +- The effect is one node call (one write, one API operation). A read-back of + the destination may be a second `execute` call of a read operation. +- The input items are already at hand — pasted into the chat or read earlier + in the conversation — so you can pass them literally as `input` items. + Parameters must not use expressions referencing other nodes; they cannot + resolve (the node runs alone). +- The node runs standalone on `main` input alone — no required sub-node + connections (e.g. an AI Agent needs a language model attached; such nodes + need a workflow). +- A usable credential already exists (`credentials(action="list")`); the + action takes resolved `{ id, name }` references. If credentials must be + created first, route that through the credentials setup as usual. +- The run fits the 60s cap and the input volume is modest. + +The run executes in the current conversation's project, so credentials shared +with that project are usable. If it fails because a credential is not +accessible there, fall back to the one-off workflow flow below. + +Call it with the same shape as a workflow-sdk node — `{ type, version, +config: { parameters, credentials } }` plus `input` items. Read +`nodes(action="type-definition")` first, as you would before configuring any +node. Approval works exactly like `executions(action="run")` — the same +run-approval card, admin policy, and session grants — and for a one-off that +prompt is the consent gate. The returned output items are +real, so the read-back rule below is satisfied by reading what came back — +report only from those items (binary content is returned as metadata only). + +If the task needs more than that — several chained steps, branching, merges, +non-trivial transformations, or data that must flow between nodes — build the +workflow and use the flow below. + +A one-off that touches external systems is still anchored on n8n nodes (you +cannot write to external services directly) — the intent changes the +*post-build flow*, not the anchor. + +## The one-off flow (multi-node) 1. **Build** the workflow with a **manual trigger** — always. A one-off is never published, so an event trigger (webhook, form, schedule) would never diff --git a/packages/@n8n/instance-ai/skills/workflow-builder/SKILL.md b/packages/@n8n/instance-ai/skills/workflow-builder/SKILL.md index ddfb51f48c6..1eb366f492d 100644 --- a/packages/@n8n/instance-ai/skills/workflow-builder/SKILL.md +++ b/packages/@n8n/instance-ai/skills/workflow-builder/SKILL.md @@ -9,6 +9,9 @@ description: >- data-table-manager first, then this skill. Do not load planning or create-tasks first. Load planning only when multiple coordinated workflows or shared cross-task data tables require a dependency-aware task graph. + Don't use this skill for explicit one-off tasks that can be done by a single + node execution: load one-off-operations and run the node with + nodes(action="execute"). recommended_tools: - read_file - write_file diff --git a/packages/@n8n/instance-ai/src/index.ts b/packages/@n8n/instance-ai/src/index.ts index 059a768f46c..8e8e9500149 100644 --- a/packages/@n8n/instance-ai/src/index.ts +++ b/packages/@n8n/instance-ai/src/index.ts @@ -640,6 +640,8 @@ export type { InstanceAiToolRegistry, InstanceAiWorkflowService, InstanceAiExecutionService, + InstanceAiExecuteNodeService, + ExecuteNodeResult, InstanceAiCredentialService, InstanceAiNodeService, InstanceAiDataTableService, diff --git a/packages/@n8n/instance-ai/src/tools/__tests__/nodes.tool.test.ts b/packages/@n8n/instance-ai/src/tools/__tests__/nodes.tool.test.ts index f9eb7270543..c2f37e696ca 100644 --- a/packages/@n8n/instance-ai/src/tools/__tests__/nodes.tool.test.ts +++ b/packages/@n8n/instance-ai/src/tools/__tests__/nodes.tool.test.ts @@ -3,6 +3,7 @@ import { suggestedNodesData, type SearchableNodeType, } from '@n8n/ai-utilities/node-catalog'; +import { validateNodeConfig } from '@n8n/workflow-sdk'; import type { Mock } from 'vitest'; import { executeTool } from '../../__tests__/tool-test-utils'; @@ -10,6 +11,11 @@ import type { InstanceAiContext } from '../../types'; import { addSetupPreference } from '../nodes/setup-preference'; import { createNodesTool } from '../nodes.tool'; +vi.mock('@n8n/workflow-sdk', async (importOriginal) => ({ + ...(await importOriginal()), + validateNodeConfig: vi.fn(() => ({ valid: true, errors: [] })), +})); + function createMockContext(overrides: Partial = {}): InstanceAiContext { return { userId: 'user-1', @@ -644,4 +650,465 @@ describe('nodes tool', () => { }); }); }); + + describe('execute action', () => { + const executeInput = { + action: 'execute', + type: 'n8n-nodes-base.set', + version: 3.4, + config: { parameters: { mode: 'manual' } }, + }; + + it('should return a structured error when required execute fields are missing', async () => { + const executeNodeService = { execute: vi.fn() }; + const tool = createNodesTool(createMockContext({ executeNodeService }), 'full'); + + const result = await executeTool( + tool, + { action: 'execute', type: 'n8n-nodes-base.set' } as never, + {} as never, + ); + + expect(result).toMatchObject({ + status: 'error', + error: { message: expect.stringContaining('version') }, + }); + expect(executeNodeService.execute).not.toHaveBeenCalled(); + }); + + it('should reject a config that fails schema validation before suspending', async () => { + const executeNodeService = { execute: vi.fn() }; + const suspendFn = vi.fn(); + vi.mocked(validateNodeConfig).mockReturnValueOnce({ + valid: false, + errors: [ + { path: 'parameters.text', message: 'Required' }, + { path: 'parameters.select', message: 'Invalid value', missingDiscriminator: true }, + ], + }); + const tool = createNodesTool(createMockContext({ executeNodeService }), 'full'); + + const result = await executeTool( + tool, + executeInput as never, + { + suspend: suspendFn, + } as never, + ); + + expect(result).toEqual({ + status: 'error', + error: { + message: 'Node parameters do not match the schema for n8n-nodes-base.set v3.4', + issues: [{ path: 'parameters.text', message: 'Required' }], + }, + }); + expect(suspendFn).not.toHaveBeenCalled(); + expect(executeNodeService.execute).not.toHaveBeenCalled(); + }); + + it('should treat missing-discriminator-only validation errors as non-blocking', async () => { + const executeNodeService = { execute: vi.fn() }; + const suspendFn = vi.fn(); + vi.mocked(validateNodeConfig).mockReturnValueOnce({ + valid: false, + errors: [{ path: 'parameters.resource', message: 'Required', missingDiscriminator: true }], + }); + const tool = createNodesTool(createMockContext({ executeNodeService }), 'full'); + + await executeTool(tool, executeInput as never, { suspend: suspendFn } as never); + + expect(suspendFn).toHaveBeenCalledTimes(1); + }); + + it('should suspend for confirmation on the first call', async () => { + const executeNodeService = { execute: vi.fn() }; + const suspendFn = vi.fn(); + const tool = createNodesTool(createMockContext({ executeNodeService }), 'full'); + + await executeTool(tool, executeInput as never, { suspend: suspendFn } as never); + + expect(suspendFn).toHaveBeenCalledTimes(1); + expect(suspendFn.mock.calls[0][0]).toEqual( + expect.objectContaining({ + requestId: expect.any(String), + message: 'manual', + resourceName: 'n8n-nodes-base.set node', + severity: 'warning', + }), + ); + expect(executeNodeService.execute).not.toHaveBeenCalled(); + }); + + /** Mirrors Google Sheets: `create` is an operation of both resources, with its own label. */ + const splitNodeDescription = { + name: 'n8n-nodes-base.googleSheets', + displayName: 'Google Sheets', + properties: [ + { + name: 'resource', + displayName: 'Resource', + type: 'options', + default: 'sheet', + options: [ + { name: 'Document', value: 'spreadsheet' }, + { name: 'Sheet Within Document', value: 'sheet' }, + ], + }, + { + name: 'operation', + displayName: 'Operation', + type: 'options', + default: 'read', + displayOptions: { show: { resource: ['sheet'] } }, + options: [ + { name: 'Create Sheet', value: 'create' }, + { name: 'Get Row(s)', value: 'read' }, + ], + }, + { + name: 'operation', + displayName: 'Operation', + type: 'options', + default: 'create', + displayOptions: { show: { resource: ['spreadsheet'] } }, + options: [{ name: 'Create Document', value: 'create' }], + }, + ], + }; + + async function suspendPayloadFor( + parameters: Record, + description: unknown = splitNodeDescription, + ) { + const suspendFn = vi.fn(); + const context = createMockContext({ executeNodeService: { execute: vi.fn() } }); + (context.nodeService.getDescription as Mock).mockResolvedValue(description); + + await executeTool( + createNodesTool(context, 'full'), + { + action: 'execute', + type: 'n8n-nodes-base.googleSheets', + version: 4.7, + config: { parameters }, + } as never, + { suspend: suspendFn } as never, + ); + + return suspendFn.mock.calls[0][0]; + } + + it('should name the node in the title and the operation in the description', async () => { + expect(await suspendPayloadFor({ resource: 'sheet', operation: 'create' })).toMatchObject({ + resourceName: 'Google Sheets node', + message: 'Sheet Within Document > Create Sheet', + }); + }); + + it('should label a shared operation value by the resource it belongs to', async () => { + expect( + await suspendPayloadFor({ resource: 'spreadsheet', operation: 'create' }), + ).toMatchObject({ message: 'Document > Create Document' }); + }); + + it('should fall back to the node defaults for omitted discriminators', async () => { + expect(await suspendPayloadFor({ operation: 'create' })).toMatchObject({ + message: 'Sheet Within Document > Create Sheet', + }); + expect(await suspendPayloadFor({})).toMatchObject({ + message: 'Sheet Within Document > Get Row(s)', + }); + }); + + it('should name the first headline parameter for a node without resource or operation', async () => { + const httpRequest = { + name: 'n8n-nodes-base.httpRequest', + displayName: 'HTTP Request', + properties: [ + { name: 'method', displayName: 'Method', type: 'options', default: 'GET' }, + { name: 'url', displayName: 'URL', type: 'string', default: '' }, + ], + }; + expect( + await suspendPayloadFor( + { method: 'POST', url: 'https://example.com/v4/sheets' }, + httpRequest, + ), + ).toMatchObject({ + resourceName: 'HTTP Request node', + message: 'https://example.com/v4/sheets', + }); + }); + + it('should prefer mode over the later headline parameters and label it', async () => { + const set = { + name: 'n8n-nodes-base.set', + displayName: 'Edit Fields (Set)', + properties: [ + { + name: 'mode', + displayName: 'Mode', + type: 'options', + default: 'manual', + options: [ + { name: 'Manual Mapping', value: 'manual' }, + { name: 'JSON', value: 'raw' }, + ], + }, + { name: 'url', displayName: 'URL', type: 'string', default: '' }, + ], + }; + expect(await suspendPayloadFor({ url: 'https://example.com' }, set)).toMatchObject({ + message: 'Manual Mapping', + }); + }); + + it('should fall back to a generic description when no parameter names the call', async () => { + const filter = { name: 'n8n-nodes-base.filter', displayName: 'Filter', properties: [] }; + expect(await suspendPayloadFor({}, filter)).toMatchObject({ + resourceName: 'Filter node', + message: 'Single run', + }); + }); + + it('should fall back to raw values when the node description does not resolve', async () => { + const suspendFn = vi.fn(); + const context = createMockContext({ executeNodeService: { execute: vi.fn() } }); + (context.nodeService.getDescription as Mock).mockRejectedValue(new Error('not found')); + + await executeTool( + createNodesTool(context, 'full'), + { + action: 'execute', + type: 'n8n-nodes-base.slack', + version: 2.3, + config: { parameters: { resource: 'message', operation: 'post' } }, + } as never, + { suspend: suspendFn } as never, + ); + + expect(suspendFn.mock.calls[0][0]).toMatchObject({ + resourceName: 'n8n-nodes-base.slack node', + message: 'message > post', + }); + }); + + it('should deny without suspending when the admin policy blocks workflow runs', async () => { + const executeNodeService = { execute: vi.fn() }; + const suspendFn = vi.fn(); + const tool = createNodesTool( + createMockContext({ executeNodeService, permissions: { runWorkflow: 'blocked' } as never }), + 'full', + ); + + const result = await executeTool( + tool, + executeInput as never, + { + suspend: suspendFn, + } as never, + ); + + expect(result).toEqual({ status: 'error', denied: true, reason: 'Action blocked by admin' }); + expect(suspendFn).not.toHaveBeenCalled(); + expect(executeNodeService.execute).not.toHaveBeenCalled(); + }); + + it('should skip approval when the admin policy is always_allow', async () => { + const serviceResult = { status: 'success', output: [[{ json: {} }]] }; + const executeNodeService = { execute: vi.fn().mockResolvedValue(serviceResult) }; + const suspendFn = vi.fn(); + const tool = createNodesTool( + createMockContext({ + executeNodeService, + permissions: { runWorkflow: 'always_allow' } as never, + }), + 'full', + ); + + const result = await executeTool( + tool, + executeInput as never, + { + suspend: suspendFn, + } as never, + ); + + expect(suspendFn).not.toHaveBeenCalled(); + expect(result).toEqual(serviceResult); + }); + + it('should require approval when always_allow is scoped to specific workflows', async () => { + const executeNodeService = { execute: vi.fn() }; + const suspendFn = vi.fn(); + const tool = createNodesTool( + createMockContext({ + executeNodeService, + permissions: { runWorkflow: 'always_allow' } as never, + allowedRunWorkflowIds: new Set(['wf-under-verification']), + }), + 'full', + ); + + await executeTool(tool, executeInput as never, { suspend: suspendFn } as never); + + expect(suspendFn).toHaveBeenCalledTimes(1); + expect(executeNodeService.execute).not.toHaveBeenCalled(); + }); + + it('should skip approval when a session grant exists for the node type', async () => { + const serviceResult = { status: 'success', output: [[{ json: {} }]] }; + const executeNodeService = { execute: vi.fn().mockResolvedValue(serviceResult) }; + const suspendFn = vi.fn(); + const tool = createNodesTool( + createMockContext({ + executeNodeService, + sessionApprovedToolKeys: new Set(['nodes:execute:n8n-nodes-base.set:manual']), + }), + 'full', + ); + + const result = await executeTool( + tool, + executeInput as never, + { + suspend: suspendFn, + } as never, + ); + + expect(suspendFn).not.toHaveBeenCalled(); + expect(result).toEqual(serviceResult); + }); + + it('should persist a session grant split by type, resource, and operation on "always allow"', async () => { + const executeNodeService = { execute: vi.fn().mockResolvedValue({ status: 'success' }) }; + const grantSessionToolApproval = vi.fn(); + const tool = createNodesTool( + createMockContext({ executeNodeService, grantSessionToolApproval }), + 'full', + ); + + await executeTool( + tool, + { + action: 'execute', + type: 'n8n-nodes-base.slack', + version: 2.7, + config: { parameters: { resource: 'message', operation: 'post', text: 'hi' } }, + } as never, + { resumeData: { approved: true, scope: 'session' } } as never, + ); + + expect(grantSessionToolApproval).toHaveBeenCalledWith( + 'nodes:execute:n8n-nodes-base.slack:message:post', + ); + expect(executeNodeService.execute).toHaveBeenCalled(); + }); + + it('should persist a grant scoped by the fallback parameter without resource/operation', async () => { + const executeNodeService = { execute: vi.fn().mockResolvedValue({ status: 'success' }) }; + const grantSessionToolApproval = vi.fn(); + const tool = createNodesTool( + createMockContext({ executeNodeService, grantSessionToolApproval }), + 'full', + ); + + await executeTool( + tool, + executeInput as never, + { + resumeData: { approved: true, scope: 'session' }, + } as never, + ); + + expect(grantSessionToolApproval).toHaveBeenCalledWith( + 'nodes:execute:n8n-nodes-base.set:manual', + ); + }); + + it('should persist a per-type grant for a node with no scoping parameter at all', async () => { + const executeNodeService = { execute: vi.fn().mockResolvedValue({ status: 'success' }) }; + const grantSessionToolApproval = vi.fn(); + const tool = createNodesTool( + createMockContext({ executeNodeService, grantSessionToolApproval }), + 'full', + ); + + await executeTool( + tool, + { + action: 'execute', + type: 'n8n-nodes-base.filter', + version: 2.2, + config: { parameters: { conditions: {} } }, + } as never, + { resumeData: { approved: true, scope: 'session' } } as never, + ); + + expect(grantSessionToolApproval).toHaveBeenCalledWith('nodes:execute:n8n-nodes-base.filter'); + }); + + it('should return a denied result without executing when the user denies', async () => { + const executeNodeService = { execute: vi.fn() }; + const tool = createNodesTool(createMockContext({ executeNodeService }), 'full'); + + const result = await executeTool( + tool, + executeInput as never, + { + resumeData: { approved: false }, + } as never, + ); + + expect(result).toEqual({ status: 'error', denied: true, reason: 'User denied the action' }); + expect(executeNodeService.execute).not.toHaveBeenCalled(); + }); + + it('should execute the node with the workflow-sdk-shaped request when approved', async () => { + const serviceResult = { status: 'success', output: [[{ json: { done: true } }]] }; + const executeNodeService = { execute: vi.fn().mockResolvedValue(serviceResult) }; + const tool = createNodesTool(createMockContext({ executeNodeService }), 'full'); + + const result = await executeTool( + tool, + { + ...executeInput, + config: { + parameters: { mode: 'manual' }, + credentials: { slackApi: { id: 'cred-1', name: 'Slack' } }, + }, + input: [{ json: { text: 'hi' } }], + timeoutMs: 10_000, + } as never, + { resumeData: { approved: true } } as never, + ); + + expect(executeNodeService.execute).toHaveBeenCalledWith({ + type: 'n8n-nodes-base.set', + version: 3.4, + config: { + parameters: { mode: 'manual' }, + credentials: { slackApi: { id: 'cred-1', name: 'Slack' } }, + }, + input: [{ json: { text: 'hi' } }], + timeoutMs: 10_000, + }); + expect(result).toEqual(serviceResult); + }); + + it('should return an error when executeNodeService is not wired', async () => { + const tool = createNodesTool(createMockContext(), 'full'); + + const result = await executeTool( + tool, + executeInput as never, + { + resumeData: { approved: true }, + } as never, + ); + + expect(result).toMatchObject({ status: 'error' }); + }); + }); }); diff --git a/packages/@n8n/instance-ai/src/tools/nodes.tool.ts b/packages/@n8n/instance-ai/src/tools/nodes.tool.ts index 7b206682511..ef5b1c30f16 100644 --- a/packages/@n8n/instance-ai/src/tools/nodes.tool.ts +++ b/packages/@n8n/instance-ai/src/tools/nodes.tool.ts @@ -1,5 +1,6 @@ /** - * Consolidated nodes tool — list, search, describe, type-definition, suggested, explore-resources. + * Consolidated nodes tool — list, search, describe, type-definition, suggested, + * explore-resources, execute. */ import { Tool } from '@n8n/agents'; import { @@ -10,10 +11,18 @@ import { type CategorySuggestedNode, type SearchableNodeType, } from '@n8n/ai-utilities/node-catalog'; +import { + buildExecuteNodeSessionGrantKey, + instanceAiApprovalResumeSchema, + instanceAiConfirmationSeveritySchema, + NODE_RESOURCE_GRANT_FALLBACK_KEYS, +} from '@n8n/api-types'; +import { validateNodeConfig } from '@n8n/workflow-sdk'; +import { nanoid } from 'nanoid'; import { z } from 'zod'; import { sanitizeInputSchema } from '../agent/sanitize-mcp-schemas'; -import type { InstanceAiContext } from '../types'; +import type { InstanceAiContext, NodeDescription } from '../types'; import { pickPreferredChatModelNode } from './nodes/preferred-chat-model'; import { addSetupPreference, type NodeWithSetupPreference } from './nodes/setup-preference'; import { buildCredentialMap } from './workflows/resolve-credentials'; @@ -128,6 +137,63 @@ const exploreResourcesAction = z.object({ .describe(CURRENT_NODE_PARAMETERS_DESCRIPTION), }); +const MAX_EXECUTE_TIMEOUT_MS = 60_000; + +// Envelope mirrors a workflow-sdk node so the agent can pass a node it is +// building verbatim. Credentials take the resolved `{ id, name }` form only — +// the SDK's placeholder/new-credential forms have no stored row to execute with. +const executeAction = z.object({ + action: z + .literal('execute') + .describe( + 'Execute a single node standalone with real credentials and return its real output ' + + 'items. Use it to learn the exact output shape of a node before wiring downstream ' + + "expressions, or to test one node in isolation. Always read the node's " + + '`action: "type-definition"` first and build the parameters from it — never guess ' + + 'parameter names, resource/operation values, or the version. The node really runs — ' + + 'side effects happen (messages get sent, rows get written). ' + + 'Expressions referencing other nodes cannot resolve; binary ' + + 'output is returned as metadata only.', + ), + type: z.string().min(1).describe(NODE_TYPE_ID_DESCRIPTION), + version: z.number().describe('Node version, e.g. 4.7'), + config: z + .object({ + parameters: z + .record(z.unknown()) + .describe('Node parameters — same shape as workflow-sdk NodeConfig.parameters'), + credentials: z + .record( + z.object({ + id: z.string().nullable(), + name: z.string(), + __aiGatewayManaged: z.boolean().optional(), + }), + ) + .optional() + .describe( + 'Resolved credential references by credential type, e.g. { slackApi: { id, name } }. ' + + 'Ask the user which credentials to use if there are multiple credentials available - ' + + 'do not pick a credential on your own', + ), + }) + .describe('Node config — same shape as a workflow-sdk node config'), + input: z + .array(z.object({ json: z.record(z.unknown()) })) + .optional() + .describe('Input items for the node (defaults to one empty item)'), + timeoutMs: z.number().int().positive().max(MAX_EXECUTE_TIMEOUT_MS).optional(), +}); + +type ExecuteInput = z.infer; + +const suspendSchema = z.object({ + requestId: z.string(), + message: z.string(), + resourceName: z.string().optional(), + severity: instanceAiConfirmationSeveritySchema, +}); + const fullInputSchema = sanitizeInputSchema( z.discriminatedUnion('action', [ listAction, @@ -136,6 +202,7 @@ const fullInputSchema = sanitizeInputSchema( typeDefinitionAction, suggestedAction, exploreResourcesAction, + executeAction, ]), ); @@ -400,6 +467,160 @@ async function handleExploreResources( } } +/** True when the property is not gated on `resource`, or is gated on the resolved one. */ +function isShownForResource( + property: NodeDescription['properties'][number], + resource: string | undefined, +): boolean { + const shownFor = property.displayOptions?.show?.resource; + if (!Array.isArray(shownFor)) return true; + return resource !== undefined && shownFor.includes(resource); +} + +/** Description for the 45 node types that carry no parameter worth naming (If, Filter, Sort, ...). */ +const EXECUTE_DESCRIPTION_FALLBACK = 'Single run'; + +/** + * Human-readable subjects for the execute-node approval card: the node the card names in its + * title (`Google Sheets node`) and what the call does below it (`Sheet Within Document > Append + * Row`). Falls back to the node type ID and the raw parameter values when the node description + * or its option lists don't resolve. + */ +async function buildExecuteNodeLabels( + context: InstanceAiContext, + input: ExecuteInput, +): Promise<{ resourceName: string; message: string }> { + let description: NodeDescription | undefined; + try { + description = await context.nodeService.getDescription(input.type, input.version); + } catch { + // Keep the raw node type in the prompt. + } + const properties = description?.properties ?? []; + + /** + * A split node declares one `operation` property for each resource, so the same operation + * value can carry a different label under another resource — only the properties shown for + * the resolved resource may name it. An omitted parameter falls back to the node default at + * runtime, so the prompt resolves it the same way and names the call the user really gets. + */ + const resolveParameter = (name: string, resource?: string) => { + const declaredBy = properties.filter( + (property) => property.name === name && isShownForResource(property, resource), + ); + const value = input.config.parameters[name] ?? declaredBy[0]?.default; + if (typeof value !== 'string' || value.length === 0) return undefined; + const option = declaredBy + .flatMap((property) => property.options ?? []) + .find((candidate) => candidate.value === value); + return { value, label: option?.name ?? value }; + }; + + const resource = resolveParameter('resource'); + const operation = resolveParameter('operation', resource?.value); + let label = [resource?.label, operation?.label].filter(Boolean).join(' > '); + for (const name of NODE_RESOURCE_GRANT_FALLBACK_KEYS) { + if (label) break; + label = resolveParameter(name)?.label ?? ''; + } + + return { + resourceName: `${description?.displayName ?? input.type} node`, + message: label || EXECUTE_DESCRIPTION_FALLBACK, + }; +} + +async function handleExecute( + context: InstanceAiContext, + rawInput: ExecuteInput, + resumeData: z.infer | undefined | null, + suspend: (payload: z.infer) => Promise, +) { + const { executeNodeService } = context; + if (!executeNodeService) { + return { + status: 'error' as const, + error: { message: 'Node execution is not available on this instance' }, + }; + } + + // The flattened runtime schema makes every variant field optional — re-assert + // the variant contract so a missing field returns a structured error. + const parsedInput = executeAction.safeParse(rawInput); + if (!parsedInput.success) { + return { + status: 'error' as const, + error: { + message: parsedInput.error.issues + .map((issue) => `${issue.path.join('.') || ''}: ${issue.message}`) + .join('; '), + }, + }; + } + const input = parsedInput.data; + + const validation = validateNodeConfig(input.type, input.version, input.config); + // Missing discriminators fall back to node defaults at runtime, so they don't block. + const blockingErrors = validation.errors.filter((error) => !error.missingDiscriminator); + if (blockingErrors.length > 0) { + return { + status: 'error' as const, + error: { + message: `Node parameters do not match the schema for ${input.type} v${input.version}`, + issues: blockingErrors.map(({ path, message }) => ({ path, message })), + }, + }; + } + + // Executing one node is equivalent to running a one-node workflow, so the + // `runWorkflow` policy applies as-is. + if (context.permissions?.runWorkflow === 'blocked') { + return { + status: 'error' as const, + denied: true, + reason: 'Action blocked by admin', + }; + } + + const grantKey = buildExecuteNodeSessionGrantKey(input.type, input.config.parameters); + const requireApproval = context.requireRunWorkflowApproval === true; + // A scoped `always_allow` names the workflow IDs it covers, and a standalone node run has no + // workflow ID to match, so a scoped override never covers this call. + const scopedRunOverride = + context.allowedRunWorkflowIds !== undefined || context.allowedRunWorkflowNames !== undefined; + const allowedByScope = + !requireApproval && !scopedRunOverride && context.permissions?.runWorkflow === 'always_allow'; + const allowedBySessionGrant = + !requireApproval && + grantKey !== null && + context.sessionApprovedToolKeys?.has(grantKey) === true; + const needsApproval = !allowedByScope && !allowedBySessionGrant; + + if (needsApproval && (resumeData === undefined || resumeData === null)) { + return await suspend({ + requestId: nanoid(), + ...(await buildExecuteNodeLabels(context, input)), + severity: 'warning' as const, + }); + } + + if (resumeData !== undefined && resumeData !== null && !resumeData.approved) { + return { status: 'error' as const, denied: true, reason: 'User denied the action' }; + } + + if (resumeData?.approved && resumeData.scope === 'session' && grantKey !== null) { + await context.grantSessionToolApproval?.(grantKey); + } + + return await executeNodeService.execute({ + type: input.type, + version: input.version, + config: input.config, + input: input.input, + timeoutMs: input.timeoutMs, + }); +} + // ── Tool factory ──────────────────────────────────────────────────────────── export function createNodesTool( @@ -457,10 +678,12 @@ export function createNodesTool( return new Tool('nodes') .description( - 'Work with n8n node types. Use `suggested` for known workflow categories, `search` for service-specific discovery, `type-definition` before configuring nodes, and `explore-resources` for live credential-backed lists.', + 'Work with n8n node types. Use `suggested` for known workflow categories, `search` for service-specific discovery, `type-definition` before configuring nodes, `explore-resources` for live credential-backed lists, and `execute` to run one node standalone (requires user approval, real side effects).', ) .input(fullInputSchema) - .handler(async (input: FullInput) => { + .suspend(suspendSchema) + .resume(instanceAiApprovalResumeSchema) + .handler(async (input: FullInput, ctx) => { switch (input.action) { case 'list': return await handleList(context, input); @@ -474,6 +697,8 @@ export function createNodesTool( return await handleSuggested(context, input); case 'explore-resources': return await handleExploreResources(context, input); + case 'execute': + return await handleExecute(context, input, ctx.resumeData, ctx.suspend); } }) .build(); diff --git a/packages/@n8n/instance-ai/src/types.ts b/packages/@n8n/instance-ai/src/types.ts index 3073146e25d..ea38c13a33d 100644 --- a/packages/@n8n/instance-ai/src/types.ts +++ b/packages/@n8n/instance-ai/src/types.ts @@ -31,6 +31,7 @@ import type { import type { OutputSchemaLookup, WorkflowJSON } from '@n8n/workflow-sdk'; import type { GenericValue, + IDisplayOptions, INodeInputConfiguration, INodeTypes, ITaskData, @@ -370,6 +371,7 @@ export interface NodeDescription extends NodeSummary { description?: string; default?: unknown; options?: Array<{ name: string; value: string | number | boolean }>; + displayOptions?: IDisplayOptions; }>; credentials?: Array<{ name: string; @@ -678,6 +680,34 @@ export interface InstanceAiExecutionService { ): Promise; } +export type ExecuteNodeResult = + | { + status: 'success'; + /** Serialized output items, wrapped in the untrusted-data boundary tag. */ + output: string; + truncated?: { totalItems: number; shownItems: number; message: string }; + outputSuppressed?: string; + } + | { status: 'error'; error: { message: string; description?: string; nodeErrorType?: string } }; + +/** Executes a single node standalone through the regular execution engine. + * The request mirrors a workflow-sdk node (`{ type, version, config }`). */ +export interface InstanceAiExecuteNodeService { + execute(request: { + type: string; + version: number; + config: { + parameters: Record; + credentials?: Record< + string, + { id: string | null; name: string; __aiGatewayManaged?: boolean } + >; + }; + input?: Array<{ json: Record }>; + timeoutMs?: number; + }): Promise; +} + export interface CredentialTypeSearchResult { type: string; displayName: string; @@ -1472,6 +1502,8 @@ export interface InstanceAiContext { /** Optional — present when the host allows MCP registry discovery for this * user. Presence gates the `mcp-servers` tool. */ mcpService?: InstanceAiMcpService; + /** Optional — presence gates the `execute` action on the `nodes` tool. */ + executeNodeService?: InstanceAiExecuteNodeService; /** Optional — wired by the host when the run has a bound project. Presence * gates the `conversation-history` tool (orchestrator only). */ conversationHistoryService?: InstanceAiConversationHistoryReader; diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts index abedefd71a5..167ae1dbd70 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts @@ -98,6 +98,7 @@ import { resolveDataTableByIdOrName, resolveMetricProviders, truncateNodeOutput, + redactExecuteNodeResult, truncateResultData, } from '../instance-ai.adapter.service'; import { LlmJudgeProviderRegistry } from '@/evaluation.ee/llm-judge-provider-registry'; @@ -649,6 +650,82 @@ describe('formatExecutionError', () => { }); }); +// --------------------------------------------------------------------------- +// redactExecuteNodeResult +// --------------------------------------------------------------------------- + +describe('redactExecuteNodeResult', () => { + const successResult = (items: Array<{ json: Record }>) => + ({ status: 'success', output: [items] }) as Parameters[0]; + + it('returns the items inside the untrusted-data boundary when sending values is allowed', () => { + const result = redactExecuteNodeResult(successResult([{ json: { id: 1 } }]), true); + + expect(result.status).toBe('success'); + if (result.status !== 'success') return; + expect(result.output).toContain(''); + expect(result.output).toContain('"id": 1'); + expect(result.output.trimEnd().endsWith('')).toBe(true); + }); + + it('caps oversized output and reports shown vs total items', () => { + const items = Array.from({ length: 100 }, (_, i) => ({ + json: { id: i, payload: 'x'.repeat(80) }, + })); + + const result = redactExecuteNodeResult(successResult(items), true); + + expect(result.status).toBe('success'); + if (result.status !== 'success') return; + expect(result.truncated).toEqual( + expect.objectContaining({ totalItems: 100, shownItems: expect.any(Number) }), + ); + const shown = result.truncated?.shownItems ?? 0; + expect(shown).toBeLessThan(100); + expect(result.output.match(/"payload"/g)).toHaveLength(shown); + }); + + it('suppresses output items when sending values is disabled', () => { + const result = redactExecuteNodeResult(successResult([{ json: { secret: 'x' } }]), false); + + expect(result).toEqual({ + status: 'success', + output: '', + outputSuppressed: expect.stringContaining('privacy setting'), + }); + }); + + it('suppresses upstream error details when sending values is disabled', () => { + const result = redactExecuteNodeResult( + { + status: 'error', + error: { message: 'boom', description: 'api key leaked', nodeErrorType: 'NodeApiError' }, + }, + false, + ); + + expect(result).toEqual({ + status: 'error', + error: { + message: 'boom', + description: expect.stringContaining('suppressed'), + nodeErrorType: 'NodeApiError', + }, + }); + }); + + it('caps an oversized error description when sending values is allowed', () => { + const result = redactExecuteNodeResult( + { status: 'error', error: { message: 'boom', description: 'y'.repeat(10_000) } }, + true, + ); + + expect(result.status).toBe('error'); + if (result.status !== 'error') return; + expect(result.error.description?.length).toBeLessThanOrEqual(4_001); + }); +}); + // --------------------------------------------------------------------------- // truncateNodeOutput // --------------------------------------------------------------------------- @@ -1527,6 +1604,7 @@ function createNodeAdapterServiceForTests( loadNodesAndCredentials?: Record; credentialsService?: Record; credentialsFinderService?: Record; + executeNodeService?: Record; }, ) { const mockUser = { id: 'user-1', role: { slug: 'global:member' } } as unknown as User; @@ -1598,6 +1676,15 @@ function createNodeAdapterServiceForTests( typeof InstanceAiAdapterService >[35], nodeCatalogService, + undefined, + undefined, + undefined, + undefined, + undefined, + undefined, + options?.executeNodeService as unknown as ConstructorParameters< + typeof InstanceAiAdapterService + >[43], ); ( @@ -1613,12 +1700,57 @@ function createNodeAdapterServiceForTests( return { service, + mockUser, nodeService: context.nodeService, credentialService: context.credentialService, nodeCatalogService, }; } +describe('executeNodeService adapter', () => { + const executeRequest = { + type: 'n8n-nodes-base.set', + version: 3, + config: { parameters: {} }, + }; + + beforeEach(() => { + mockedUserHasScopes.mockReset(); + }); + + it('runs the node in the bound project after asserting execute scope there', async () => { + mockedUserHasScopes.mockResolvedValue(true); + const executeNodeService = { + run: vi.fn().mockResolvedValue({ status: 'success', output: [] }), + }; + const { service, mockUser } = createNodeAdapterServiceForTests([], { executeNodeService }); + + const context = service.createContext(mockUser, { projectId: 'team-project-1' }); + await context.executeNodeService?.execute(executeRequest); + + expect(mockedUserHasScopes).toHaveBeenCalledWith(mockUser, ['workflow:execute'], false, { + projectId: 'team-project-1', + }); + expect(executeNodeService.run).toHaveBeenCalledWith( + mockUser, + expect.objectContaining({ projectId: 'team-project-1' }), + ); + }); + + it('does not run when the user lacks execute scope in the bound project', async () => { + mockedUserHasScopes.mockResolvedValue(false); + const executeNodeService = { run: vi.fn() }; + const { service, mockUser } = createNodeAdapterServiceForTests([], { executeNodeService }); + + const context = service.createContext(mockUser, { projectId: 'team-project-1' }); + + await expect(context.executeNodeService?.execute(executeRequest)).rejects.toThrow( + 'required permissions', + ); + expect(executeNodeService.run).not.toHaveBeenCalled(); + }); +}); + function createNodeAdapterForTests( nodes: Array>, nodeCatalogService?: Mocked, diff --git a/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts b/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts index cd10464fd77..914db3b1483 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts @@ -81,6 +81,8 @@ import type { UpsertEvaluationConfigInput, InstanceAiActivityService, InstanceAiMcpService, + InstanceAiExecuteNodeService, + ExecuteNodeResult as InstanceAiExecuteNodeResult, McpRegistryConnectServerSummary, McpRegistryServerSummary, ModelConfig, @@ -102,6 +104,7 @@ import type { WorkflowJSON } from '@n8n/workflow-sdk'; import { InstanceSettings } from 'n8n-core'; import { type ICredentialsDecrypted, + type IDataObject, type INode, type INodeParameters, type INodeProperties, @@ -162,6 +165,8 @@ import type { McpRegistrySearchResult } from '@/modules/mcp-registry/registry/mc import { McpRegistryService } from '@/modules/mcp-registry/registry/mcp-registry.service'; import { WorkflowDependencyQueryService } from '@/modules/workflow-index/workflow-dependency-query.service'; import { NodeCatalogService } from '@/node-catalog'; +import { ExecuteNodeService } from '@/node-execution'; +import type { ExecuteNodeResult } from '@/node-execution'; import { NodeTypes } from '@/node-types'; import { userHasScopes } from '@/permissions.ee/check-access'; import { PolicyEnforcementService } from '@/policy/policy-enforcement.service'; @@ -405,6 +410,7 @@ export class InstanceAiAdapterService { private readonly folderRepository?: FolderRepository, private readonly folderFinderService?: FolderFinderService, private readonly instanceContext?: InstanceContextService, + private readonly executeNodeService?: ExecuteNodeService, ) { this.logger = logger.scoped('instance-ai'); this.allowSendingParameterValues = globalConfig.ai.allowSendingParameterValues; @@ -499,6 +505,9 @@ export class InstanceAiAdapterService { } : {}), mcpService: mcpConnectionsEnabled ? this.createMcpAdapter(user) : undefined, + executeNodeService: this.executeNodeService + ? this.createExecuteNodeAdapter(this.executeNodeService, user, projectId) + : undefined, conversationHistoryService: conversationHistory, // Presence is the gate, as with the services above: no reader, no `activity` tool. ...(this.instanceContext?.enabled @@ -699,6 +708,32 @@ export class InstanceAiAdapterService { }; } + private createExecuteNodeAdapter( + executeNodeService: ExecuteNodeService, + user: User, + boundProjectId?: string, + ): InstanceAiExecuteNodeService { + const { resolveProjectId } = this.createProjectScopeHelpers(user, boundProjectId); + return { + execute: async (request) => { + this.assertInstanceNotReadOnly('executions'); + const projectId = await resolveProjectId(['workflow:execute']); + const result = await executeNodeService.run(user, { + type: request.type, + version: request.version, + config: { + parameters: request.config.parameters as INodeParameters, + credentials: request.config.credentials, + }, + input: request.input as Array<{ json: IDataObject }> | undefined, + timeoutMs: request.timeoutMs, + projectId, + }); + return redactExecuteNodeResult(result, this.allowSendingParameterValues); + }, + }; + } + /** Reads the connection rows rather than resolving them into loadable servers: * resolving decrypts credentials per connection, and a row that fails to resolve * is still connected (one connection per user+slug). */ @@ -3513,6 +3548,7 @@ export class InstanceAiAdapterService { name: String(o.name), value: o.value, })), + ...(p.displayOptions ? { displayOptions: p.displayOptions } : {}), })), credentials: desc.credentials?.map((c) => ({ name: c.name, @@ -4408,19 +4444,8 @@ export function formatExecutionError( const MAX_NODE_OUTPUT_BYTES = 5_000; export function truncateNodeOutput(items: unknown[]): unknown[] | unknown { - const serialized = JSON.stringify(items); - if (serialized.length <= MAX_NODE_OUTPUT_BYTES) return items; - - // Binary search for the number of items that fit within the limit - const truncated: unknown[] = []; - let size = 2; // account for "[]" - for (const item of items) { - const itemStr = JSON.stringify(item); - // +1 for comma separator, +1 margin - if (size + itemStr.length + 2 > MAX_NODE_OUTPUT_BYTES) break; - truncated.push(item); - size += itemStr.length + 1; - } + const truncated = capItemsBySerializedSize(items, MAX_NODE_OUTPUT_BYTES); + if (truncated === items) return items; return { items: truncated, @@ -4431,6 +4456,87 @@ export function truncateNodeOutput(items: unknown[]): unknown[] | unknown { }; } +/** Returns the input array itself when it fits, so callers can detect truncation by identity. */ +function capItemsBySerializedSize(items: T[], maxBytes: number): T[] { + if (JSON.stringify(items).length <= maxBytes) return items; + + const kept: T[] = []; + let size = 2; // account for "[]" + for (const item of items) { + const itemStr = JSON.stringify(item); + // +1 for comma separator, +1 margin + if (size + itemStr.length + 2 > maxBytes) break; + kept.push(item); + size += itemStr.length + 1; + } + return kept; +} + +const EXECUTE_NODE_DETAILS_SUPPRESSED = + '(upstream error details suppressed by the instance AI privacy setting; ask the user to share the node error from the UI)'; + +/** + * The standalone execution row is deleted after the result is read, so unlike + * workflow runs there is no get-node-output fallback for truncated or + * suppressed output. + */ +export function redactExecuteNodeResult( + result: ExecuteNodeResult, + allowSendingParameterValues: boolean, +): InstanceAiExecuteNodeResult { + if (result.status === 'error') { + const { message, description, nodeErrorType } = result.error; + const cappedDescription = + description && description.length > MAX_ERROR_CHARS + ? `${description.slice(0, MAX_ERROR_CHARS)}…` + : description; + return { + status: 'error', + error: { + message, + description: allowSendingParameterValues + ? cappedDescription + : description + ? EXECUTE_NODE_DETAILS_SUPPRESSED + : undefined, + nodeErrorType, + }, + }; + } + + if (!allowSendingParameterValues) { + return { + status: 'success', + output: '', + outputSuppressed: + 'The node executed successfully, but its output items are suppressed by the instance AI privacy setting.', + }; + } + + let totalItems = 0; + let shownItems = 0; + const kept = result.output.map((branch) => { + totalItems += branch.length; + const keptBranch = capItemsBySerializedSize(branch, MAX_NODE_OUTPUT_BYTES); + shownItems += keptBranch.length; + return keptBranch; + }); + // The items come from whatever service the node called, so they carry the same + // boundary tag the workflow-execution path puts on node output. + const output = wrapUntrustedData(JSON.stringify(kept, null, 2), 'execution-output'); + if (shownItems === totalItems) return { status: 'success', output }; + + return { + status: 'success', + output, + truncated: { + totalItems, + shownItems, + message: `Output truncated: showing ${shownItems} of ${totalItems} items. Re-run with fewer input items or a narrower operation to see more.`, + }, + }; +} + /** Maximum characters for a single item returned by get-node-output. */ const MAX_ITEM_CHARS = 50_000; diff --git a/packages/cli/src/node-execution/__tests__/execute-node.service.test.ts b/packages/cli/src/node-execution/__tests__/execute-node.service.test.ts new file mode 100644 index 00000000000..672a4f02875 --- /dev/null +++ b/packages/cli/src/node-execution/__tests__/execute-node.service.test.ts @@ -0,0 +1,402 @@ +import { Logger } from '@n8n/backend-common'; +import { mockInstance } from '@n8n/backend-test-utils'; +import { WorkflowRepository } from '@n8n/db'; +import type { User } from '@n8n/db'; +import { InstanceSettings } from 'n8n-core'; +import type { INodeType, INodeTypeDescription, IRunExecutionData } from 'n8n-workflow'; +import { mock } from 'vitest-mock-extended'; + +import { ActiveExecutions } from '@/active-executions'; +import { CredentialsFinderService } from '@/credentials/credentials-finder.service'; +import { BadRequestError } from '@/errors/response-errors/bad-request.error'; +import { ForbiddenError } from '@/errors/response-errors/forbidden.error'; +import { ExecutionPersistence } from '@/executions/execution-persistence'; +import { NodeTypes } from '@/node-types'; +import { WorkflowRunner } from '@/workflow-runner'; + +import { ExecuteNodeService } from '../execute-node.service'; +import type { ExecuteNodeRequest } from '../execute-node.service'; + +const testNodeDescription: INodeTypeDescription = { + displayName: 'Test', + name: 'test', + group: [], + version: 1, + description: '', + defaults: { name: 'Test' }, + inputs: [], + outputs: [], + properties: [], +}; + +// Assign onto an empty mock instead of passing overrides to mock(): the latter +// deep-wraps nested objects in proxies. `description` must be a real object. +const mockNodeType = (overrides: object = {}) => + Object.assign(mock(), { description: testNodeDescription }, overrides); + +const baseRequest = (overrides: Partial = {}): ExecuteNodeRequest => ({ + type: 'n8n-nodes-base.set', + version: 3, + config: { parameters: {} }, + projectId: 'project-1', + ...overrides, +}); + +const successExecution = (runDataForNode: unknown) => ({ + data: { resultData: { runData: { Node: runDataForNode } } } as unknown as IRunExecutionData, +}); + +describe('ExecuteNodeService', () => { + const nodeTypes = mockInstance(NodeTypes); + const credentialsFinderService = mockInstance(CredentialsFinderService); + const logger = mockInstance(Logger); + const workflowRepository = mockInstance(WorkflowRepository); + const workflowRunner = mockInstance(WorkflowRunner); + const activeExecutions = mockInstance(ActiveExecutions); + const executionPersistence = mockInstance(ExecutionPersistence); + const instanceSettings = mockInstance(InstanceSettings); + + const service = new ExecuteNodeService( + nodeTypes, + credentialsFinderService, + logger, + workflowRepository, + workflowRunner, + activeExecutions, + executionPersistence, + instanceSettings, + ); + + const user = mock({ id: 'user-1' }); + + beforeEach(() => { + vi.clearAllMocks(); + + Object.defineProperty(instanceSettings, 'isMultiMain', { value: false, configurable: true }); + nodeTypes.getByNameAndVersion.mockReturnValue(mockNodeType()); + workflowRepository.createWorkflowWithOwner.mockImplementation(async (workflow) => { + workflow.id = 'temp-wf-1'; + return await Promise.resolve(workflow); + }); + workflowRunner.run.mockResolvedValue('exec-1'); + activeExecutions.has.mockReturnValue(true); + activeExecutions.getPostExecutePromise.mockResolvedValue(undefined); + executionPersistence.findSingleExecution.mockResolvedValue( + successExecution([{ data: { main: [[{ json: { done: true } }]] } }]) as never, + ); + }); + + describe('pre-flight validation', () => { + it('rejects an unknown node type before creating any workflow row', async () => { + nodeTypes.getByNameAndVersion.mockImplementation(() => { + throw new Error('unknown'); + }); + + await expect(service.run(user, baseRequest())).rejects.toThrow(BadRequestError); + expect(workflowRepository.createWorkflowWithOwner).not.toHaveBeenCalled(); + }); + + it('rejects a trigger/webhook-only node before creating any workflow row', async () => { + nodeTypes.getByNameAndVersion.mockReturnValue(mockNodeType({ execute: undefined })); + + await expect(service.run(user, baseRequest())).rejects.toThrow(BadRequestError); + expect(workflowRepository.createWorkflowWithOwner).not.toHaveBeenCalled(); + }); + + it('accepts a declarative node without execute (requestDefaults present)', async () => { + nodeTypes.getByNameAndVersion.mockReturnValue( + mockNodeType({ + execute: undefined, + description: { ...testNodeDescription, requestDefaults: { baseURL: 'https://x' } }, + }), + ); + + const result = await service.run(user, baseRequest()); + + expect(result.status).toBe('success'); + }); + + it('rejects a credential reference without id', async () => { + await expect( + service.run( + user, + baseRequest({ + config: { parameters: {}, credentials: { slackApi: { id: null, name: 'Slack' } } }, + }), + ), + ).rejects.toThrow(BadRequestError); + expect(workflowRepository.createWorkflowWithOwner).not.toHaveBeenCalled(); + }); + + it('rejects a credential the user cannot read', async () => { + credentialsFinderService.findCredentialForUser.mockResolvedValue(null); + + await expect( + service.run( + user, + baseRequest({ + config: { parameters: {}, credentials: { slackApi: { id: 'cred-1', name: 'Slack' } } }, + }), + ), + ).rejects.toThrow(ForbiddenError); + expect(workflowRepository.createWorkflowWithOwner).not.toHaveBeenCalled(); + }); + + it('forces id null on managed credentials and skips the access check for them', async () => { + await service.run( + user, + baseRequest({ + config: { + parameters: {}, + credentials: { + aiGateway: { id: 'stale-id', name: 'Managed', __aiGatewayManaged: true }, + }, + }, + }), + ); + + expect(credentialsFinderService.findCredentialForUser).not.toHaveBeenCalled(); + const runData = workflowRunner.run.mock.calls[0][0]; + expect(runData.workflowData.nodes[0].credentials).toEqual({ + aiGateway: { id: null, name: 'Managed', __aiGatewayManaged: true }, + }); + }); + }); + + describe('engine execution', () => { + it('creates an archived single-node workflow owned by the requested project', async () => { + await service.run(user, baseRequest({ timeoutMs: 10_000 })); + + const savedWorkflow = workflowRepository.createWorkflowWithOwner.mock.calls[0][0]; + expect(savedWorkflow.isArchived).toBe(true); + expect(savedWorkflow.active).toBe(false); + expect(savedWorkflow.nodes).toHaveLength(1); + expect(savedWorkflow.connections).toEqual({}); + expect(savedWorkflow.settings).toEqual( + expect.objectContaining({ + saveManualExecutions: true, + saveDataSuccessExecution: 'all', + saveDataErrorExecution: 'all', + executionTimeout: 10, + }), + ); + expect(workflowRepository.createWorkflowWithOwner).toHaveBeenCalledWith( + savedWorkflow, + 'project-1', + ); + }); + + it('clamps the timeout to the maximum', async () => { + await service.run(user, baseRequest({ timeoutMs: 120_000 })); + + const savedWorkflow = workflowRepository.createWorkflowWithOwner.mock.calls[0][0]; + expect(savedWorkflow.settings?.executionTimeout).toBe(60); + }); + + it('runs through WorkflowRunner in manual mode with a seeded execution stack', async () => { + const input = [{ json: { text: 'hi' } }]; + await service.run(user, baseRequest({ input })); + + const runData = workflowRunner.run.mock.calls[0][0]; + expect(runData.executionMode).toBe('manual'); + expect(runData.userId).toBe('user-1'); + expect(runData.workflowData.id).toBe('temp-wf-1'); + expect(runData.executionData?.manualData).toEqual({ userId: 'user-1' }); + const stack = runData.executionData?.executionData?.nodeExecutionStack; + expect(stack).toHaveLength(1); + expect(stack?.[0]).toEqual( + expect.objectContaining({ + node: expect.objectContaining({ name: 'Node', type: 'n8n-nodes-base.set' }), + data: { main: [input] }, + source: null, + }), + ); + }); + + it('defaults to a single empty input item', async () => { + await service.run(user, baseRequest()); + + const runData = workflowRunner.run.mock.calls[0][0]; + expect(runData.executionData?.executionData?.nodeExecutionStack[0].data).toEqual({ + main: [[{ json: {} }]], + }); + }); + + it('maps the persisted output of the last node run, reducing binary to metadata', async () => { + executionPersistence.findSingleExecution.mockResolvedValue( + successExecution([ + { data: { main: [[{ json: { attempt: 1 } }]] } }, + { + data: { + main: [ + [ + { + json: { attempt: 2 }, + binary: { + file: { + data: 'x'.repeat(1000), + fileName: 'a.pdf', + mimeType: 'application/pdf', + fileSize: '1 kB', + }, + }, + }, + ], + ], + }, + }, + ]) as never, + ); + + const result = await service.run(user, baseRequest()); + + expect(result).toEqual({ + status: 'success', + output: [ + [ + { + json: { attempt: 2 }, + binary: { + file: { fileName: 'a.pdf', mimeType: 'application/pdf', fileSize: '1 kB' }, + }, + }, + ], + ], + }); + }); + + it('maps an execution error to an error result', async () => { + executionPersistence.findSingleExecution.mockResolvedValue({ + data: { + resultData: { + runData: {}, + error: { message: 'boom', description: 'details', name: 'NodeOperationError' }, + }, + }, + } as never); + + const result = await service.run(user, baseRequest()); + + expect(result).toEqual({ + status: 'error', + error: { message: 'boom', description: 'details', nodeErrorType: 'NodeOperationError' }, + }); + }); + + it('maps an error recorded on the node run itself to an error result', async () => { + executionPersistence.findSingleExecution.mockResolvedValue( + successExecution([ + { error: { message: 'node failed', name: 'NodeApiError' }, data: undefined }, + ]) as never, + ); + + const result = await service.run(user, baseRequest()); + + expect(result).toEqual({ + status: 'error', + error: { message: 'node failed', nodeErrorType: 'NodeApiError' }, + }); + }); + + it('returns an error result when the execution entered a wait state', async () => { + executionPersistence.findSingleExecution.mockResolvedValue({ + status: 'waiting', + ...successExecution([{ data: { main: [[{ json: {} }]] } }]), + } as never); + + const result = await service.run(user, baseRequest()); + + expect(result).toEqual({ + status: 'error', + error: { message: expect.stringContaining('wait state') }, + }); + expect(workflowRepository.delete).toHaveBeenCalledWith('temp-wf-1'); + }); + + it('returns an error result when the node produced no output', async () => { + executionPersistence.findSingleExecution.mockResolvedValue( + successExecution(undefined) as never, + ); + + const result = await service.run(user, baseRequest()); + + expect(result).toEqual({ + status: 'error', + error: { message: expect.stringContaining('produced no output') }, + }); + }); + + it('deletes the temporary workflow after a successful run', async () => { + await service.run(user, baseRequest()); + + expect(workflowRepository.delete).toHaveBeenCalledWith('temp-wf-1'); + }); + + it('hard-deletes the executions before the workflow row, so their stored data goes too', async () => { + await service.run(user, baseRequest()); + + expect(executionPersistence.hardDeleteByWorkflowId).toHaveBeenCalledWith('temp-wf-1'); + expect(executionPersistence.hardDeleteByWorkflowId.mock.invocationCallOrder[0]).toBeLessThan( + workflowRepository.delete.mock.invocationCallOrder[0], + ); + }); + + it('still deletes the workflow row when hard-deleting the executions fails', async () => { + executionPersistence.hardDeleteByWorkflowId.mockRejectedValue(new Error('blob store down')); + + const result = await service.run(user, baseRequest()); + + expect(result.status).toBe('success'); + expect(workflowRepository.delete).toHaveBeenCalledWith('temp-wf-1'); + }); + + it('deletes the temporary workflow when the launch fails and returns an error result', async () => { + workflowRunner.run.mockRejectedValue(new Error('launch failed')); + + const result = await service.run(user, baseRequest()); + + expect(result).toEqual({ status: 'error', error: { message: 'launch failed' } }); + expect(workflowRepository.delete).toHaveBeenCalledWith('temp-wf-1'); + }); + + it('cancels the execution and reports a timeout when it never settles', async () => { + vi.useFakeTimers(); + try { + activeExecutions.getPostExecutePromise.mockReturnValue(new Promise(() => {})); + + const promise = service.run(user, baseRequest({ timeoutMs: 5_000 })); + await vi.advanceTimersByTimeAsync(7_000); + const result = await promise; + + expect(result).toEqual({ + status: 'error', + error: { message: 'Execution timed out after 5000ms and was cancelled' }, + }); + expect(activeExecutions.stopExecution).toHaveBeenCalledWith('exec-1', expect.any(Error)); + expect(workflowRepository.delete).toHaveBeenCalledWith('temp-wf-1'); + } finally { + vi.useRealTimers(); + } + }); + + it('polls the execution row instead of the post-execute promise on multi-main', async () => { + Object.defineProperty(instanceSettings, 'isMultiMain', { value: true, configurable: true }); + vi.useFakeTimers(); + try { + executionPersistence.findSingleExecution + .mockResolvedValueOnce({ status: 'running' } as never) + .mockResolvedValueOnce({ status: 'success' } as never); + + const promise = service.run(user, baseRequest()); + await vi.advanceTimersByTimeAsync(1_500); + const result = await promise; + + expect(activeExecutions.getPostExecutePromise).not.toHaveBeenCalled(); + expect(executionPersistence.findSingleExecution).toHaveBeenCalledTimes(3); + expect(result).toEqual({ status: 'success', output: [[{ json: { done: true } }]] }); + } finally { + vi.useRealTimers(); + } + }); + }); +}); diff --git a/packages/cli/src/node-execution/execute-node.service.ts b/packages/cli/src/node-execution/execute-node.service.ts new file mode 100644 index 00000000000..514b52da2de --- /dev/null +++ b/packages/cli/src/node-execution/execute-node.service.ts @@ -0,0 +1,410 @@ +import { Logger } from '@n8n/backend-common'; +import { WorkflowEntity, WorkflowRepository } from '@n8n/db'; +import type { User } from '@n8n/db'; +import { Service } from '@n8n/di'; +import { getErrorMessage } from '@n8n/utils/errors/get-error-message'; +import { sleep } from '@n8n/utils/sleep'; +import { InstanceSettings } from 'n8n-core'; +import { createRunExecutionData, NodeError, TimeoutExecutionCancelledError } from 'n8n-workflow'; +import type { + IDataObject, + INode, + INodeCredentials, + INodeExecutionData, + INodeParameters, + INodeType, +} from 'n8n-workflow'; +import { v4 as uuid } from 'uuid'; + +import { ActiveExecutions } from '@/active-executions'; +import { CredentialsFinderService } from '@/credentials/credentials-finder.service'; +import { BadRequestError } from '@/errors/response-errors/bad-request.error'; +import { ForbiddenError } from '@/errors/response-errors/forbidden.error'; +import { ExecutionPersistence } from '@/executions/execution-persistence'; +import { NodeTypes } from '@/node-types'; +import { WorkflowRunner } from '@/workflow-runner'; + +export const DEFAULT_EXECUTE_NODE_TIMEOUT_MS = 30_000; +export const MAX_EXECUTE_NODE_TIMEOUT_MS = 60_000; + +const NODE_NAME = 'Node'; +const MULTI_MAIN_POLL_INTERVAL_MS = 1_000; + +/** Mirrors a workflow-sdk node (`{ type, version, config }`) so callers can + * pass a node they are building verbatim. */ +export interface ExecuteNodeRequest { + type: string; + version: number; + config: { + parameters: INodeParameters; + credentials?: INodeCredentials; + }; + input?: Array<{ json: IDataObject }>; + timeoutMs?: number; + projectId: string; +} + +export interface ExecuteNodeOutputItem { + json: IDataObject; + binary?: Record; +} + +export interface ExecuteNodeError { + message: string; + description?: string; + nodeErrorType?: string; +} + +export type ExecuteNodeResult = + | { status: 'success'; output: ExecuteNodeOutputItem[][] } + | { status: 'error'; error: ExecuteNodeError }; + +/** + * Executes a single node standalone with real credentials and caller-supplied + * input items. + * + * Runs through `WorkflowRunner` so the configured isolation applies — in queue + * mode node code runs on a worker, never in a process the engine would not run + * it in (task runners only isolate Code-node scripts). `WorkflowRunner` + * requires a persisted workflow row (FK on `execution_entity.workflowId`), so + * an archived single-node workflow is created per run — the chat-hub pattern — + * and deleted afterwards. + * + * `findCredentialForUser` is the caller-facing authorization gate; nothing + * downstream re-checks per user. The temp workflow is created in the caller's + * `projectId` (the Instance AI conversation's project), and the engine's + * project-level check runs against it, so credentials shared with that project + * are usable while anything else fails at run start (fails closed). + */ +@Service() +export class ExecuteNodeService { + constructor( + private readonly nodeTypes: NodeTypes, + private readonly credentialsFinderService: CredentialsFinderService, + private readonly logger: Logger, + private readonly workflowRepository: WorkflowRepository, + private readonly workflowRunner: WorkflowRunner, + private readonly activeExecutions: ActiveExecutions, + private readonly executionPersistence: ExecutionPersistence, + private readonly instanceSettings: InstanceSettings, + ) {} + + async run(user: User, request: ExecuteNodeRequest): Promise { + const nodeType = this.resolveNodeType(request.type, request.version); + this.assertExecutable(nodeType, request.type); + + const credentials = request.config.credentials + ? this.normalizeCredentials(request.config.credentials) + : undefined; + await this.checkCredentialAccess(user, credentials); + + const node = this.buildNode(request, credentials); + const timeoutMs = Math.min( + request.timeoutMs ?? DEFAULT_EXECUTE_NODE_TIMEOUT_MS, + MAX_EXECUTE_NODE_TIMEOUT_MS, + ); + const inputItems: INodeExecutionData[] = request.input ?? [{ json: {} }]; + + const workflow = await this.createTemporaryWorkflow(request.projectId, node, timeoutMs); + try { + return await this.execute(user, workflow, node, inputItems, timeoutMs); + } catch (error) { + return { status: 'error', error: this.toErrorResponse(error) }; + } finally { + await this.deleteTemporaryWorkflow(workflow.id); + } + } + + private resolveNodeType(type: string, version: number): INodeType { + try { + return this.nodeTypes.getByNameAndVersion(type, version); + } catch (error) { + throw new BadRequestError( + `Unknown node type "${type}" (version ${version})`, + undefined, + undefined, + error, + ); + } + } + + /** `requestDefaults` marks a declarative node, which the engine can route + * without an `execute` method; trigger/webhook-only nodes have neither. */ + private assertExecutable(nodeType: INodeType, type: string) { + if (nodeType.execute) return; + if (nodeType.description.requestDefaults !== undefined) return; + + throw new BadRequestError( + `Node type "${type}" has no execute method and no routing support (e.g. it is trigger/webhook-only)`, + ); + } + + /** Managed credentials (AI Gateway) are minted per execution and carry no + * stored row — force `id: null` so a stale or foreign id never reaches the node. */ + private normalizeCredentials(credentials: INodeCredentials): INodeCredentials { + const normalized: INodeCredentials = {}; + for (const [credentialType, credential] of Object.entries(credentials)) { + normalized[credentialType] = credential.__aiGatewayManaged + ? { id: null, name: credential.name, __aiGatewayManaged: true } + : credential; + } + return normalized; + } + + private async checkCredentialAccess(user: User, credentials?: INodeCredentials) { + for (const [credentialType, credential] of Object.entries(credentials ?? {})) { + if (credential.__aiGatewayManaged) continue; + + if (!credential.id) { + throw new BadRequestError(`Credential reference for "${credentialType}" is missing an id`); + } + + const accessible = await this.credentialsFinderService.findCredentialForUser( + credential.id, + user, + ['credential:read'], + ); + if (!accessible) { + throw new ForbiddenError( + `You do not have access to the "${credentialType}" credential used by this node`, + ); + } + } + } + + private buildNode(request: ExecuteNodeRequest, credentials?: INodeCredentials): INode { + return { + id: uuid(), + name: NODE_NAME, + type: request.type, + typeVersion: request.version, + position: [0, 0], + parameters: request.config.parameters, + credentials, + }; + } + + private async createTemporaryWorkflow( + projectId: string, + node: INode, + timeoutMs: number, + ): Promise { + const newWorkflow = new WorkflowEntity(); + newWorkflow.isArchived = true; + newWorkflow.versionId = uuid(); + newWorkflow.name = `Execute node ${node.type}`; + newWorkflow.active = false; + newWorkflow.activeVersionId = null; + newWorkflow.nodes = [node]; + newWorkflow.connections = {}; + newWorkflow.settings = { + executionOrder: 'v1', + // Force-save: the execution row is the only result channel that works in queue mode. + saveManualExecutions: true, + saveDataSuccessExecution: 'all', + saveDataErrorExecution: 'all', + executionTimeout: Math.ceil(timeoutMs / 1000), + }; + + return await this.workflowRepository.createWorkflowWithOwner(newWorkflow, projectId); + } + + /** delete temporary workflow and executions */ + private async deleteTemporaryWorkflow(workflowId: string) { + try { + await this.executionPersistence.hardDeleteByWorkflowId(workflowId); + } catch (error) { + this.warnCleanupFailed(workflowId, error); + } + + try { + await this.workflowRepository.delete(workflowId); + } catch (error) { + this.warnCleanupFailed(workflowId, error); + } + } + + private warnCleanupFailed(workflowId: string, error: unknown) { + this.logger.warn('Failed to clean up temporary execute-node workflow', { + workflowId, + error: getErrorMessage(error), + }); + } + + private async execute( + user: User, + workflowData: WorkflowEntity, + node: INode, + inputItems: INodeExecutionData[], + timeoutMs: number, + ): Promise { + const executionData = createRunExecutionData({ + startData: {}, + resultData: { runData: {} }, + executionData: { + contextData: {}, + metadata: {}, + nodeExecutionStack: [{ node, data: { main: [inputItems] }, source: null }], + waitingExecution: {}, + waitingExecutionSource: {}, + }, + // Top-level runData fields don't survive queue serialization; the worker reads this. + manualData: { userId: user.id }, + }); + + const executionId = await this.workflowRunner.run({ + executionMode: 'manual', + workflowData, + executionData, + userId: user.id, + }); + + const timedOut = await this.waitForCompletion(executionId, timeoutMs); + if (timedOut) { + return { + status: 'error', + error: { message: `Execution timed out after ${timeoutMs}ms and was cancelled` }, + }; + } + + return await this.extractResult(executionId, node.type); + } + + private async waitForCompletion(executionId: string, timeoutMs: number): Promise { + if (!this.activeExecutions.has(executionId)) return false; + + const abort = new AbortController(); + let timeoutId: NodeJS.Timeout | undefined; + // The workflow's executionTimeout is the primary bound; this race only + // catches an execution that never settles. + const timeout = new Promise((_, reject) => { + timeoutId = setTimeout( + () => reject(new TimeoutExecutionCancelledError(executionId)), + timeoutMs + 1_000, + ); + }); + + try { + await Promise.race([this.waitForSettled(executionId, abort.signal), timeout]); + return false; + } catch (error) { + if (error instanceof TimeoutExecutionCancelledError) { + abort.abort(); + try { + this.activeExecutions.stopExecution(executionId, error); + } catch { + // Execution may have completed between timeout and cancel + } + return true; + } + throw error; + } finally { + clearTimeout(timeoutId); + } + } + + /** The post-execute promise does not reliably settle on multi-main (Bull's + * `job.finished()` behind it), so poll the execution row there — same as + * chat-hub's `waitForExecutionCompletion`. */ + private async waitForSettled(executionId: string, signal: AbortSignal): Promise { + if (!this.instanceSettings.isMultiMain) { + await this.activeExecutions.getPostExecutePromise(executionId); + return; + } + + while (!signal.aborted) { + const execution = await this.executionPersistence.findSingleExecution(executionId, { + includeData: false, + unflattenData: false, + }); + const inFlight = execution?.status === 'new' || execution?.status === 'running'; + if (!inFlight) return; + await sleep(MULTI_MAIN_POLL_INTERVAL_MS, signal); + } + } + + private async extractResult( + executionId: string, + nodeTypeName: string, + ): Promise { + const execution = await this.executionPersistence.findSingleExecution(executionId, { + includeData: true, + unflattenData: true, + }); + + // The temp workflow is deleted after this read, so a waiting execution + // could never be resumed — report it instead of a false success. + if (execution?.status === 'waiting' || execution?.waitTill) { + return { + status: 'error', + error: { + message: `Node type "${nodeTypeName}" entered a wait state; standalone node execution does not support waiting or resuming (e.g. Wait nodes or send-and-wait operations)`, + }, + }; + } + + const executionError = execution?.data?.resultData?.error; + if (executionError) { + return { status: 'error', error: this.toErrorResponse(executionError) }; + } + + const nodeRuns = execution?.data?.resultData?.runData?.[NODE_NAME]; + const lastRun = nodeRuns?.[nodeRuns.length - 1]; + if (lastRun?.error) { + return { status: 'error', error: this.toErrorResponse(lastRun.error) }; + } + + const branches = lastRun?.data?.main; + if (!branches) { + return { + status: 'error', + error: { message: `Node type "${nodeTypeName}" produced no output` }, + }; + } + + return { status: 'success', output: this.toOutput(branches) }; + } + + private toOutput(branches: Array): ExecuteNodeOutputItem[][] { + return branches.map((branch) => (branch ?? []).map((item) => this.toOutputItem(item))); + } + + private toOutputItem(item: INodeExecutionData): ExecuteNodeOutputItem { + if (!item.binary) return { json: item.json }; + + const binary: ExecuteNodeOutputItem['binary'] = {}; + for (const [key, value] of Object.entries(item.binary)) { + binary[key] = { + fileName: value.fileName, + mimeType: value.mimeType, + fileSize: value.fileSize, + }; + } + return { json: item.json, binary }; + } + + private toErrorResponse(error: unknown): ExecuteNodeError { + if (error instanceof NodeError) { + return { + message: error.message, + description: error.description ?? undefined, + nodeErrorType: error.constructor.name, + }; + } + if (error instanceof Error) { + this.logger.debug('Execute node failed', { error: error.message }); + return { message: error.message }; + } + // Errors read back from a persisted execution are revived as plain + // objects, not Error instances. + if (typeof error === 'object' && error !== null && 'message' in error) { + const revived = error as { message: string; description?: string | null; name?: string }; + return { + message: revived.message, + description: revived.description ?? undefined, + nodeErrorType: revived.name, + }; + } + return { message: 'Unknown error' }; + } +} diff --git a/packages/cli/src/node-execution/index.ts b/packages/cli/src/node-execution/index.ts index a2bdd322dd5..1f7b57719aa 100644 --- a/packages/cli/src/node-execution/index.ts +++ b/packages/cli/src/node-execution/index.ts @@ -13,3 +13,14 @@ export { isUnsupportedEphemeralNodeOperation, unsupportedEphemeralNodeOperationMessage, } from './node-tool-operation-support'; +export type { + ExecuteNodeRequest, + ExecuteNodeOutputItem, + ExecuteNodeError, + ExecuteNodeResult, +} from './execute-node.service'; +export { + ExecuteNodeService, + DEFAULT_EXECUTE_NODE_TIMEOUT_MS, + MAX_EXECUTE_NODE_TIMEOUT_MS, +} from './execute-node.service'; diff --git a/packages/frontend/@n8n/i18n/src/locales/en.json b/packages/frontend/@n8n/i18n/src/locales/en.json index f2bfe306757..65ebbd4d159 100644 --- a/packages/frontend/@n8n/i18n/src/locales/en.json +++ b/packages/frontend/@n8n/i18n/src/locales/en.json @@ -7561,6 +7561,8 @@ "instanceAi.tools.nodes.type-definition": "Loading node schema", "instanceAi.tools.nodes.suggested": "Getting suggested nodes", "instanceAi.tools.nodes.explore-resources": "Exploring node resources", + "instanceAi.tools.nodes.execute": "Executing node", + "instanceAi.tools.nodes.execute.imperativeWithResource": "execute the", "instanceAi.tools.task-control": "Task control", "instanceAi.tools.task-control.update-checklist": "Updating tasks", "instanceAi.tools.task-control.cancel-task": "Cancelling task", diff --git a/packages/frontend/editor-ui/src/features/ai/instanceAi/__tests__/InstanceAiConfirmationPanel.test.ts b/packages/frontend/editor-ui/src/features/ai/instanceAi/__tests__/InstanceAiConfirmationPanel.test.ts index 558ca478f66..f9327f9e87a 100644 --- a/packages/frontend/editor-ui/src/features/ai/instanceAi/__tests__/InstanceAiConfirmationPanel.test.ts +++ b/packages/frontend/editor-ui/src/features/ai/instanceAi/__tests__/InstanceAiConfirmationPanel.test.ts @@ -42,6 +42,7 @@ vi.mock('@n8n/i18n', async (importOriginal) => ({ 'instanceAi.tools.build-workflow.imperativeWithResource': 'edit', 'instanceAi.tools.data-tables.add-column.imperative': 'add column', 'instanceAi.tools.data-tables.add-column.imperativeWithResource': 'add a column to', + 'instanceAi.tools.nodes.execute.imperativeWithResource': 'execute the', }; if (key === 'agents.chat.approval.description') { return `The agent wants to run the ${opts?.interpolate?.toolName ?? ''} tool.`; @@ -356,6 +357,24 @@ describe('InstanceAiConfirmationPanel telemetry', () => { expect(getByText('Allow n8n Assistant to edit workflow?')).toBeVisible(); }); + it('names the node in the title when the assistant executes one', () => { + injectPendingConfirmation( + thread, + { + requestId: 'execute-node', + severity: 'warning', + message: 'Document > Create', + resourceName: 'Google Sheets node', + }, + { action: 'execute', type: 'n8n-nodes-base.googleSheets' }, + 'nodes', + ); + const { getByText } = renderComponent({ props: { kind: 'floating' } }); + + expect(getByText('Assistant wants to execute the Google Sheets node')).toBeVisible(); + expect(getByText('Document > Create')).toBeVisible(); + }); + it('preserves the saved description after a question mark', () => { const message = 'Save changes to workflow "Paid?": Add a Slack notification after the payment check and update the daily report'; diff --git a/packages/frontend/editor-ui/src/features/ai/instanceAi/components/InstanceAiConfirmationPanel.vue b/packages/frontend/editor-ui/src/features/ai/instanceAi/components/InstanceAiConfirmationPanel.vue index 55eff1ee779..9f55100b69e 100644 --- a/packages/frontend/editor-ui/src/features/ai/instanceAi/components/InstanceAiConfirmationPanel.vue +++ b/packages/frontend/editor-ui/src/features/ai/instanceAi/components/InstanceAiConfirmationPanel.vue @@ -140,6 +140,7 @@ const approvalTitleKeys = new Map( 'instanceAi.tools.workflows.update-version.imperativeWithResource', 'instanceAi.tools.workflows.restore-version.imperative', 'instanceAi.tools.workflows.restore-version.imperativeWithResource', + 'instanceAi.tools.nodes.execute.imperativeWithResource', 'instanceAi.tools.executions.run.imperative', 'instanceAi.tools.executions.run.imperativeWithResource', 'instanceAi.tools.credentials.delete.imperative', diff --git a/packages/frontend/editor-ui/src/features/ai/instanceAi/instanceAi.threadRuntime.ts b/packages/frontend/editor-ui/src/features/ai/instanceAi/instanceAi.threadRuntime.ts index 1a9a00a8bb1..162f859c7f1 100644 --- a/packages/frontend/editor-ui/src/features/ai/instanceAi/instanceAi.threadRuntime.ts +++ b/packages/frontend/editor-ui/src/features/ai/instanceAi/instanceAi.threadRuntime.ts @@ -3,6 +3,7 @@ import { v4 as uuidv4 } from 'uuid'; import { ResponseError } from '@n8n/rest-api-client'; import { buildDataTablesSessionGrantKey, + buildExecuteNodeSessionGrantKey, buildRunStepSessionGrantKey, buildRunWorkflowSessionGrantKey, buildUpdateWorkflowSessionGrantKey, @@ -29,6 +30,7 @@ import { type AgentRunState, type InstanceAiRunLimitReason, } from '@n8n/api-types'; +import { isRecord } from '@n8n/utils/is-record'; import { useRootStore } from '@n8n/stores/useRootStore'; import { useInstanceAiSettingsStore } from './instanceAiSettings.store'; import { redactTelemetryProperties, TELEMETRY_EVENT } from '@n8n/telemetry'; @@ -899,6 +901,15 @@ export function createThreadRuntime( if (toolName === 'data-tables') { return buildDataTablesSessionGrantKey(action); } + // Executing a node grants "always allow" per node type + resource + operation, + // mirroring the backend thread grant. Without a type, fail closed. + if (toolName === 'nodes' && action === 'execute') { + const nodeType = typeof args.type === 'string' ? args.type : ''; + if (!nodeType) return null; + const config = isRecord(args.config) ? args.config : undefined; + const parameters = isRecord(config?.parameters) ? config.parameters : undefined; + return buildExecuteNodeSessionGrantKey(nodeType, parameters); + } return `${toolName}:${action}`; }