mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 13:39:44 +02:00
feat: Add standalone node execution as a Workflow Builder and AI Assistant tool (#37512)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: Jaakko Husso <jaakko@n8n.io>
This commit is contained in:
parent
d42a753176
commit
2d56eee224
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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');
|
||||
|
|
|
|||
|
|
@ -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, unknown>,
|
||||
): 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").
|
||||
|
|
|
|||
|
|
@ -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<TransactionRunner>();
|
||||
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<unknown>) =>
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
|
@ -308,6 +308,28 @@ export class WorkflowRepository extends BaseRepository<WorkflowEntity> {
|
|||
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<WorkflowEntity> {
|
||||
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.
|
||||
*
|
||||
|
|
|
|||
|
|
@ -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:<type>:<resource>:<operation>`) —
|
||||
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 `<untrusted_data
|
||||
source="execution-output">` 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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -640,6 +640,8 @@ export type {
|
|||
InstanceAiToolRegistry,
|
||||
InstanceAiWorkflowService,
|
||||
InstanceAiExecutionService,
|
||||
InstanceAiExecuteNodeService,
|
||||
ExecuteNodeResult,
|
||||
InstanceAiCredentialService,
|
||||
InstanceAiNodeService,
|
||||
InstanceAiDataTableService,
|
||||
|
|
|
|||
|
|
@ -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<typeof import('@n8n/workflow-sdk')>()),
|
||||
validateNodeConfig: vi.fn(() => ({ valid: true, errors: [] })),
|
||||
}));
|
||||
|
||||
function createMockContext(overrides: Partial<InstanceAiContext> = {}): 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<string, unknown>,
|
||||
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' });
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<typeof executeAction>;
|
||||
|
||||
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<typeof instanceAiApprovalResumeSchema> | undefined | null,
|
||||
suspend: (payload: z.infer<typeof suspendSchema>) => Promise<never>,
|
||||
) {
|
||||
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('.') || '<root>'}: ${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();
|
||||
|
|
|
|||
|
|
@ -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<ResolvedNodeParametersResult>;
|
||||
}
|
||||
|
||||
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<string, unknown>;
|
||||
credentials?: Record<
|
||||
string,
|
||||
{ id: string | null; name: string; __aiGatewayManaged?: boolean }
|
||||
>;
|
||||
};
|
||||
input?: Array<{ json: Record<string, unknown> }>;
|
||||
timeoutMs?: number;
|
||||
}): Promise<ExecuteNodeResult>;
|
||||
}
|
||||
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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<string, unknown> }>) =>
|
||||
({ status: 'success', output: [items] }) as Parameters<typeof redactExecuteNodeResult>[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('<untrusted_data source="execution-output">');
|
||||
expect(result.output).toContain('"id": 1');
|
||||
expect(result.output.trimEnd().endsWith('</untrusted_data>')).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<string, unknown>;
|
||||
credentialsService?: Record<string, unknown>;
|
||||
credentialsFinderService?: Record<string, unknown>;
|
||||
executeNodeService?: Record<string, unknown>;
|
||||
},
|
||||
) {
|
||||
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<Record<string, unknown>>,
|
||||
nodeCatalogService?: Mocked<NodeCatalogService>,
|
||||
|
|
|
|||
|
|
@ -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<T>(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;
|
||||
|
||||
|
|
|
|||
|
|
@ -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<INodeType>(), { description: testNodeDescription }, overrides);
|
||||
|
||||
const baseRequest = (overrides: Partial<ExecuteNodeRequest> = {}): 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<User>({ 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();
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
410
packages/cli/src/node-execution/execute-node.service.ts
Normal file
410
packages/cli/src/node-execution/execute-node.service.ts
Normal file
|
|
@ -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<string, { fileName?: string; mimeType?: string; fileSize?: string }>;
|
||||
}
|
||||
|
||||
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<ExecuteNodeResult> {
|
||||
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<WorkflowEntity> {
|
||||
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<ExecuteNodeResult> {
|
||||
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<boolean> {
|
||||
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<never>((_, 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<void> {
|
||||
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<ExecuteNodeResult> {
|
||||
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<INodeExecutionData[] | null>): 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' };
|
||||
}
|
||||
}
|
||||
|
|
@ -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';
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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';
|
||||
|
|
|
|||
|
|
@ -140,6 +140,7 @@ const approvalTitleKeys = new Map<string, BaseTextKey>(
|
|||
'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',
|
||||
|
|
|
|||
|
|
@ -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}`;
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user