fix(core): Run one node of a tool through the node that runs it (no-changelog) (#38759)

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaakko Husso 2026-09-24 12:26:45 +00:00 committed by GitHub
parent 25eeb74fe9
commit 482ffb5306
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 2755 additions and 61 deletions

View File

@ -641,6 +641,7 @@ evaluation mode, so any other mode would drop the workflow's pins.
| `nodeName` | string | yes | — | Node to run |
| `reuseExecutionId` | string | no | — | Replay this past execution's data for the nodes above the target |
| `mockInput` | object[] | no | — | Items to feed the target, skipping every node above it |
| `toolArguments` | object \| string | no | — | Arguments for a tool target — what an agent would fill from `$fromAI` |
| `versionId` | string | no | current draft | Run a past version's graph |
| `timeout` | number | no | 300000 | Max wait time in ms (max 600000) |
@ -669,6 +670,96 @@ placeholder on an upstream IF or Switch picks a branch that real data may pick
differently, which is why a mocked step is never evidence that the workflow
works.
The action refuses a run whose input would not keep the nodes above the target
out of it. It applies the rules of `findStartNodes`, so the engine re-runs:
- A node with neither run data nor pin data.
- A node whose saved run failed, even a pinned one. The engine retries it.
- A Loop Over Items node whose last run left the `done` output empty. The
engine restarts the loop. A loop edge that runs through the target does not
count, because `findSubgraph` drops it, and then the second output decides.
`mockInput` can hit only the last rule: the placeholder on a loop leaves
`done` empty when the target hangs off the loop body. Use `reuseExecutionId`
with an execution where the loop finished.
**Sub-node targets**: a tool has no main input, and the engine never runs one on
its own. It replaces the node that owns the tool (the Agent) with a virtual Tool
Executor that inherits that node's main parents, then runs the tool from there.
So a step on a tool is planned against the Agent: `mockInput` feeds the Agent's
input, and `reuseExecutionId` replays the Agent's ancestors. A chain run on a
tool runs every node above the Agent, so supply `reuseExecutionId` or
`mockInput` when one of those nodes writes.
The engine runs the tool through a virtual node the workflow does not contain,
`PartialExecutionToolExecutor`. The result never carries that name: the run is
reported under the node the caller named, in the output data, the executed
names, the last node, and a node error. The tool's own record is the one kept,
because the executor re-serializes the result as a single string.
`ranThroughNodeNames` names the nodes that can run the tool, not the one that
ran it. A tool that hangs off several agents lists them all: the engine picks
one and reports no choice, so any single name here would be a guess. The plan
covers every candidate's ancestry, so the run is right whichever one the engine
takes.
The action refuses a tool when one of its agents runs above another. If the
engine picks the lower agent, it drops the data the plan gave the upper agent
and the nodes between them, and runs those nodes for real. The error tells the
caller to run the upper agent instead and read the tool with
`get-node-output`. Agents on parallel branches are not affected.
`toolArguments` supplies what the agent would normally decide — the values
behind the tool's `$fromAI` calls, keyed by argument name, or a bare string for
a tool that takes one free-text input (Wikipedia, Code Tool, a vector store used
as a tool). It is **required** when the node declares `$fromAI` arguments: the
action refuses the run rather than execute the tool on empty arguments and
report a failure that says nothing about the user's problem.
The agent request names no tool. The Tool Executor looks the arguments up by the
tool's *runtime* name, which is `nodeNameToToolName(node)` on current versions
but comes from a parameter on older ones (`name` on Code Tool <= 1.1, Vector
Store Tool <= 1, Workflow Tool <= 2.1; `toolName` on a retrieve-as-tool vector
store < 1.3) and is hardcoded on Think 1. An empty name makes the Tool Executor
run the only tool connected to it, so a name this cannot know never stops the
run; the arguments are keyed under every name the tool can have so the lookup
finds them. Think 1's hardcoded `thinking_tool` is the one name this cannot key,
and there it costs the arguments, not the run.
A node that holds several tools is refused: the Tool Executor runs the member
whose name matches the request, that name is `buildMcpToolName` of the node
name and the server's tool name, and a miss reports success with no result at
all. Run the owning Agent instead and read the node's output from that
execution. The check is on the node type, because a node name is the user's to
change:
- `@n8n/n8n-nodes-langchain.mcpClientTool` — "MCP Client Tool" on the canvas.
- `@n8n/mcp-registry.<slug>` — a server the MCP registry added, one node type
for each server. All of them run on one hidden class
(`mcpRegistryClientTool`), and that class name never appears as a node type,
so the match is on the `@n8n/mcp-registry` package — the same test
`agents-tools.service.ts` makes.
Every **other** sub-node kind — a model, memory, embeddings — is refused: n8n
runs those only as part of the node that owns them, so the action points the
caller at that node instead of starting a run that cannot work. Their output is
still readable afterwards: a sub-node records each call under its own connection
type, and `action="get-node-output"` reads that when a node has no `main`
output.
**Refusals**: the action fails, and starts nothing, rather than run something
whose result would mislead:
- a `reuseExecutionId` whose execution holds no data for any node above the
target — falling back to a chain run would execute those nodes for real, which
is what asking for replayed input rules out;
- a tool that declares `$fromAI` arguments with no `toolArguments`;
- a node that holds several tools;
- a sub-node that is not a tool;
- a tool no node is connected to run — the engine has no node to stand in for,
so the run would start and then die;
- `toolArguments` on a node in the main graph.
**Pin data**: the target's own pin, and any pin on a node whose output the
mocked mode replaced, come off this run's copy — a pinned node never
executes, so leaving them on would make the step replay stale output. The saved
@ -719,12 +810,27 @@ Get the output data of a specific node from an execution.
| `startIndex` | number | no | First item index to return. Defaults to `0` |
| `maxItems` | number | no | Maximum items to return. Defaults to `10`; maximum `50` |
**Returns**: `{ nodeName, outputs: [{ index, name?, totalItems, items }], totalItems, returned: { from, to } }`.
**Returns**: `{ nodeName, outputs: [{ index, name?, totalItems, items }], totalItems, returned: { from, to }, totalRuns? }`.
One `outputs` entry per node output, in output order; a Filter reports `Kept` and
`Discarded` separately. `name` follows the node's output pane labels, including
renamed Switch outputs and `Success` / `Error` for nodes that route errors to an
extra output. `totalItems` and `returned` count across all outputs.
A node records one run for each time it ran, and `totalRuns` reports how many
when there was more than one. Which runs are read depends on the node:
- A node in the **main graph** reports its **last** run. A node inside a loop
has one run for each iteration, so this is the run the caller usually means.
- A **sub-node** — a model, a memory, a tool — reports **every** run, in call
order, because one run is one call its owner made. That is what makes this
action the answer to a step run the tool refuses. An item's label names the
call it came from, `node:Search Tickets[call 2][0][0]`, and `startIndex` and
`maxItems` page across the calls as one sequence.
A run that failed records no items, so `totalRuns` can count more runs than the
items account for. Read the error of that run from `action="get"`, which
reports it under `nodeErrors`.
### `executions(action="get-resolved-node-parameters")`
Replay expression resolution for a node's parameters against a past execution.

View File

@ -86,6 +86,10 @@ first move whenever the user is debugging a **read** node that already failed a
real execution: the node ran for real once already, and a mock-only check is
what sends the user back for a second session.
The action refuses a replay that would re-run a node above the target: an
unpinned one the execution never reached, one that failed there, or a loop
that did not finish. Pick another execution, or tell the user which node would run again.
### Decide whether the node is safe to run first
A step run is a real run. The node uses the user's real credentials and reaches
@ -142,6 +146,55 @@ Mocked input does **not** make a write node safe. The node still runs for real
against the user's systems; only its input is invented, which makes the effect
less predictable, not more.
### Running a tool
A tool never runs on its own. n8n runs it through the node that owns it —
usually the Agent. A step run on a tool therefore behaves like a step run on
that Agent:
- `mockInput` feeds the **Agent's** input, not the tool's arguments.
- `reuseExecutionId` replays the nodes above the **Agent**.
- A chain run (neither option) runs every node above the Agent for real. Give
one of the two options when a node up there writes.
- `ranThroughNodeNames` lists **every** node that can run the tool, not the one
that ran it. A tool on several agents lists them all: the engine picks one,
and it does not report which. Do not name a single Agent to the user when
this field holds more than one.
- A tool on two Agents where one runs above the other is refused. Run the
upper Agent, then read the tool with `executions(action="get-node-output")`.
The tool's own arguments come from `toolArguments` — the values the agent would
normally decide:
```
executions(action="run-step", workflowId, nodeName="Search Tickets Tool",
reuseExecutionId=<the failed execution>,
toolArguments={"query": "login fails", "status": "open"})
```
The example targets a **read** tool on purpose. The write rule above holds here
too, and a tool hides the write behind a friendly name: a step run on a "Create
Ticket" tool creates the ticket again, and `toolArguments` does not change that.
Only run one when the user has accepted a second write.
Use the argument names from the node's `$fromAI` calls, which
`workflows(action="get-as-code")` shows. Pass a plain string instead for a tool
that takes one free-text input (Wikipedia, Code Tool, a vector store used as a
tool).
A tool that declares `$fromAI` arguments is refused without them: it would
otherwise fail for a reason that has nothing to do with the user's problem, and
you would report that as the defect. A node that holds several tools is refused
outright, because nothing here can name one of its tools the way the agent
does: the "MCP Client Tool" node, and every node the MCP registry added, whose
type is `@n8n/mcp-registry.<server slug>`. Run the Agent for those, and read
the node's output from that execution.
A sub-node that is not a tool — a model, memory, embeddings — cannot be run this
way at all. Run the Agent, and read the sub-node with
`executions(action="get-node-output")` on **that** execution: n8n records every
call a sub-node made while the Agent ran.
## Successful execution with wrong or empty value
When `debug` doesn't apply because nothing errored, call

View File

@ -995,6 +995,7 @@ describe('executions tool', () => {
...stepInput,
reuseExecutionId: 'exec-9',
mockInput: [{ text: 'hi' }],
toolArguments: { title: 'Login fails' },
versionId: 'v-2',
timeout: 30_000,
},
@ -1007,12 +1008,31 @@ describe('executions tool', () => {
expect.objectContaining({
reuseExecutionId: 'exec-9',
mockInput: [{ text: 'hi' }],
toolArguments: { title: 'Login fails' },
versionId: 'v-2',
timeout: 30_000,
}),
);
});
it('accepts a bare string as the tool arguments', async () => {
const context = createMockContext({ permissions: {} });
const tool = createExecutionsTool(context);
await executeTool(
tool,
{ ...stepInput, toolArguments: 'Napoleon' },
createAgentCtx({ resumeData: { approved: true } }) as never,
);
// A tool with one free-text input takes the query directly, not wrapped.
expect(context.executionService.runStep).toHaveBeenCalledWith(
'wf-1',
'Send Slack message',
expect.objectContaining({ toolArguments: 'Napoleon' }),
);
});
it('persists a per-node grant when the user picks "always allow"', async () => {
const grantSessionToolApproval = vi.fn();
const context = createMockContext({ permissions: {}, grantSessionToolApproval });

View File

@ -113,7 +113,9 @@ const runStepAction = z.object({
'that writes (create/update/delete/send/append, non-GET HTTP) — that ' +
'performs the effect again. Read the failed execution with action="debug" ' +
'and action="get-resolved-node-parameters" instead. When unsure, treat the ' +
'node as a write.',
'node as a write. A tool runs through the node that owns it (the Agent), so ' +
"mockInput and reuseExecutionId feed that node and the tool's own " +
'arguments come from toolArguments. No other sub-node kind can run.',
),
workflowId: z.string().describe('Workflow ID'),
nodeName: z.string().describe('Name of the node, as named in the workflow the action targets'),
@ -138,6 +140,16 @@ const runStepAction = z.object({
'question. This does not make a write node safe: the node still runs for ' +
'real, only its input is invented.',
),
toolArguments: z
.union([z.string(), z.record(z.unknown())])
.optional()
.describe(
'Only for a tool node: the arguments an agent would pass it. Keys are the ' +
"tool's $fromAI argument names. Pass a plain string for a tool that takes " +
'one free-text input (Wikipedia, Code Tool, a vector store used as a ' +
'tool). Required when the node declares $fromAI arguments, because the ' +
'tool runs with empty arguments otherwise.',
),
versionId: z
.string()
.optional()
@ -519,6 +531,7 @@ async function handleRunStep(
return await context.executionService.runStep(input.workflowId, input.nodeName, {
reuseExecutionId: input.reuseExecutionId,
mockInput: input.mockInput,
toolArguments: input.toolArguments,
versionId: input.versionId,
timeout: input.timeout,
abortSignal,

View File

@ -215,6 +215,16 @@ export interface StepExecutionResult extends ExecutionResult {
replayedNodeNames?: string[];
/** Execution the replayed run data came from. */
reusedFromExecutionId?: string;
/**
* Nodes that can run the target, when it is a sub-node (a tool, a model, a
* memory). The engine runs a sub-node from the node that owns it, so the
* input came from that node's input, and `mockInput` fed that node.
*
* A tool wired to several agents lists them all: n8n runs the step through
* one of them, and which one is the engine's choice, not this caller's.
* A step run refuses a tool when one of its agents runs above another.
*/
ranThroughNodeNames?: string[];
}
export interface NodeOutputBranch {
@ -230,15 +240,24 @@ export interface NodeOutputBranch {
export interface NodeOutputResult {
nodeName: string;
/**
* One entry per output of the node's last run, in output order. Multi-output
* nodes (Filter, IF, Switch) keep each output separate, so their items are
* never merged into one list.
* One entry per output, in output order. Multi-output nodes (Filter, IF,
* Switch) keep each output separate, so their items are never merged into
* one list.
*
* A node in the main graph reports its last run. A sub-node (a model, a
* memory, a tool) reports every run, because one run is one call its owner
* made, and an item's label names the call it came from.
*/
outputs: NodeOutputBranch[];
/** Item count across all outputs. */
totalItems: number;
/** Page position over the items of all outputs, first output first. */
returned: { from: number; to: number };
/**
* Runs the node recorded, when it recorded more than one. A sub-node's runs
* are all read; a node in the main graph reports the last of them.
*/
totalRuns?: number;
}
export interface ResolvedExpressionFailure {
@ -634,6 +653,9 @@ export interface InstanceAiExecutionService {
* - neither option — run every ancestor that has no data yet, then the target.
* - `mockInput` — supply the input and skip the ancestors entirely.
*
* A sub-node has no input of its own, so these options apply to the node that
* runs it. A tool's own arguments come from `toolArguments` instead.
*
* The first two say something about the workflow, because the input is data
* the workflow really produced. `mockInput` says something about the node
* alone, which is what you want when isolating it — but a caller must not
@ -653,6 +675,15 @@ export interface InstanceAiExecutionService {
* Applied to each of the target's direct inputs.
*/
mockInput?: Array<Record<string, unknown>>;
/**
* Arguments for a tool target — the values an agent would fill from its
* `$fromAI` calls. A string for a tool that takes one free-text input.
*
* Required when the tool declares `$fromAI` arguments: the
* implementation rejects the call rather than run the tool on empty ones.
* A tool with no such arguments needs nothing here.
*/
toolArguments?: Record<string, unknown> | string;
/** Run a past version's graph instead of the current draft. */
versionId?: string;
timeout?: number;

View File

@ -1,19 +1,33 @@
import type { IConnections, INode, IRunData, ITaskData } from 'n8n-workflow';
import { TOOL_EXECUTOR_NODE_NAME } from '@n8n/constants';
import {
cleanRunData,
DirectedGraph,
findStartNodes,
findSubgraph,
handleCycles,
rewireGraph,
} from 'n8n-core';
import type { IConnections, INode, IRunData, ITaskData, NodeConnectionType } from 'n8n-workflow';
import { NodeConnectionTypes } from 'n8n-workflow';
import {
buildMockedStepRunData,
buildToolAgentRequest,
collectAncestorNames,
declaredToolArguments,
findRootsAboveOtherRoots,
isToolkitNode,
pinDataForStepRun,
planStepRun,
resolveStepRunRoots,
toExecutionItems,
} from '../instance-ai-step-run';
function node(name: string, options: { disabled?: boolean } = {}): INode {
function node(name: string, options: { disabled?: boolean; type?: string } = {}): INode {
return {
id: name,
name,
type: 'n8n-nodes-base.noOp',
type: options.type ?? 'n8n-nodes-base.noOp',
typeVersion: 1,
position: [0, 0],
parameters: {},
@ -42,10 +56,33 @@ function connect(...edges: Array<[string, string]>): IConnections {
return connections;
}
/** Adds a sub-node connection: `from` feeds `to` on a non-main type. */
function connectSubNode(
connections: IConnections,
from: string,
to: string,
type: NodeConnectionType = NodeConnectionTypes.AiTool,
): IConnections {
connections[from] ??= {};
connections[from][type] = [[{ node: to, type, index: 0 }]];
return connections;
}
function itemsOn(runData: IRunData, nodeName: string, outputIndex = 0) {
return runData[nodeName]?.[0]?.data?.[NodeConnectionTypes.Main]?.[outputIndex];
}
/** Task data with one entry for each output, for a branching node. */
function taskDataOnOutputs(outputs: Array<Array<{ json: Record<string, never> }>>): ITaskData {
return {
startTime: 0,
executionTime: 0,
executionIndex: 0,
source: [],
data: { [NodeConnectionTypes.Main]: outputs },
};
}
function taskData(items: Array<{ json: Record<string, never> }>): ITaskData {
return {
startTime: 0,
@ -165,6 +202,13 @@ describe('buildMockedStepRunData', () => {
});
describe('collectAncestorNames', () => {
it('does not count the target as its own ancestor', () => {
const selfLooping = [node('Target')];
const connections = connect(['Target', 'Target']);
expect(collectAncestorNames(selfLooping, connections, 'Target')).toEqual([]);
});
it('returns every node that can reach the target, and nothing downstream', () => {
const nodes = [node('Trigger'), node('A'), node('Target'), node('After')];
const connections = connect(['Trigger', 'A'], ['A', 'Target'], ['Target', 'After']);
@ -229,14 +273,335 @@ describe('planStepRun', () => {
expect(plan.mockedNodeNames).toEqual([]);
});
it('runs the chain when the prior run never reached an ancestor', () => {
// The prior run only covers a node on an unrelated branch.
it('refuses the run when the prior run never reached an ancestor', () => {
// The prior run only covers a node on an unrelated branch. Falling back to
// a chain run would execute Trigger and Fetch for real, which is what the
// caller asked to avoid.
const priorRunData: IRunData = { Elsewhere: [taskData([{ json: {} }])] };
const plan = planStepRun({ nodes, connections, targetName: 'Target', priorRunData });
expect(plan.inputMode).toBe('chain');
expect(plan.runData).toBeUndefined();
expect(plan.unhonoredInput?.requested).toBe('reused-execution');
expect(plan.unhonoredInput?.upstreamNodeNames.sort()).toEqual(['Fetch', 'Trigger']);
});
// Run data for *some* ancestor is not enough. The engine walks down from the
// trigger and re-runs the first node that has none, plus everything after
// it, so a gap anywhere on the path is a node that runs for real.
describe('partial coverage of the path', () => {
it('refuses a replay that stops above the target', () => {
const priorRunData: IRunData = { Trigger: [taskData([{ json: {} }])] };
const plan = planStepRun({ nodes, connections, targetName: 'Target', priorRunData });
expect(plan.inputMode).toBe('chain');
expect(plan.unhonoredInput?.requested).toBe('reused-execution');
// Only Fetch is missing. Trigger is covered, so it stays out of the message.
expect(plan.unhonoredInput?.upstreamNodeNames).toEqual(['Fetch']);
});
it('refuses a replay that misses the trigger, which restarts the whole chain', () => {
const priorRunData: IRunData = { Fetch: [taskData([{ json: {} }])] };
const plan = planStepRun({ nodes, connections, targetName: 'Target', priorRunData });
expect(plan.inputMode).toBe('chain');
expect(plan.unhonoredInput?.upstreamNodeNames.sort()).toEqual(['Fetch', 'Trigger']);
});
it('accepts a replay that skips a branch the earlier run never took', () => {
// Kept and Discarded both reach the target. The earlier run sent every
// item down output 0, so the engine never walks into Discarded.
const branching = [
node('Trigger'),
node('IF'),
node('Kept'),
node('Discarded'),
node('Target'),
];
const branchConnections = connect(
['Trigger', 'IF'],
['IF:0', 'Kept'],
['IF:1', 'Discarded'],
['Kept', 'Target'],
['Discarded', 'Target'],
);
const priorRunData: IRunData = {
Trigger: [taskData([{ json: {} }])],
IF: [taskDataOnOutputs([[{ json: {} }], []])],
Kept: [taskData([{ json: {} }])],
};
const plan = planStepRun({
nodes: branching,
connections: branchConnections,
targetName: 'Target',
priorRunData,
});
expect(plan.inputMode).toBe('reused-execution');
});
it('accepts a replay that skips a trigger which never fired', () => {
// The engine prefers the trigger that has run data, so the other one is
// not part of the run.
const twoTriggers = [node('Trigger'), node('Schedule'), node('Fetch'), node('Target')];
const triggerConnections = connect(
['Trigger', 'Fetch'],
['Schedule', 'Fetch'],
['Fetch', 'Target'],
);
const priorRunData: IRunData = {
Trigger: [taskData([{ json: {} }])],
Fetch: [taskData([{ json: {} }])],
};
const plan = planStepRun({
nodes: twoTriggers,
connections: triggerConnections,
targetName: 'Target',
priorRunData,
});
expect(plan.inputMode).toBe('reused-execution');
});
it('counts a pinned node as covered, because it never executes', () => {
const priorRunData: IRunData = { Fetch: [taskData([{ json: {} }])] };
const plan = planStepRun({
nodes,
connections,
targetName: 'Target',
priorRunData,
pinnedNodeNames: ['Trigger'],
});
expect(plan.inputMode).toBe('reused-execution');
});
/**
* Where the engine starts a run to `Target`: the steps `runPartialWorkflow2`
* takes, from the subgraph to the cycle handling.
*/
function engineStartNodes(
graphNodes: INode[],
graphConnections: IConnections,
runData: IRunData,
) {
const graph = DirectedGraph.fromNodesAndConnections(graphNodes, graphConnections);
const byName = graph.getNodes();
const trigger = byName.get('Trigger')!;
const destination = byName.get('Target')!;
const subgraph = findSubgraph({ graph, destination, trigger });
const cleaned = cleanRunData(runData, subgraph, new Set([destination]));
const startNodes = findStartNodes({
graph: subgraph,
trigger,
destination,
runData: cleaned,
pinData: {},
});
return [...handleCycles(subgraph, startNodes, trigger)].map((startNode) => startNode.name);
}
const failedTask = (): ITaskData => ({
...taskData([{ json: {} }]),
error: new Error('timed out') as ITaskData['error'],
});
it('refuses a replay whose node failed, because the engine retries it', () => {
const priorRunData: IRunData = {
Trigger: [taskData([{ json: {} }])],
Fetch: [failedTask()],
};
const plan = planStepRun({ nodes, connections, targetName: 'Target', priorRunData });
expect(plan.inputMode).toBe('chain');
expect(plan.unhonoredInput?.upstreamNodeNames).toEqual(['Fetch']);
expect(engineStartNodes(nodes, connections, priorRunData)).toEqual(['Fetch']);
});
it('refuses a replay whose pinned node failed, because the error outranks the pin', () => {
const priorRunData: IRunData = {
Trigger: [taskData([{ json: {} }])],
Fetch: [failedTask()],
};
const plan = planStepRun({
nodes,
connections,
targetName: 'Target',
priorRunData,
pinnedNodeNames: ['Fetch'],
});
expect(plan.unhonoredInput?.upstreamNodeNames).toEqual(['Fetch']);
});
describe('a Loop Over Items node above the target', () => {
const loopNodes = [
node('Trigger'),
node('Loop', { type: 'n8n-nodes-base.splitInBatches' }),
node('Body'),
node('Target'),
];
const loopRunning = taskDataOnOutputs([[], [{ json: {} }]]);
const loopDone = taskDataOnOutputs([[{ json: {} }], []]);
const runOf = (...loopRuns: ITaskData[]): IRunData => ({
Trigger: [taskData([{ json: {} }])],
Loop: loopRuns,
Body: [taskData([{ json: {} }])],
});
const plan = (connections: IConnections, input: Partial<Parameters<typeof planStepRun>[0]>) =>
planStepRun({ nodes: loopNodes, connections, targetName: 'Target', ...input });
const mockItems = toExecutionItems([{ id: 1 }]);
it('refuses a replay that starts at an unfinished loop with no input', () => {
// Loop "loop" -> Target, and nothing feeds the Loop node. With no
// trigger, the engine starts its walk at the Loop node and still
// applies the loop rule to it.
const sourceLoop = connect(['Loop:1', 'Target']);
const priorRunData: IRunData = { Loop: [loopDone] };
expect(plan(sourceLoop, { priorRunData }).unhonoredInput?.upstreamNodeNames).toEqual([
'Loop',
]);
const graph = DirectedGraph.fromNodesAndConnections(loopNodes, sourceLoop);
const loop = graph.getNodes().get('Loop')!;
const target = graph.getNodes().get('Target')!;
const subgraph = findSubgraph({ graph, destination: target, trigger: loop });
const startNodes = findStartNodes({
graph: subgraph,
trigger: loop,
destination: target,
runData: priorRunData,
pinData: {},
});
expect([...startNodes].map((startNode) => startNode.name)).toEqual(['Loop']);
});
describe('with the target after the done output', () => {
// Trigger -> Loop; Loop "loop" -> Body -> Loop; Loop "done" -> Target.
const afterDone = connect(
['Trigger', 'Loop'],
['Loop:1', 'Body'],
['Body', 'Loop'],
['Loop:0', 'Target'],
);
it('refuses a replay of a loop that stopped before its done output', () => {
const priorRunData = runOf(loopRunning);
expect(plan(afterDone, { priorRunData }).unhonoredInput?.upstreamNodeNames).toEqual([
'Loop',
]);
expect(engineStartNodes(loopNodes, afterDone, priorRunData)).toEqual(['Loop']);
});
it('accepts a replay of a loop that finished', () => {
const priorRunData = runOf(loopRunning, loopDone);
expect(plan(afterDone, { priorRunData }).inputMode).toBe('reused-execution');
expect(engineStartNodes(loopNodes, afterDone, priorRunData)).toEqual(['Target']);
});
it('mocks the path, because the mock fills the done output', () => {
const mocked = plan(afterDone, { mockItems });
expect(mocked.inputMode).toBe('mocked');
expect(engineStartNodes(loopNodes, afterDone, mocked.runData!)).toEqual(['Target']);
});
});
describe('with the target on the cycle', () => {
// Trigger -> Loop; Loop "loop" -> Body -> Target -> Loop.
// `findSubgraph` drops a cycle through the destination, so the engine
// treats the Loop node as a plain node.
const onCycle = connect(
['Trigger', 'Loop'],
['Loop:1', 'Body'],
['Body', 'Target'],
['Target', 'Loop'],
);
it('refuses a replay of a finished loop, which the engine restarts', () => {
const priorRunData = runOf(loopRunning, loopDone);
expect(plan(onCycle, { priorRunData }).unhonoredInput?.upstreamNodeNames).toEqual([
'Loop',
]);
expect(engineStartNodes(loopNodes, onCycle, priorRunData)).toEqual(['Loop']);
});
it('mocks the path', () => {
const mocked = plan(onCycle, { mockItems });
expect(mocked.inputMode).toBe('mocked');
expect(engineStartNodes(loopNodes, onCycle, mocked.runData!)).toEqual(['Target']);
});
});
describe('with the target off the loop body', () => {
// Trigger -> Loop; Loop "loop" -> Body -> Loop, and Body -> Target.
const offBody = connect(
['Trigger', 'Loop'],
['Loop:1', 'Body'],
['Body', 'Loop'],
['Body', 'Target'],
);
it('refuses mocked input, which leaves the done output empty', () => {
const refused = plan(offBody, { mockItems });
expect(refused.inputMode).toBe('chain');
expect(refused.unhonoredInput).toEqual({
requested: 'mocked',
upstreamNodeNames: ['Loop'],
});
// Fed to the engine, the mock restarts the loop, and "Body" runs.
const { runData } = buildMockedStepRunData({
nodes: loopNodes,
connections: offBody,
targetName: 'Target',
mockItems,
});
expect(engineStartNodes(loopNodes, offBody, runData)).toEqual(['Loop']);
});
it('accepts a replay of a loop that finished', () => {
const priorRunData = runOf(loopRunning, loopDone);
expect(plan(offBody, { priorRunData }).inputMode).toBe('reused-execution');
expect(engineStartNodes(loopNodes, offBody, priorRunData)).toEqual(['Target']);
});
});
});
});
it('refuses the run when the reused execution stored no run data at all', () => {
const plan = planStepRun({ nodes, connections, targetName: 'Target', priorRunData: {} });
expect(plan.inputMode).toBe('chain');
expect(plan.unhonoredInput?.requested).toBe('reused-execution');
});
it('runs the chain without complaint when there is nothing above the target', () => {
const plan = planStepRun({
nodes: [node('Alone')],
connections: {},
targetName: 'Alone',
priorRunData: { Elsewhere: [taskData([{ json: {} }])] },
});
expect(plan.inputMode).toBe('chain');
expect(plan.unhonoredInput).toBeUndefined();
});
it('prefers mocked input over prior run data', () => {
@ -254,6 +619,43 @@ describe('planStepRun', () => {
});
});
// Mocking walks the same edges from the same roots as the ancestor walk, so a
// target with a node above it always has something to mock, and a mocked
// request never has to be refused. A self loop broke that: the mock skipped the
// target, the ancestor walk counted it, and the plan reported the target as a
// node above itself.
describe('a mocked request without a loop never goes unhonoured', () => {
const shapes: Array<[string, INode[], IConnections]> = [
['a chain above the target', [node('Trigger'), node('Target')], connect(['Trigger', 'Target'])],
['nothing above the target', [node('Target')], {}],
['a self loop', [node('Target')], connect(['Target', 'Target'])],
[
'a two-node cycle',
[node('Target'), node('Other')],
connect(['Target', 'Other'], ['Other', 'Target']),
],
[
'a self loop below a trigger',
[node('Trigger'), node('Target')],
connect(['Trigger', 'Target'], ['Target', 'Target']),
],
];
it.each(shapes)('%s', (_label, nodes, connections) => {
const plan = planStepRun({
nodes,
connections,
targetName: 'Target',
mockItems: toExecutionItems([{ id: 1 }]),
});
expect(plan.unhonoredInput).toBeUndefined();
// Either the plan mocked the path, or there was no path to mock.
const ancestors = collectAncestorNames(nodes, connections, 'Target');
expect(plan.mockedNodeNames.length > 0 || ancestors.length === 0).toBe(true);
});
});
describe('pinDataForStepRun', () => {
const pins = {
Trigger: [{ json: { a: 1 } }],
@ -290,3 +692,447 @@ describe('pinDataForStepRun', () => {
).toBeUndefined();
});
});
describe('sub-node targets', () => {
// Trigger -> Create Ticket -> Agent, with a tool hanging off the Agent.
const nodes = [node('Trigger'), node('Create Ticket'), node('Agent'), node('Calculator')];
const connections = connectSubNode(
connect(['Trigger', 'Create Ticket'], ['Create Ticket', 'Agent']),
'Calculator',
'Agent',
);
it('resolves a tool to the node that runs it', () => {
expect(resolveStepRunRoots(nodes, connections, 'Calculator')).toEqual(['Agent']);
});
it('resolves a node in the main graph to itself', () => {
expect(resolveStepRunRoots(nodes, connections, 'Agent')).toEqual(['Agent']);
expect(resolveStepRunRoots(nodes, connections, 'Trigger')).toEqual(['Trigger']);
});
it('follows a chain of sub-nodes up to the root', () => {
// Embeddings -> Vector Store -> Agent, all through non-main connections.
const chained = [node('Trigger'), node('Agent'), node('Vector Store'), node('Embeddings')];
const chainedConnections = connectSubNode(
connectSubNode(connect(['Trigger', 'Agent']), 'Vector Store', 'Agent'),
'Embeddings',
'Vector Store',
NodeConnectionTypes.AiEmbedding,
);
expect(resolveStepRunRoots(chained, chainedConnections, 'Embeddings')).toEqual(['Agent']);
});
it('returns every root a shared tool hangs off', () => {
const shared = [node('Trigger'), node('Agent A'), node('Agent B'), node('Calculator')];
const sharedConnections = connect(['Trigger', 'Agent A'], ['Agent A', 'Agent B']);
sharedConnections.Calculator = {
[NodeConnectionTypes.AiTool]: [
[
{ node: 'Agent A', type: NodeConnectionTypes.AiTool, index: 0 },
{ node: 'Agent B', type: NodeConnectionTypes.AiTool, index: 0 },
],
],
};
expect(resolveStepRunRoots(shared, sharedConnections, 'Calculator').sort()).toEqual([
'Agent A',
'Agent B',
]);
});
it('mocks the path above the Agent when the target is its tool', () => {
const plan = planStepRun({
nodes,
connections,
targetName: 'Calculator',
mockItems: toExecutionItems([{ id: 1 }]),
});
// Without this the plan finds no main parent of the tool, degrades to a
// chain run, and "Create Ticket" writes again.
expect(plan.inputMode).toBe('mocked');
expect(plan.mockedNodeNames.sort()).toEqual(['Create Ticket', 'Trigger']);
expect(plan.rootNodeNames).toEqual(['Agent']);
expect(itemsOn(plan.runData!, 'Create Ticket')).toEqual([{ json: { id: 1 } }]);
// The Agent is replaced by the engine's Tool Executor, so it gets no run
// data of its own.
expect(plan.runData?.Agent).toBeUndefined();
expect(plan.dirtyNodeNames).toEqual(['Calculator']);
});
it('replays the Agent path when the target is its tool', () => {
const priorRunData: IRunData = {
Trigger: [taskData([{ json: {} }])],
'Create Ticket': [taskData([{ json: {} }])],
Agent: [taskData([{ json: {} }])],
};
const plan = planStepRun({ nodes, connections, targetName: 'Calculator', priorRunData });
expect(plan.inputMode).toBe('reused-execution');
expect(plan.reusedNodeNames.sort()).toEqual(['Create Ticket', 'Trigger']);
expect(plan.rootNodeNames).toEqual(['Agent']);
expect(plan.dirtyNodeNames).toEqual(['Calculator']);
});
describe('findRootsAboveOtherRoots', () => {
const stacked = [node('Trigger'), node('Agent A'), node('POST'), node('Agent B')];
it('names the root that runs above another root', () => {
const stackedConnections = connect(
['Trigger', 'Agent A'],
['Agent A', 'POST'],
['POST', 'Agent B'],
);
expect(findRootsAboveOtherRoots(stacked, stackedConnections, ['Agent B', 'Agent A'])).toEqual(
['Agent A'],
);
});
it('names nothing for roots on parallel branches', () => {
const parallelConnections = connect(
['Trigger', 'Agent A'],
['Trigger', 'POST'],
['POST', 'Agent B'],
);
expect(
findRootsAboveOtherRoots(stacked, parallelConnections, ['Agent A', 'Agent B']),
).toEqual([]);
});
it('names every root on a loop, where none is the topmost', () => {
const loopConnections = connect(
['Trigger', 'Agent A'],
['Agent A', 'Agent B'],
['Agent B', 'Agent A'],
);
expect(findRootsAboveOtherRoots(stacked, loopConnections, ['Agent A', 'Agent B'])).toEqual([
'Agent A',
'Agent B',
]);
});
it('names nothing for a single root', () => {
expect(findRootsAboveOtherRoots(nodes, connections, ['Agent'])).toEqual([]);
});
});
it('runs the chain for a tool with no node to run it', () => {
const orphan = [node('Trigger'), node('Calculator')];
const plan = planStepRun({
nodes: orphan,
connections: {},
targetName: 'Calculator',
mockItems: toExecutionItems([{ id: 1 }]),
});
// Nothing above it to keep from running, so the plan has nothing to say.
// The run is refused a step later, where the node types are known and the
// message can name the missing Agent — see the adapter tests.
expect(plan.inputMode).toBe('chain');
expect(plan.unhonoredInput).toBeUndefined();
});
});
/**
* The engine's own graph rules, run on the plan. The planner exists to satisfy
* `findStartNodes`, so a unit test of the plan alone cannot prove it works.
*/
describe('the plan against the engine rules', () => {
const trigger = node('Trigger');
const createTicket = node('Create Ticket');
const agent = node('Agent');
const calculator = node('Calculator');
const nodes = [trigger, createTicket, agent, calculator];
const connections = connectSubNode(
connect(['Trigger', 'Create Ticket'], ['Create Ticket', 'Agent']),
'Calculator',
'Agent',
);
/** What `runPartialWorkflow2` does to a tool destination, then who starts. */
function startNodesFor(runData: IRunData) {
const graph = new DirectedGraph()
.addNodes(...nodes)
.addConnections(
{ from: trigger, to: createTicket },
{ from: createTicket, to: agent },
{ from: calculator, to: agent, type: NodeConnectionTypes.AiTool },
);
const rewired = rewireGraph(calculator, graph);
const destination = rewired.getNodes().get(TOOL_EXECUTOR_NODE_NAME);
expect(destination).toBeDefined();
const subgraph = findSubgraph({ graph: rewired, destination: destination!, trigger });
return [
...findStartNodes({
graph: subgraph,
trigger,
destination: destination!,
runData,
pinData: {},
}),
].map((startNode) => startNode.name);
}
it('starts at the Tool Executor, so no node above the Agent runs again', () => {
const plan = planStepRun({
nodes,
connections,
targetName: 'Calculator',
mockItems: toExecutionItems([{ id: 1 }]),
});
expect(startNodesFor(plan.runData!)).toEqual([TOOL_EXECUTOR_NODE_NAME]);
});
it('starts at the trigger without run data, which is the chain run to avoid', () => {
expect(startNodesFor({})).toEqual(['Trigger']);
});
// Proof that refusing a partly covered replay is not over-caution: fed to the
// engine, that run data starts at a real node, and "Create Ticket" writes.
it('refuses a replay whose gap would run a real node', () => {
const priorRunData: IRunData = { Trigger: [taskData([{ json: {} }])] };
const plan = planStepRun({
nodes,
connections,
targetName: 'Calculator',
priorRunData,
});
expect(plan.inputMode).toBe('chain');
expect(plan.unhonoredInput?.upstreamNodeNames).toEqual(['Create Ticket']);
expect(startNodesFor(priorRunData)).toEqual(['Create Ticket']);
});
// The Agent is rarely the last node. `rewireGraph` used to walk every
// descendant of the tool and stand in for the *farthest* one, so a node below
// the Agent supplied the Tool Executor's main parents and the Agent — plus
// everything between it and that node — ran again.
it('starts at the Tool Executor when the Agent has nodes below it', () => {
const notify = node('Notify');
const summarize = node('Summarize');
const withDownstream = [...nodes, notify, summarize];
const connectionsWithDownstream = connectSubNode(
connect(
['Trigger', 'Create Ticket'],
['Create Ticket', 'Agent'],
['Agent', 'Notify'],
['Notify', 'Summarize'],
),
'Calculator',
'Agent',
);
const plan = planStepRun({
nodes: withDownstream,
connections: connectionsWithDownstream,
targetName: 'Calculator',
mockItems: toExecutionItems([{ id: 1 }]),
});
const graph = new DirectedGraph()
.addNodes(...withDownstream)
.addConnections(
{ from: trigger, to: createTicket },
{ from: createTicket, to: agent },
{ from: agent, to: notify },
{ from: notify, to: summarize },
{ from: calculator, to: agent, type: NodeConnectionTypes.AiTool },
);
const rewired = rewireGraph(calculator, graph);
const destination = rewired.getNodes().get(TOOL_EXECUTOR_NODE_NAME);
expect(destination).toBeDefined();
const subgraph = findSubgraph({ graph: rewired, destination: destination!, trigger });
const startNodes = [
...findStartNodes({
graph: subgraph,
trigger,
destination: destination!,
runData: plan.runData!,
pinData: {},
}),
].map((startNode) => startNode.name);
expect(startNodes).toEqual([TOOL_EXECUTOR_NODE_NAME]);
// Neither node below the Agent is even part of the run.
expect([...subgraph.getNodes().keys()]).not.toContain('Notify');
expect([...subgraph.getNodes().keys()]).not.toContain('Summarize');
});
// Proof that refusing a tool on stacked Agents is not over-caution. The plan
// covers both Agents, yet the engine drops that data when it picks the
// lower one, and the upper Agent and "POST" run for real.
describe('a tool shared by an Agent and an Agent below it', () => {
const agentA = node('Agent A');
const post = node('POST');
const agentB = node('Agent B');
const tool = node('Calculator');
const stacked = [trigger, agentA, post, agentB, tool];
const stackedConnections = connect(
['Trigger', 'Agent A'],
['Agent A', 'POST'],
['POST', 'Agent B'],
);
stackedConnections.Calculator = {
[NodeConnectionTypes.AiTool]: [
[
{ node: 'Agent A', type: NodeConnectionTypes.AiTool, index: 0 },
{ node: 'Agent B', type: NodeConnectionTypes.AiTool, index: 0 },
],
],
};
/** Runs the engine's partial-run steps, with the tool's edges in `order`. */
function startNodesFor(order: INode[]) {
const plan = planStepRun({
nodes: stacked,
connections: stackedConnections,
targetName: 'Calculator',
mockItems: toExecutionItems([{ id: 1 }]),
});
const graph = new DirectedGraph()
.addNodes(...stacked)
.addConnections(
{ from: trigger, to: agentA },
{ from: agentA, to: post },
{ from: post, to: agentB },
...order.map((agent) => ({ from: tool, to: agent, type: NodeConnectionTypes.AiTool })),
);
const rewired = rewireGraph(tool, graph);
const destination = rewired.getNodes().get(TOOL_EXECUTOR_NODE_NAME)!;
const subgraph = findSubgraph({ graph: rewired, destination, trigger });
const runData = cleanRunData(
plan.runData!,
subgraph,
subgraph.getNodesByNames(['Calculator']),
);
return [
...findStartNodes({ graph: subgraph, trigger, destination, runData, pinData: {} }),
].map((startNode) => startNode.name);
}
it('runs the upper Agent again when the engine picks the lower one', () => {
expect(startNodesFor([agentB, agentA])).toEqual(['Agent A']);
});
it('starts at the Tool Executor when the engine picks the upper one', () => {
expect(startNodesFor([agentA, agentB])).toEqual([TOOL_EXECUTOR_NODE_NAME]);
});
});
});
describe('tool arguments', () => {
describe('declaredToolArguments', () => {
it('collects the $fromAI keys a tool expects the agent to fill', () => {
const target: INode = {
...node('Create Ticket'),
parameters: {
url: "={{ $fromAI('url', 'the endpoint') }}",
body: {
title: "={{ $fromAI('title') }}",
assignee: "={{ $fromAI('assignee', '', 'string') }}",
},
},
};
expect(declaredToolArguments(target).sort()).toEqual(['assignee', 'title', 'url']);
});
it('returns nothing for a tool whose parameters are all static', () => {
expect(declaredToolArguments(node('Calculator'))).toEqual([]);
});
});
describe('isToolkitNode', () => {
it('is true for the MCP Client Tool, which holds several tools', () => {
expect(
isToolkitNode(node('MCP Client', { type: '@n8n/n8n-nodes-langchain.mcpClientTool' })),
).toBe(true);
});
// A registry server is saved as `@n8n/mcp-registry.<slug>`. The runtime
// class behind every one of them is hidden and never appears as a node
// type, so matching that class name matches nothing a user can build.
it('is true for a node the MCP registry added, whatever the server slug', () => {
expect(isToolkitNode(node('Linear', { type: '@n8n/mcp-registry.linear' }))).toBe(true);
expect(isToolkitNode(node('Notion', { type: '@n8n/mcp-registry.notionMcp' }))).toBe(true);
});
it('is false for a type that only starts like the registry package', () => {
expect(isToolkitNode(node('Decoy', { type: '@n8n/mcp-registryish.thing' }))).toBe(false);
});
it('is false for a node that supplies one tool', () => {
expect(
isToolkitNode(node('Calculator', { type: '@n8n/n8n-nodes-langchain.toolCalculator' })),
).toBe(false);
});
});
describe('buildToolAgentRequest', () => {
it('names no tool, so the runtime name cannot stop the tool from running', () => {
// An empty name makes the Tool Executor run the only tool the rewired
// graph connects to it. Naming one risks a miss: older tool versions read
// the runtime name from a parameter, and Think 1 hardcodes its own.
expect(buildToolAgentRequest({ target: node('Calculator') }).tool).toEqual({ name: '' });
});
it('keys the arguments by every name the tool can have', () => {
expect(
buildToolAgentRequest({
target: node('Create Ticket'),
toolArguments: { title: 'Broken login' },
}).query,
).toEqual({
// `nodeNameToToolName` of the node, which current versions use...
Create_Ticket: { title: 'Broken login' },
// ...and the node name itself, which the lookup falls back to.
'Create Ticket': { title: 'Broken login' },
});
});
it("adds a legacy version's configured name", () => {
// Code Tool <= 1.1, Vector Store Tool <= 1 and Workflow Tool <= 2.1 read
// the tool name from `name`.
const target = node('Search Tickets');
target.parameters = { name: 'search_tickets' };
expect(buildToolAgentRequest({ target, toolArguments: { q: 'login' } }).query).toEqual({
Search_Tickets: { q: 'login' },
'Search Tickets': { q: 'login' },
search_tickets: { q: 'login' },
});
});
it("adds a retrieve-as-tool vector store's configured name", () => {
const target = node('Docs');
target.parameters = { toolName: 'company_docs' };
expect(buildToolAgentRequest({ target }).query).toEqual({ Docs: {}, company_docs: {} });
});
it('passes a bare string through for a tool with one free-text input', () => {
expect(
buildToolAgentRequest({ target: node('Wikipedia'), toolArguments: 'Napoleon' }).query,
).toEqual({ Wikipedia: 'Napoleon' });
});
it('sends an empty argument set when the caller supplies none', () => {
expect(buildToolAgentRequest({ target: node('Calculator') }).query).toEqual({
Calculator: {},
});
});
});
});

View File

@ -49,6 +49,7 @@ vi.mock('@n8n/ai-utilities', () => ({
}));
import type { PolicyCleared } from '@n8n/decorators';
import { TOOL_EXECUTOR_NODE_NAME } from '@n8n/constants';
import { Container } from '@n8n/di';
import { generateWorkflowCode, parseWorkflowCode } from '@n8n/workflow-sdk';
import { mock } from 'vitest-mock-extended';
@ -165,6 +166,7 @@ function makeExecution(
pinData?: IPinData;
error?: Partial<ExecutionError>;
workflowNodes?: WorkflowNode[];
lastNodeExecuted?: string;
} = {},
) {
const runData = overrides.runData ?? {};
@ -183,6 +185,7 @@ function makeExecution(
runData,
pinData: overrides.pinData,
error: overrides.error,
lastNodeExecuted: overrides.lastNodeExecuted,
},
} as unknown as IRunExecutionData,
};
@ -1332,6 +1335,236 @@ describe('extractNodeOutput', () => {
expect(result.returned).toEqual({ from: 0, to: 10 });
});
it("reads a sub-node's run, which is recorded under its connection type", async () => {
// A model, memory or tool never writes a `main` output. Without this the
// node was found but reported no items at all, which reads as "the model
// returned nothing".
createMockExecutionRepository(
makeExecution({
status: 'success',
runData: {
'OpenAI Chat Model': [
{
startTime: 1000,
executionTime: 500,
executionIndex: 0,
executionStatus: 'success',
source: [],
data: { ai_languageModel: [[{ json: { response: 'four' } }]] },
} as ITaskData,
],
},
}),
);
const result = await extractNodeOutput('exec-1', 'OpenAI Chat Model');
expect(result.totalItems).toBe(1);
expect(result.outputs[0].items[0]).toContain('"response": "four"');
});
// A sub-node records one run for each call its owner made, and the refusals
// on a step run send the caller here to read them. Reading the last run only
// hid every earlier call.
it('reads every call a sub-node made, in call order', async () => {
const call = (response: string, executionIndex: number) =>
({
startTime: 1000,
executionTime: 5,
executionIndex,
executionStatus: 'success',
source: [],
data: { ai_tool: [[{ json: { response } }]] },
}) as ITaskData;
createMockExecutionRepository(
makeExecution({
status: 'success',
runData: { 'Search Tickets': [call('first', 0), call('second', 1), call('third', 2)] },
}),
);
const result = await extractNodeOutput('exec-1', 'Search Tickets');
expect(result.totalItems).toBe(3);
expect(result.totalRuns).toBe(3);
expect(result.outputs[0].totalItems).toBe(3);
expect(String(result.outputs[0].items[0])).toContain('"first"');
expect(String(result.outputs[0].items[2])).toContain('"third"');
});
it('names the call an item came from when it read several', async () => {
const call = (executionIndex: number) =>
({
startTime: 1000,
executionTime: 5,
executionIndex,
executionStatus: 'success',
source: [],
data: { ai_tool: [[{ json: { n: executionIndex } }]] },
}) as ITaskData;
createMockExecutionRepository(
makeExecution({ status: 'success', runData: { Calculator: [call(0), call(1)] } }),
);
const result = await extractNodeOutput('exec-1', 'Calculator');
expect(String(result.outputs[0].items[0])).toContain('node:Calculator[call 1][0][0]');
expect(String(result.outputs[0].items[1])).toContain('node:Calculator[call 2][0][1]');
});
it('keeps a node in the main graph on its last run', async () => {
// A node inside a loop records one run for each iteration. Merging those
// would change what every existing caller gets.
const iteration = (value: string, executionIndex: number) =>
({
startTime: 1000,
executionTime: 5,
executionIndex,
executionStatus: 'success',
source: [],
data: { main: [[{ json: { value } }]] },
}) as ITaskData;
createMockExecutionRepository(
makeExecution({
status: 'success',
runData: { 'Set Node': [iteration('early', 0), iteration('late', 1)] },
}),
);
const result = await extractNodeOutput('exec-1', 'Set Node');
expect(result.totalItems).toBe(1);
expect(String(result.outputs[0].items[0])).toContain('"late"');
// The caller still learns the node ran more than once.
expect(result.totalRuns).toBe(2);
});
// A failed run records no data at all: `createTaskData` never sets it, and
// only a rewired tool gets it filled on the error path. So "no main output"
// does not mean "is a sub-node" — a node that failed on the last iteration
// of a loop looks the same, and merging its earlier iterations would report
// stale output for the run the caller is asking about.
it('keeps a failed node in the main graph on its last run', async () => {
const iteration = (value: string, executionIndex: number) =>
({
startTime: 1000,
executionTime: 5,
executionIndex,
executionStatus: 'success',
source: [],
data: { main: [[{ json: { value } }]] },
}) as ITaskData;
const failed = {
startTime: 1000,
executionTime: 5,
executionIndex: 2,
executionStatus: 'error',
source: [],
error: { message: 'boom' },
} as unknown as ITaskData;
createMockExecutionRepository(
makeExecution({
status: 'error',
runData: { 'Set Node': [iteration('early', 0), iteration('late', 1), failed] },
}),
);
const result = await extractNodeOutput('exec-1', 'Set Node');
expect(result.totalItems).toBe(0);
expect(result.outputs).toEqual([]);
// The caller still learns the node ran three times.
expect(result.totalRuns).toBe(3);
});
// A sub-node call that throws records no data either, so the last run cannot
// be the only thing consulted: the calls that did return are the answer to
// "what did this tool do before it failed".
it('still reads the earlier calls when a sub-node fails on its last', async () => {
const call = (response: string, executionIndex: number) =>
({
startTime: 1000,
executionTime: 5,
executionIndex,
executionStatus: 'success',
source: [],
data: { ai_tool: [[{ json: { response } }]] },
}) as ITaskData;
const failed = {
startTime: 1000,
executionTime: 5,
executionIndex: 2,
executionStatus: 'error',
source: [],
error: { message: 'rate limited' },
} as unknown as ITaskData;
createMockExecutionRepository(
makeExecution({
status: 'error',
runData: { 'Search Tickets': [call('first', 0), call('second', 1), failed] },
}),
);
const result = await extractNodeOutput('exec-1', 'Search Tickets');
expect(result.totalItems).toBe(2);
expect(result.totalRuns).toBe(3);
expect(String(result.outputs[0].items[0])).toContain('"first"');
expect(String(result.outputs[0].items[1])).toContain('"second"');
});
it('pages across the calls of a sub-node', async () => {
const call = (executionIndex: number) =>
({
startTime: 1000,
executionTime: 5,
executionIndex,
executionStatus: 'success',
source: [],
data: { ai_tool: [[{ json: { n: executionIndex } }]] },
}) as ITaskData;
createMockExecutionRepository(
makeExecution({
status: 'success',
runData: { Calculator: [call(0), call(1), call(2), call(3)] },
}),
);
const result = await extractNodeOutput('exec-1', 'Calculator', { startIndex: 2, maxItems: 1 });
expect(result.returned).toEqual({ from: 2, to: 3 });
expect(result.outputs[0].items).toHaveLength(1);
expect(String(result.outputs[0].items[0])).toContain('"n": 2');
});
it('prefers the main output when a node has both', async () => {
createMockExecutionRepository(
makeExecution({
status: 'success',
runData: {
'Set Node': [
{
startTime: 1000,
executionTime: 500,
executionIndex: 0,
executionStatus: 'success',
source: [],
data: {
main: [[{ json: { real: true } }]],
ai_tool: [[{ json: { ignored: true } }]],
},
} as ITaskData,
],
},
}),
);
const result = await extractNodeOutput('exec-1', 'Set Node');
expect(result.outputs[0].items[0]).toContain('"real": true');
expect(result.outputs).toHaveLength(1);
});
it('supports startIndex pagination', async () => {
const items = Array.from({ length: 25 }, (_, i) => ({ json: { id: i } }));
createMockExecutionRepository(
@ -4645,6 +4878,31 @@ describe('resolveDataTableByIdOrName', () => {
// createExecutionAdapter – run() forces save settings
// ---------------------------------------------------------------------------
/**
* Node types for the run tests. A type whose name says "tool" describes an
* `ai_tool` output, which is what `NodeHelpers.isTool` reads, so a step run on
* one takes the same branch it takes in a real instance. A node the MCP
* registry added is named for its server (`@n8n/mcp-registry.<slug>`) and says
* "tool" nowhere, yet it describes an `ai_tool` output all the same.
*/
function runNodeTypesStub(): NodeTypes {
const nodeTypes = mock<NodeTypes>();
nodeTypes.getByNameAndVersion.mockImplementation(
(type: string) =>
({
description: {
name: type,
outputs: [
/tool/i.test(type) || type.startsWith('@n8n/mcp-registry.')
? NodeConnectionTypes.AiTool
: NodeConnectionTypes.Main,
],
},
}) as never,
);
return nodeTypes;
}
function createRunAdapterForTests(
workflow: Record<string, unknown>,
options?: {
@ -4654,6 +4912,7 @@ function createRunAdapterForTests(
threadId?: string;
queueMode?: boolean;
allowSendingParameterValues?: boolean;
nodeTypes?: NodeTypes;
},
) {
const mockWorkflowFinderService = {
@ -4704,7 +4963,9 @@ function createRunAdapterForTests(
mockActiveExecutions as unknown as ConstructorParameters<typeof InstanceAiAdapterService>[10],
mockWorkflowRunner as unknown as ConstructorParameters<typeof InstanceAiAdapterService>[11],
{} as unknown as ConstructorParameters<typeof InstanceAiAdapterService>[12],
{} as unknown as ConstructorParameters<typeof InstanceAiAdapterService>[13],
(options?.nodeTypes ?? runNodeTypesStub()) as unknown as ConstructorParameters<
typeof InstanceAiAdapterService
>[13],
{ n8nFolder: '/tmp' } as unknown as ConstructorParameters<typeof InstanceAiAdapterService>[14],
{} as unknown as ConstructorParameters<typeof InstanceAiAdapterService>[15],
{} as unknown as ConstructorParameters<typeof InstanceAiAdapterService>[16],
@ -5520,6 +5781,7 @@ describe('createExecutionAdapter runStep()', () => {
type StepOptions = {
reuseExecutionId?: string;
mockInput?: Array<Record<string, unknown>>;
toolArguments?: Record<string, unknown> | string;
versionId?: string;
timeout?: number;
};
@ -5698,7 +5960,10 @@ describe('createExecutionAdapter runStep()', () => {
expect(runData.executionData.manualData.dirtyNodeNames).toEqual(['Send']);
expect(runData.executionData.manualData.source).toBe('instance_ai');
} finally {
process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS = previous;
// Assigning `undefined` stores the string "undefined", which leaves the
// variable set for every later test in this file.
if (previous === undefined) delete process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS;
else process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS = previous;
}
});
@ -5715,7 +5980,10 @@ describe('createExecutionAdapter runStep()', () => {
// `runManually` instead of replaying a prepared stack.
expect(runData.executionData.executionData).toBeUndefined();
} finally {
process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS = previous;
// Assigning `undefined` stores the string "undefined", which leaves the
// variable set for every later test in this file.
if (previous === undefined) delete process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS;
else process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS = previous;
}
});
@ -5863,6 +6131,585 @@ describe('createExecutionAdapter runStep()', () => {
}),
);
});
it('refuses mocked input that would restart a loop above the target', async () => {
// Trigger -> Loop; Loop "loop" -> Body -> Loop, and Body -> Send. The mock
// leaves the done output empty, so the engine would restart the loop.
const loopWorkflow = {
id: 'wf-1',
versionId: 'v-current',
nodes: [
{ name: 'Trigger', type: 'n8n-nodes-base.manualTrigger', typeVersion: 1, position: [0, 0] },
{ name: 'Loop', type: 'n8n-nodes-base.splitInBatches', typeVersion: 3, position: [1, 0] },
{ name: 'Body', type: 'n8n-nodes-base.httpRequest', typeVersion: 1, position: [2, 0] },
{ name: 'Send', type: 'n8n-nodes-base.noOp', typeVersion: 1, position: [3, 0] },
],
connections: {
Trigger: { main: [[{ node: 'Loop', type: 'main', index: 0 }]] },
Loop: { main: [[], [{ node: 'Body', type: 'main', index: 0 }]] },
Body: {
main: [
[
{ node: 'Loop', type: 'main', index: 0 },
{ node: 'Send', type: 'main', index: 0 },
],
],
},
},
};
const harness = createRunAdapterForTests(loopWorkflow, {
execution: makeExecution({ status: 'success' }),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
await expect(runStep('wf-1', 'Send', { mockInput: [{}] })).rejects.toThrow(
'the mock leaves a Loop Over Items node unfinished, and n8n would restart it and execute these nodes for real (Loop)',
);
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
it('refuses to run the chain when the reused execution covers nothing above the target', async () => {
const harness = createRunAdapterForTests(chainWorkflow, {
// The execution never reached the nodes above `Send`.
execution: makeExecution({
status: 'success',
runData: { 'Sibling Branch': [makeTaskData([{}])] },
}),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
// Falling back to a chain run would call `Fetch` for real, which is the
// opposite of what asking for replayed input means.
await expect(runStep('wf-1', 'Send', { reuseExecutionId: 'exec-past' })).rejects.toThrow(
'does not cover every node above "Send"',
);
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
it('refuses to run the chain when the reused execution covers only part of the path', async () => {
const harness = createRunAdapterForTests(chainWorkflow, {
// The execution reached the trigger but stopped before `Fetch`, so a
// replay would run `Fetch` for real.
execution: makeExecution({
status: 'success',
runData: { Trigger: [makeTaskData([{}])] },
}),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
await expect(runStep('wf-1', 'Send', { reuseExecutionId: 'exec-past' })).rejects.toThrow(
'(Fetch)',
);
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
describe('a tool attached to an Agent', () => {
// Trigger -> Create Ticket -> Agent, with a Calculator tool on the Agent.
const agentWorkflow = {
id: 'wf-1',
versionId: 'v-current',
nodes: [
{ name: 'Trigger', type: 'n8n-nodes-base.manualTrigger', typeVersion: 1, position: [0, 0] },
{
name: 'Create Ticket',
type: 'n8n-nodes-base.httpRequest',
typeVersion: 1,
position: [1, 0],
},
{
name: 'Agent',
type: '@n8n/n8n-nodes-langchain.agent',
typeVersion: 1,
position: [2, 0],
},
{
name: 'Calculator',
type: '@n8n/n8n-nodes-langchain.toolCalculator',
typeVersion: 1,
position: [2, 1],
},
],
connections: {
Trigger: { main: [[{ node: 'Create Ticket', type: 'main', index: 0 }]] },
'Create Ticket': { main: [[{ node: 'Agent', type: 'main', index: 0 }]] },
Calculator: { ai_tool: [[{ node: 'Agent', type: 'ai_tool', index: 0 }]] },
},
};
/** The same graph with a node that holds several tools in place of the tool. */
const toolkitWorkflow = {
...agentWorkflow,
nodes: agentWorkflow.nodes.map((node) =>
node.name === 'Calculator'
? {
...node,
name: 'MCP Client',
type: '@n8n/n8n-nodes-langchain.mcpClientTool',
parameters: {},
}
: node,
),
connections: {
Trigger: agentWorkflow.connections.Trigger,
'Create Ticket': agentWorkflow.connections['Create Ticket'],
'MCP Client': { ai_tool: [[{ node: 'Agent', type: 'ai_tool', index: 0 }]] },
},
};
it('mocks the path above the Agent instead of running it', async () => {
const { runData, result } = await runStepOn(agentWorkflow, 'Calculator', {
mockInput: [{ text: 'hello' }],
});
// The tool has no main input of its own, so the nodes to keep from
// running are the Agent's. Without this the plan degraded to a chain run
// and "Create Ticket" wrote again.
expect(runData.runData['Create Ticket'][0].data.main[0]).toEqual([
{ json: { text: 'hello' } },
]);
expect(runData.runData.Trigger[0].data.main[0]).toEqual([{ json: {} }]);
expect(runData.runData.Agent).toBeUndefined();
expect(runData.dirtyNodeNames).toEqual(['Calculator']);
expect(result.inputMode).toBe('mocked');
expect(result.ranThroughNodeNames).toEqual(['Agent']);
});
it('replays the path above the Agent instead of running it', async () => {
const priorRunData = {
Trigger: [makeTaskData([{}])],
'Create Ticket': [makeTaskData([{ id: 9 }])],
Agent: [makeTaskData([{ answer: 'old' }])],
};
const { runData, result } = await runStepOn(
agentWorkflow,
'Calculator',
{ reuseExecutionId: 'exec-past' },
{ execution: makeExecution({ status: 'success', runData: priorRunData }) },
);
expect(runData.runData).toEqual(priorRunData);
expect(runData.dirtyNodeNames).toEqual(['Calculator']);
expect(result.inputMode).toBe('reused-execution');
expect(result.ranThroughNodeNames).toEqual(['Agent']);
});
it('refuses to replay an execution where a node above the Agent failed', async () => {
// The engine retries a failed node instead of reusing it, so "Create
// Ticket" would send its request again.
const harness = createRunAdapterForTests(agentWorkflow, {
execution: makeExecution({
status: 'error',
runData: {
Trigger: [makeTaskData([{}])],
'Create Ticket': [makeTaskData([], { error: new Error('timed out') })],
},
}),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
await expect(
runStep('wf-1', 'Calculator', { reuseExecutionId: 'exec-past' }),
).rejects.toThrow('so the run would execute them for real (Create Ticket)');
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
it('omits the root node list for a node that is its own root', async () => {
const { result } = await runStepOn(agentWorkflow, 'Agent', { mockInput: [{ a: 1 }] });
expect(result).not.toHaveProperty('ranThroughNodeNames');
});
it('passes the tool its arguments through the agent request', async () => {
const { runData } = await runStepOn(agentWorkflow, 'Calculator', {
mockInput: [{}],
toolArguments: { input: '2 + 2' },
});
// `rewireGraph` copies this onto the virtual Tool Executor, which runs the
// only connected tool when the request names none, and looks the arguments
// up by the tool's runtime name.
expect(runData.agentRequest).toEqual({
query: { Calculator: { input: '2 + 2' } },
tool: { name: '' },
});
});
it('sends an empty argument set for a tool that declares none', async () => {
const { runData } = await runStepOn(agentWorkflow, 'Calculator', { mockInput: [{}] });
expect(runData.agentRequest).toEqual({
query: { Calculator: {} },
tool: { name: '' },
});
});
it('refuses a tool that takes its arguments from the agent when none are given', async () => {
const withFromAi = {
...agentWorkflow,
nodes: agentWorkflow.nodes.map((node) =>
node.name === 'Calculator'
? {
...node,
name: 'Create Ticket Tool',
type: 'n8n-nodes-base.httpRequestTool',
parameters: { url: "={{ $fromAI('url') }}", body: "={{ $fromAI('title') }}" },
}
: node,
),
connections: {
Trigger: agentWorkflow.connections.Trigger,
'Create Ticket': agentWorkflow.connections['Create Ticket'],
'Create Ticket Tool': { ai_tool: [[{ node: 'Agent', type: 'ai_tool', index: 0 }]] },
},
};
const harness = createRunAdapterForTests(withFromAi, {
execution: makeExecution({ status: 'success' }),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
// Running it with empty arguments fails for a reason that has nothing to
// do with the user's problem, and the model would report that failure.
await expect(runStep('wf-1', 'Create Ticket Tool', { mockInput: [{}] })).rejects.toThrow(
'Pass toolArguments with: url, title',
);
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
it('refuses a node that holds several tools', async () => {
// The Tool Executor matches a member by a name built from the node name
// and the server's tool name (`buildMcpToolName`), so nothing here can
// name one. A miss reports success with no result and no error.
const harness = createRunAdapterForTests(toolkitWorkflow, {
execution: makeExecution({ status: 'success' }),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
await expect(runStep('wf-1', 'MCP Client', { mockInput: [{}] })).rejects.toThrow(
'holds several tools, and a step run cannot pick one of them',
);
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
describe('a tool shared by two Agents', () => {
const agentNode = (name: string, x: number) => ({
name,
type: '@n8n/n8n-nodes-langchain.agent',
typeVersion: 1,
position: [x, 0],
});
const sharedToolWorkflow = (main: Record<string, unknown>) => ({
id: 'wf-1',
versionId: 'v-current',
nodes: [
{
name: 'Trigger',
type: 'n8n-nodes-base.manualTrigger',
typeVersion: 1,
position: [0, 0],
},
agentNode('Agent A', 1),
{ name: 'POST', type: 'n8n-nodes-base.httpRequest', typeVersion: 1, position: [2, 0] },
agentNode('Agent B', 3),
{
name: 'Calculator',
type: '@n8n/n8n-nodes-langchain.toolCalculator',
typeVersion: 1,
position: [2, 1],
},
],
connections: {
...main,
Calculator: {
ai_tool: [
[
{ node: 'Agent B', type: 'ai_tool', index: 0 },
{ node: 'Agent A', type: 'ai_tool', index: 0 },
],
],
},
},
});
it('refuses the run when one Agent runs above the other', async () => {
// Trigger -> Agent A -> POST -> Agent B. The engine can pick Agent B,
// and then Agent A and "POST" run for real.
const harness = createRunAdapterForTests(
sharedToolWorkflow({
Trigger: { main: [[{ node: 'Agent A', type: 'main', index: 0 }]] },
'Agent A': { main: [[{ node: 'POST', type: 'main', index: 0 }]] },
POST: { main: [[{ node: 'Agent B', type: 'main', index: 0 }]] },
}),
{ execution: makeExecution({ status: 'success' }) },
);
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
await expect(runStep('wf-1', 'Calculator', { mockInput: [{}] })).rejects.toThrow(
'"Agent A" runs above another of them, so a step run would run "Agent A" and the nodes after it again',
);
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
it('runs the tool when the Agents sit on parallel branches', async () => {
const { result } = await runStepOn(
sharedToolWorkflow({
Trigger: {
main: [
[
{ node: 'Agent A', type: 'main', index: 0 },
{ node: 'POST', type: 'main', index: 0 },
],
],
},
POST: { main: [[{ node: 'Agent B', type: 'main', index: 0 }]] },
}),
'Calculator',
{ mockInput: [{}] },
);
expect(result.inputMode).toBe('mocked');
expect(result.ranThroughNodeNames?.sort()).toEqual(['Agent A', 'Agent B']);
});
});
// The engine records a tool's run under `ai_tool`, and the node that
// carries the run — `PartialExecutionToolExecutor` — is not in the
// workflow. Reported as-is, the result hid the tool and named a node the
// user cannot open.
describe('the result of a tool step run', () => {
/** An execution shaped the way the engine leaves a tool step run. */
const toolRunExecution = () =>
makeExecution({
status: 'success',
lastNodeExecuted: TOOL_EXECUTOR_NODE_NAME,
runData: {
Trigger: [makeTaskData([{ ok: true }])],
// The tool keeps its items, under the connection that carried them.
Calculator: [
{
startTime: 1,
executionTime: 1,
executionIndex: 2,
executionStatus: 'success',
source: [],
data: { ai_tool: [[{ json: { response: 4 } }]] },
} as unknown as ITaskData,
],
// The executor re-serializes the same result as one string.
[TOOL_EXECUTOR_NODE_NAME]: [makeTaskData([{ json: '4' } as unknown as IDataObject])],
},
});
it("reports the tool's own output under the tool's name", async () => {
const { result } = await runStepOn(
agentWorkflow,
'Calculator',
{ mockInput: [{}] },
{ execution: toolRunExecution(), allowSendingParameterValues: true },
);
expect(result.data).toHaveProperty('Calculator');
expect(String(result.data?.Calculator)).toContain('"response": 4');
});
it('never names the virtual node the workflow does not contain', async () => {
const { result } = await runStepOn(
agentWorkflow,
'Calculator',
{ mockInput: [{}] },
{ execution: toolRunExecution(), allowSendingParameterValues: true },
);
expect(result.data).not.toHaveProperty(TOOL_EXECUTOR_NODE_NAME);
expect(result.executedNodeNames).not.toContain(TOOL_EXECUTOR_NODE_NAME);
expect(result.lastNodeExecuted).toBe('Calculator');
});
it('reports an error on the virtual node against the tool', async () => {
const failed = makeExecution({
status: 'error',
lastNodeExecuted: TOOL_EXECUTOR_NODE_NAME,
workflowNodes: agentWorkflow.nodes as WorkflowNode[],
runData: {
[TOOL_EXECUTOR_NODE_NAME]: [
{
startTime: 1,
executionTime: 1,
executionIndex: 1,
executionStatus: 'error',
source: [],
error: { message: 'tool blew up' },
} as unknown as ITaskData,
],
},
});
const { result } = await runStepOn(
agentWorkflow,
'Calculator',
{ mockInput: [{}] },
{ execution: failed },
);
expect(result.nodeErrors?.map((nodeError) => nodeError.nodeName)).toEqual(['Calculator']);
});
it('leaves a run on a node in the main graph alone', async () => {
const { result } = await runStepOn(chainWorkflow, 'Send', undefined, {
allowSendingParameterValues: true,
execution: makeExecution({
status: 'success',
lastNodeExecuted: 'Send',
runData: { Send: [makeTaskData([{ sent: true }])] },
}),
});
expect(result.lastNodeExecuted).toBe('Send');
expect(result.data).toHaveProperty('Send');
});
});
// A tool whose output goes nowhere has no node to run it. The plan reads
// as a chain run, which the engine would start and then kill with an error
// about a graph the caller never asked about.
describe('a tool nothing runs', () => {
/** The same graph with the tool's connection to the Agent removed. */
const orphanToolWorkflow = {
...agentWorkflow,
connections: {
Trigger: agentWorkflow.connections.Trigger,
'Create Ticket': agentWorkflow.connections['Create Ticket'],
},
};
it('refuses the run instead of starting one that cannot work', async () => {
const harness = createRunAdapterForTests(orphanToolWorkflow, {
execution: makeExecution({ status: 'success' }),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
await expect(runStep('wf-1', 'Calculator', { mockInput: [{}] })).rejects.toThrow(
'no node is connected to run it',
);
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
it('does not claim the tool runs in the main graph', async () => {
const harness = createRunAdapterForTests(orphanToolWorkflow, {
execution: makeExecution({ status: 'success' }),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
// The old message said `toolArguments` applies only to a tool node and
// that this one runs in the main graph. Both are wrong about a tool.
const failure = await runStep('wf-1', 'Calculator', {
toolArguments: { input: '2 + 2' },
}).catch((error: Error) => error);
expect(failure).toBeInstanceOf(Error);
expect((failure as Error).message).not.toContain('runs in the main graph');
expect((failure as Error).message).toContain('connect it to an Agent');
});
});
it('refuses a node the MCP registry added, whose type carries the server slug', async () => {
// The registry saves a server as `@n8n/mcp-registry.<slug>` and routes
// every one of them to one hidden runtime class, so the type on the
// canvas is never that class's name. Matching the class alone let this
// node through, and the Tool Executor then matched no member and
// reported success with no result.
const registryWorkflow = {
...toolkitWorkflow,
nodes: toolkitWorkflow.nodes.map((node) =>
node.name === 'MCP Client'
? { ...node, name: 'Linear', type: '@n8n/mcp-registry.linear' }
: node,
),
connections: {
Trigger: agentWorkflow.connections.Trigger,
'Create Ticket': agentWorkflow.connections['Create Ticket'],
Linear: { ai_tool: [[{ node: 'Agent', type: 'ai_tool', index: 0 }]] },
},
};
const harness = createRunAdapterForTests(registryWorkflow, {
execution: makeExecution({ status: 'success' }),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
await expect(runStep('wf-1', 'Linear', { mockInput: [{}] })).rejects.toThrow(
'holds several tools, and a step run cannot pick one of them',
);
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
it('refuses a sub-node that is not a tool', async () => {
const withModel = {
...agentWorkflow,
nodes: [
...agentWorkflow.nodes,
{
name: 'OpenAI Chat Model',
type: '@n8n/n8n-nodes-langchain.lmChatOpenAi',
typeVersion: 1,
position: [3, 1],
},
],
connections: {
...agentWorkflow.connections,
'OpenAI Chat Model': {
ai_languageModel: [[{ node: 'Agent', type: 'ai_languageModel', index: 0 }]],
},
},
};
const harness = createRunAdapterForTests(withModel, {
execution: makeExecution({ status: 'success' }),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
// The engine has no Tool Executor path for a model, so the run would fail
// with "connect a trigger" and teach the model nothing.
await expect(runStep('wf-1', 'OpenAI Chat Model', undefined)).rejects.toThrow(
'cannot run on its own — n8n runs it as part of "Agent"',
);
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
it('carries the agent request to a worker in queue mode', async () => {
const previous = process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS;
process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS = 'true';
try {
const { runData } = await runStepOn(
agentWorkflow,
'Calculator',
{ mockInput: [{}], toolArguments: { input: '2 + 2' } },
{ queueMode: true },
);
// The transient top-level fields do not survive the hand-off, so the
// worker rebuilds the run from `execution.data`.
expect(runData.executionData.manualData.agentRequest).toEqual({
query: { Calculator: { input: '2 + 2' } },
tool: { name: '' },
});
} finally {
// Assigning `undefined` stores the string "undefined", which leaves the
// variable set for every later test in this file.
if (previous === undefined) delete process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS;
else process.env.OFFLOAD_MANUAL_EXECUTIONS_TO_WORKERS = previous;
}
});
});
it('refuses tool arguments for a node in the main graph', async () => {
const harness = createRunAdapterForTests(chainWorkflow, {
execution: makeExecution({ status: 'success' }),
});
const runStep = harness.adapter.runStep as NonNullable<typeof harness.adapter.runStep>;
await expect(runStep('wf-1', 'Send', { toolArguments: { a: 1 } })).rejects.toThrow(
'applies only to a tool node',
);
expect(harness.mockWorkflowRunner.run).not.toHaveBeenCalled();
});
});
describe('getGatewayConfigOrNull', () => {

View File

@ -1,4 +1,6 @@
import type {
AiAgentRequest,
FromAIArgument,
IConnections,
IDataObject,
INode,
@ -7,7 +9,15 @@ import type {
IRunData,
ITaskData,
} from 'n8n-workflow';
import { mapConnectionsByDestination, NodeConnectionTypes } from 'n8n-workflow';
import {
mapConnectionsByDestination,
MCP_CLIENT_TOOL_NODE_TYPE,
NodeConnectionTypes,
nodeNameToToolName,
traverseNodeParameters,
} from 'n8n-workflow';
import { MCP_REGISTRY_PACKAGE_NAME } from '@/modules/mcp-registry/mcp-registry-connection';
/**
* Plans the run data an "execute step" run needs.
@ -37,6 +47,24 @@ export interface StepRunPlan {
mockedNodeNames: string[];
/** Nodes replaying output from an earlier execution. */
reusedNodeNames: string[];
/**
* Root nodes the target runs through, when the target is a sub-node. Absent
* for a normal node, which is its own root.
*/
rootNodeNames?: string[];
/**
* Set when the requested run data does not cover the path, and a run would
* execute real nodes above the target. The caller must refuse the run
* instead of silently downgrading it.
*
* Mocked input reaches this only through a Loop Over Items node: the mock
* leaves its `done` output empty, and the engine restarts a loop that did
* not finish.
*/
unhonoredInput?: {
requested: 'reused-execution' | 'mocked';
upstreamNodeNames: string[];
};
}
/** One inbound edge of a node, after disabled nodes are skipped. */
@ -101,6 +129,126 @@ function directParents(
return edges;
}
/** One outbound edge of a node, after disabled nodes are skipped. */
interface ChildEdge {
node: INode;
/** Output of this node that feeds the edge. */
outputIndex: number;
}
/**
* Direct main-connection children of `nodeName`, the mirror of
* `directParents`. A disabled node is transparent here too: the engine joins
* its parents to its children, and the joined edge keeps the *parent's* output
* index.
*/
function directChildren(
connections: IConnections,
nodesByName: Map<string, INode>,
nodeName: string,
outputIndex?: number,
seen: Set<string> = new Set(),
): ChildEdge[] {
if (seen.has(nodeName)) return [];
seen.add(nodeName);
const outputs = connections[nodeName]?.[NodeConnectionTypes.Main] ?? [];
const edges: ChildEdge[] = [];
for (const [index, output] of outputs.entries()) {
for (const connection of output ?? []) {
const child = nodesByName.get(connection.node);
if (!child) continue;
const carriedBy = outputIndex ?? index;
if (child.disabled) {
edges.push(...directChildren(connections, nodesByName, child.name, carriedBy, seen));
continue;
}
edges.push({ node: child, outputIndex: carriedBy });
}
}
return edges;
}
/**
* Nodes this one feeds through a non-main connection — a sub-node's link to the
* node that runs it (`ai_tool` to an Agent, `ai_embedding` to a Vector Store).
* Sorted, so a tool on two agents resolves the same way on every call.
*/
function nonMainChildren(
connections: IConnections,
nodesByName: Map<string, INode>,
nodeName: string,
): string[] {
const children = new Set<string>();
for (const [type, outputs] of Object.entries(connections[nodeName] ?? {})) {
if (type === NodeConnectionTypes.Main) continue;
for (const output of outputs ?? []) {
for (const connection of output ?? []) {
if (nodesByName.has(connection.node)) children.add(connection.node);
}
}
}
return [...children].sort();
}
/**
* The nodes whose main ancestry a step run has to cover.
*
* A sub-node has no main input of its own. The engine never runs one on its
* own either: `rewireGraph` replaces the root node (the Agent) with a virtual
* Tool Executor that inherits *the root node's* main parents, and runs the
* sub-node from there. So the nodes that need run data are the root node's
* ancestors, not the sub-node's — it has none, and planning from the sub-node
* finds nothing to mock and falls back to a chain run that executes the real
* nodes above the Agent.
*
* Normal nodes are their own root, so this returns `[targetName]` for them.
* A sub-node on several roots returns all of them: `rewireGraph` picks one, and
* covering every candidate keeps the plan correct whichever it picks — unless
* one root sits above another, which `findRootsAboveOtherRoots` detects.
*/
export function resolveStepRunRoots(
nodes: INode[],
connections: IConnections,
targetName: string,
): string[] {
const nodesByName = new Map(nodes.map((node) => [node.name, node]));
const connectionsByDestination = mapConnectionsByDestination(connections);
const roots = new Set<string>();
const seen = new Set<string>([targetName]);
const queue = [targetName];
while (queue.length > 0) {
const current = queue.shift() as string;
// A main input means the node sits in the main graph and is its own root.
if (directParents(connectionsByDestination, nodesByName, current).length > 0) {
roots.add(current);
continue;
}
const parents = nonMainChildren(connections, nodesByName, current);
if (parents.length === 0) {
// A trigger, an orphan, or a sub-node with nothing to run it.
roots.add(current);
continue;
}
for (const parent of parents) {
if (seen.has(parent)) continue;
seen.add(parent);
queue.push(parent);
}
}
return [...roots];
}
/**
* Builds run data that makes `targetName` the only node the engine re-runs.
*
@ -117,8 +265,11 @@ export function buildMockedStepRunData(args: {
connections: IConnections;
targetName: string;
mockItems: INodeExecutionData[];
/** Where to start walking up. Defaults to the target; see `resolveStepRunRoots`. */
rootNames?: string[];
}): { runData: IRunData; mockedNodeNames: string[] } {
const { nodes, connections, targetName, mockItems } = args;
const rootNames = args.rootNames ?? [targetName];
const nodesByName = new Map(nodes.map((node) => [node.name, node]));
const connectionsByDestination = mapConnectionsByDestination(connections);
@ -130,7 +281,9 @@ export function buildMockedStepRunData(args: {
// feeding several paths is written once — the first (shallowest) edge wins,
// which is the edge closest to the target.
const visited = new Set<string>([targetName]);
let frontier: ParentEdge[] = directParents(connectionsByDestination, nodesByName, targetName);
let frontier: ParentEdge[] = rootNames.flatMap((rootName) =>
directParents(connectionsByDestination, nodesByName, rootName),
);
let depth = 0;
while (frontier.length > 0) {
@ -164,34 +317,209 @@ export function collectAncestorNames(
nodes: INode[],
connections: IConnections,
targetName: string,
/** Where to start walking up. Defaults to the target; see `resolveStepRunRoots`. */
rootNames: string[] = [targetName],
): string[] {
const nodesByName = new Map(nodes.map((node) => [node.name, node]));
const connectionsByDestination = mapConnectionsByDestination(connections);
const seen = new Set<string>();
const queue = [targetName];
// The target is never its own ancestor, even when an edge loops back to it.
// It runs either way, and `buildMockedStepRunData` skips it for the same
// reason — the two walks have to agree, or a plan can report a node above
// the target that it never mocked.
const ancestors = new Set<string>();
const visited = new Set<string>([targetName]);
const queue = [...rootNames];
while (queue.length > 0) {
const current = queue.shift() as string;
for (const edge of directParents(connectionsByDestination, nodesByName, current)) {
if (seen.has(edge.node.name)) continue;
if (visited.has(edge.node.name)) continue;
visited.add(edge.node.name);
ancestors.add(edge.node.name);
queue.push(edge.node.name);
}
}
return [...ancestors];
}
/**
* The roots of a shared sub-node that sit above another of its roots.
*
* `rewireGraph` stands in for one root only. When it picks a lower one, a root
* above it stays in the run with its edge to the sub-node, so `cleanRunData`
* treats that root as a child of the dirty target. It then drops the run data
* of that root and of every node between it and the chosen root, and the
* engine runs them for real. Roots on parallel branches never enter the run,
* so this returns nothing for them.
*/
export function findRootsAboveOtherRoots(
nodes: INode[],
connections: IConnections,
rootNames: string[],
): string[] {
if (rootNames.length < 2) return [];
const above = new Set<string>();
for (const rootName of rootNames) {
const ancestors = new Set(collectAncestorNames(nodes, connections, rootName));
for (const other of rootNames) {
if (other !== rootName && ancestors.has(other)) above.add(other);
}
}
return rootNames.filter((name) => above.has(name));
}
/**
* Whether an earlier run left items on one output of a node. Mirrors the
* engine's `getIncomingDataFromAnyRun`: any run of the node counts, and an
* empty output does not. The engine follows only the outputs that carried
* items, so an untaken IF branch never runs on a replay.
*/
function outputCarriedItems(runData: IRunData, nodeName: string, outputIndex: number): boolean {
return (runData[nodeName] ?? []).some(
(task) => (task.data?.[NodeConnectionTypes.Main]?.[outputIndex] ?? []).length > 0,
);
}
/** Loop Over Items. The engine's `findStartNodes` matches the same literal. */
const SPLIT_IN_BATCHES_NODE_TYPE = 'n8n-nodes-base.splitInBatches';
/**
* Whether `nodeName` can reach itself through main connections without
* passing a node in `avoid`. `findSubgraph` drops every cycle that runs
* through the destination, so such a cycle is no loop to the engine.
*/
function reachesItself(
connections: IConnections,
nodesByName: Map<string, INode>,
nodeName: string,
avoid: Set<string>,
): boolean {
const seen = new Set<string>();
const queue = [nodeName];
while (queue.length > 0) {
const current = queue.shift() as string;
for (const edge of directChildren(connections, nodesByName, current)) {
if (edge.node.name === nodeName) return true;
if (avoid.has(edge.node.name) || seen.has(edge.node.name)) continue;
seen.add(edge.node.name);
queue.push(edge.node.name);
}
}
return [...seen];
return false;
}
/**
* The nodes above the target that a replay of `runData` would still execute
* for real.
*
* Having run data for *some* ancestor is not enough. The engine walks down
* from the trigger and makes the first node with no run data a start node, so
* a gap anywhere on the path puts that node and everything after it back in
* the run. A gap at the trigger is the worst case: the whole chain runs again.
*
* This walk answers the same question the engine asks, with the same rules
* as `findStartNodes`:
* - A node with run data or pin data is clean, unless one of its runs failed.
* The engine retries a failed node, even a pinned one.
* - A Loop Over Items node whose last run left its `done` output empty did not
* finish, so the engine restarts the loop.
* - Only an output that carried items leads anywhere.
*
* It therefore accepts the replays the engine can honour — an untaken branch
* above the target needs no data, and a second trigger the earlier run never
* fired needs none either, because the engine prefers the trigger that has run
* data.
*/
export function findUncoveredAncestors(args: {
nodes: INode[];
connections: IConnections;
targetName: string;
rootNames: string[];
runData: IRunData;
pinnedNodeNames?: string[];
}): string[] {
const { nodes, connections, targetName, rootNames, runData } = args;
const nodesByName = new Map(nodes.map((node) => [node.name, node]));
const connectionsByDestination = mapConnectionsByDestination(connections);
const pinned = new Set(args.pinnedNodeNames ?? []);
const ancestors = new Set(collectAncestorNames(nodes, connections, targetName, rootNames));
if (ancestors.size === 0) return [];
const failed = (name: string) => (runData[name] ?? []).some((task) => task.error !== undefined);
const isClean = (name: string) =>
!failed(name) && (runData[name] !== undefined || pinned.has(name));
const stopAt = new Set([...rootNames, targetName]);
// Mirrors the engine's `isALoop` on the subgraph: without a cycle back to
// the node, the Loop Over Items node is a plain node and its second output
// leads on.
const isUnfinishedLoop = (name: string) => {
if (nodesByName.get(name)?.type !== SPLIT_IN_BATCHES_NODE_TYPE) return false;
const doneOutput = reachesItself(connections, nodesByName, name, stopAt) ? 0 : 1;
const lastRun = runData[name]?.at(-1);
return (lastRun?.data?.[NodeConnectionTypes.Main]?.[doneOutput] ?? []).length === 0;
};
// A run starts at a trigger. When the one the engine would pick has no data
// of its own it re-runs, and takes every node under it along.
const sources = [...ancestors].filter(
(name) => directParents(connectionsByDestination, nodesByName, name).length === 0,
);
// The engine applies the loop rule to the node its walk starts from as well.
const startPoints = sources.filter((name) => isClean(name) && !isUnfinishedLoop(name));
if (startPoints.length === 0) return [...ancestors];
const uncovered = new Set<string>();
const seen = new Set<string>(startPoints);
const queue = [...startPoints];
while (queue.length > 0) {
const current = queue.shift() as string;
for (const edge of directChildren(connections, nodesByName, current)) {
const child = edge.node.name;
// The root runs by design — it is the node the target runs through — and
// a node off the path to the target is not part of the run at all.
if (stopAt.has(child) || !ancestors.has(child)) continue;
// A pinned node feeds every output, so the engine always walks on.
if (!pinned.has(current) && !outputCarriedItems(runData, current, edge.outputIndex)) continue;
if (!isClean(child) || isUnfinishedLoop(child)) {
uncovered.add(child);
continue;
}
if (seen.has(child)) continue;
seen.add(child);
queue.push(child);
}
}
return [...uncovered];
}
/**
* Chooses the run data for a step run.
*
* - `mockItems` given → mock the path (mode `mocked`).
* - `priorRunData` given → replay it and mark the target dirty so it runs again
* - `priorRunData` given (even empty) → replay it and mark the target dirty so
* it runs again
* (mode `reused-execution`). The engine's `cleanRunData` drops the target and
* everything downstream of it, so stale output cannot survive the run.
* - neither → leave the run data unset and let the engine run the chain
* (mode `chain`), which is what the canvas "Execute step" does.
*
* A sub-node target is planned through its root node — see
* `resolveStepRunRoots`. When a requested mode cannot be honoured and the
* target has real nodes above it, the plan says so in `unhonoredInput` rather
* than falling back to a chain run that would execute them.
*/
export function planStepRun(args: {
nodes: INode[];
@ -199,52 +527,119 @@ export function planStepRun(args: {
targetName: string;
mockItems?: INodeExecutionData[];
priorRunData?: IRunData;
/** Nodes the workflow pins. A pinned node is clean, so a replay may skip it. */
pinnedNodeNames?: string[];
}): StepRunPlan & { inputMode: StepRunInputMode } {
const { nodes, connections, targetName, mockItems, priorRunData } = args;
// A sub-node runs through its root node, so the run data has to cover the
// root's ancestry. Planning from the sub-node itself finds no main parent.
const rootNames = resolveStepRunRoots(nodes, connections, targetName);
const throughRoots =
rootNames.length === 1 && rootNames[0] === targetName
? undefined
: { rootNodeNames: rootNames };
if (mockItems !== undefined) {
const { runData, mockedNodeNames } = buildMockedStepRunData({
nodes,
connections,
targetName,
mockItems,
rootNames,
});
// A node with no enabled parent has nothing to mock. Running the chain
// gives the engine a start point it accepts, instead of a partial run it
// rejects for having no reachable root with run data.
if (mockedNodeNames.length > 0) {
// The mock gives every ancestor run data, so the one rule it can break
// is the loop rule. Other findings, such as a missing trigger, are the
// walk's guesses about a graph the engine starts elsewhere. Pins do not
// matter either: the run drops the pin of every node it mocks.
const loopNames = new Set(
nodes.filter((n) => n.type === SPLIT_IN_BATCHES_NODE_TYPE).map((n) => n.name),
);
const notMocked = findUncoveredAncestors({
nodes,
connections,
targetName,
rootNames,
runData,
}).filter((name) => loopNames.has(name));
if (notMocked.length > 0) {
return {
inputMode: 'chain',
mockedNodeNames: [],
reusedNodeNames: [],
...throughRoots,
unhonoredInput: { requested: 'mocked', upstreamNodeNames: notMocked },
};
}
return {
inputMode: 'mocked',
runData,
dirtyNodeNames: [targetName],
mockedNodeNames,
reusedNodeNames: [],
...throughRoots,
};
}
}
if (priorRunData !== undefined && Object.keys(priorRunData).length > 0) {
const ancestors = new Set(collectAncestorNames(nodes, connections, targetName));
const reusedNodeNames = Object.keys(priorRunData).filter((name) => ancestors.has(name));
const ancestors = collectAncestorNames(nodes, connections, targetName, rootNames);
// Reuse is only worth it when the prior run actually reached an ancestor.
// Otherwise fall through to a chain run rather than start an execution
// that the engine would reject for having no usable start point.
if (reusedNodeNames.length > 0) {
// Nodes the replay leaves for the engine to run for real. Empty means the
// reused execution covers every node the run would otherwise execute.
let uncovered: string[] = [];
if (priorRunData !== undefined) {
const ancestorNames = new Set(ancestors);
const reusedNodeNames = Object.keys(priorRunData).filter((name) => ancestorNames.has(name));
uncovered = findUncoveredAncestors({
nodes,
connections,
targetName,
rootNames,
runData: priorRunData,
pinnedNodeNames: args.pinnedNodeNames,
});
if (reusedNodeNames.length > 0 && uncovered.length === 0) {
return {
inputMode: 'reused-execution',
runData: priorRunData,
dirtyNodeNames: [targetName],
mockedNodeNames: [],
reusedNodeNames,
...throughRoots,
};
}
}
// The chain is the intended mode only when the caller asked for it. Reaching
// it after a request to replay an execution means the plan could not keep the
// nodes above the target from running, and those nodes write to real systems
// — report it instead of running them.
//
// A request for mocked input reaches here only when it mocks nothing, which
// means the target has nothing above it, and then a chain run is what the
// caller wanted anyway.
return {
inputMode: 'chain',
mockedNodeNames: [],
reusedNodeNames: [],
...throughRoots,
...(priorRunData !== undefined && ancestors.length > 0
? {
unhonoredInput: {
requested: 'reused-execution',
// Name the nodes that would really run, which is the part of the
// ancestry the reused execution does not cover.
upstreamNodeNames: uncovered.length > 0 ? uncovered : ancestors,
},
}
: {}),
};
}
@ -272,6 +667,86 @@ export function pinDataForStepRun(
return Object.fromEntries(kept);
}
/**
* Node types that supply a toolkit — several tools behind one node — rather than
* one tool.
*
* A step run cannot target one of these. The Tool Executor runs the member whose
* name matches the agent request and skips every other, and that name is not the
* server's tool name: `buildMcpToolName` prefixes it with the node's name and
* caps the result at 64 characters. Resolving it needs the server's tool list
* and a rule that lives in the nodes package, and a name that misses matches
* nothing — the run then reports success with no result and no error.
*/
const TOOLKIT_NODE_TYPES = new Set<string>([MCP_CLIENT_TOOL_NODE_TYPE]);
/**
* The MCP registry saves a server as its own node type, `@n8n/mcp-registry.<slug>`,
* and routes every one of them to a single hidden runtime class. So the match is
* on the package, the way `agents-tools.service.ts` decides the same question:
* the class name never appears as a node type, and each slug is a type of its own.
*/
const MCP_REGISTRY_NODE_TYPE_PREFIX = `${MCP_REGISTRY_PACKAGE_NAME}.`;
/** Whether this node holds several tools instead of one. */
export function isToolkitNode(node: INode): boolean {
return TOOLKIT_NODE_TYPES.has(node.type) || node.type.startsWith(MCP_REGISTRY_NODE_TYPE_PREFIX);
}
/** Arguments a tool node expects an agent to fill, from its `$fromAI` calls. */
export function declaredToolArguments(node: INode): string[] {
const collected: FromAIArgument[] = [];
traverseNodeParameters(node.parameters, collected);
return [...new Set(collected.map((argument) => argument.key))];
}
/**
* Parameters that hold a tool's name on the versions that take it from the
* node's configuration instead of its name: `name` on Code Tool <= 1.1, Vector
* Store Tool <= 1 and Workflow Tool <= 2.1, `toolName` on a vector store in
* retrieve-as-tool mode < 1.3.
*/
const TOOL_NAME_PARAMETERS = ['name', 'toolName'];
/**
* The agent request that gives a tool its arguments.
*
* `rewireGraph` copies this onto the virtual Tool Executor, which looks the
* arguments up by the tool's *runtime* name. That name is
* `nodeNameToToolName(node)` on current tool versions, but older ones read it
* from a parameter and Think 1 hardcodes `thinking_tool`. Guessing it would
* need every node's version rule, and those rules live in the nodes package.
*
* So the request names no tool. The Tool Executor runs the only tool the
* rewired graph connects to it when the request leaves the name empty, which
* takes the runtime name out of the decision to run at all. The arguments are
* keyed under every name the tool can have, so the lookup finds them whichever
* one it uses. A name this cannot know (Think 1) costs the arguments, not the
* run.
*
* A bare string is a valid argument set: a tool with one free-text input
* (Wikipedia, Code Tool, a vector store used as a tool) takes the query
* directly, not wrapped in an object.
*/
export function buildToolAgentRequest(args: {
target: INode;
toolArguments?: Record<string, unknown> | string;
}): AiAgentRequest {
const { target } = args;
const toolArguments = args.toolArguments ?? {};
const names = new Set<string>([nodeNameToToolName(target.name), target.name]);
for (const parameter of TOOL_NAME_PARAMETERS) {
const configured = target.parameters?.[parameter];
if (typeof configured === 'string' && configured !== '') names.add(configured);
}
return {
query: Object.fromEntries([...names].map((name) => [name, toolArguments])),
tool: { name: '' },
};
}
/**
* Converts caller-supplied plain objects into engine items. The cast mirrors
* `sdkPinDataToRuntime`: the model hands us arbitrary JSON, and the engine

View File

@ -23,7 +23,7 @@ import type { AiGatewayConfigDto } from '@n8n/api-types';
import { LicenseState, Logger, ModuleRegistry } from '@n8n/backend-common';
import { OutboundHttp } from '@n8n/backend-network';
import { GlobalConfig } from '@n8n/config';
import { Time } from '@n8n/constants';
import { Time, TOOL_EXECUTOR_NODE_NAME } from '@n8n/constants';
import type { User, ExecutionSummaries, EvaluationConfig } from '@n8n/db';
import {
AiBuilderTemporaryWorkflowRepository,
@ -60,6 +60,7 @@ import type {
ExecutionResult,
StepExecutionResult,
ExecutionDebugInfo,
NodeOutputBranch,
NodeOutputResult,
ResolvedNodeParametersResult,
ExecutionSummary as InstanceAiExecutionSummary,
@ -116,12 +117,14 @@ import {
type IWorkflowBase,
type IWorkflowSettings,
type IWorkflowExecutionDataProcess,
type AiAgentRequest,
type DataTableRow,
type DataTableRows,
type WorkflowExecuteMode,
type ExecutionError,
type IRunData,
type ITaskData,
NodeConnectionTypes,
NodeHelpers,
Workflow,
CHAT_TRIGGER_NODE_TYPE,
@ -209,7 +212,15 @@ import {
sdkPinDataToRuntime,
} from './instance-ai-run-pin-data';
import { InstanceAiSettingsService } from './instance-ai-settings.service';
import { pinDataForStepRun, planStepRun, toExecutionItems } from './instance-ai-step-run';
import {
buildToolAgentRequest,
declaredToolArguments,
findRootsAboveOtherRoots,
isToolkitNode,
pinDataForStepRun,
planStepRun,
toExecutionItems,
} from './instance-ai-step-run';
import { InstanceContextService } from './instance-context.service';
import type { InstanceContextScope } from './instance-context.service';
import { InstanceAiMcpRegistryService } from './mcp';
@ -253,6 +264,21 @@ function resolveDisplayedDefaults(
return resolved ?? (parameters as INodeParameters);
}
/**
* Whether the engine can run this node on its own through a Tool Executor. This
* is the same test `runPartialWorkflow2` and `runManually` make, so it decides
* the same way they will. A node type that does not load is not a tool here:
* the run fails on the missing type either way.
*/
function isToolNode(nodeTypes: NodeTypes, node: INode): boolean {
try {
const { description } = nodeTypes.getByNameAndVersion(node.type, node.typeVersion);
return NodeHelpers.isTool(description, node.parameters);
} catch {
return false;
}
}
/**
* A credential class's `documentationUrl` is normally a docs slug ("slack"), but a
* few declare a full URL. Normalizes both to a URL.
@ -2244,7 +2270,10 @@ export class InstanceAiAdapterService {
includeData: true,
unflattenData: true,
});
priorRunData = stored?.data?.resultData?.runData;
// An execution with no stored run data still counts as a request to
// replay: `planStepRun` has to see the empty set to refuse the run
// rather than fall back to running the chain.
priorRunData = stored?.data?.resultData?.runData ?? {};
reusedFromExecutionId = options.reuseExecutionId;
}
@ -2254,8 +2283,100 @@ export class InstanceAiAdapterService {
targetName: nodeName,
mockItems: options?.mockInput ? toExecutionItems(options.mockInput) : undefined,
priorRunData,
pinnedNodeNames: Object.keys(workflow.pinData ?? {}),
});
// The caller asked to keep the nodes above the target out of the run and
// the graph cannot deliver that. Running the chain anyway would execute
// them for real, which is the one thing the caller ruled out.
if (plan.unhonoredInput) {
const names = plan.unhonoredInput.upstreamNodeNames;
const upstream = names.slice(0, 10).join(', ') + (names.length > 10 ? ', …' : '');
if (plan.unhonoredInput.requested === 'mocked') {
throw new UserError(
`mockInput cannot keep every node above "${nodeName}" out of the run: ` +
`the mock leaves a Loop Over Items node unfinished, and n8n would restart it and execute these nodes for real (${upstream}). ` +
'Pass reuseExecutionId with an execution where the loop finished, or omit both options to run the chain on purpose.',
);
}
throw new UserError(
`Execution ${options?.reuseExecutionId} does not cover every node above "${nodeName}", ` +
`so the run would execute them for real (${upstream}). ` +
'Pick an execution where those nodes ran to completion, or omit both options to run the chain on purpose.',
);
}
// A tool runs through a virtual Tool Executor that passes it the agent
// request. Every other sub-node type has no such path: n8n runs it as
// part of the node that owns it and nowhere else.
let agentRequest: AiAgentRequest | undefined;
if (plan.rootNodeNames) {
const roots = plan.rootNodeNames.map((name) => `"${name}"`).join(' or ');
if (!isToolNode(nodeTypes, target)) {
throw new UserError(
`Node "${nodeName}" cannot run on its own — n8n runs it as part of ${roots}. ` +
`Run ${roots} instead: its execution records what this node returned on every ` +
'call it made.',
);
}
// A toolkit node holds several tools and the Tool Executor runs only the
// one whose name matches the request. That name is built from the node
// name and the server's tool name, so nothing here can name a member
// reliably, and a miss reports success with no result at all.
if (isToolkitNode(target)) {
throw new UserError(
`Node "${nodeName}" holds several tools, and a step run cannot pick one of them. ` +
`Run ${roots} instead, then read this node with ` +
'executions(action="get-node-output") on that execution: it holds what every ' +
'tool call returned.',
);
}
// The engine runs the tool through one of its Agents only. When it picks
// a lower one, the Agent above it and the nodes in between run for
// real, whatever mocked or replayed data the plan gave them.
const upperRoots = findRootsAboveOtherRoots(nodes, connections, plan.rootNodeNames);
if (upperRoots.length > 0) {
const upper = upperRoots.map((name) => `"${name}"`).join(' or ');
throw new UserError(
`Node "${nodeName}" runs through ${roots}, and ${upper} runs above another of them, ` +
`so a step run would run ${upper} and the nodes after it again. ` +
`Run ${upper} instead, then read this node with ` +
'executions(action="get-node-output") on that execution: it holds what every ' +
'tool call returned.',
);
}
// A tool whose parameters hold `$fromAI` calls gets those values from
// the agent. With no arguments it runs on empty ones and fails for a
// reason that has nothing to do with the user's problem.
const expected = declaredToolArguments(target);
if (expected.length > 0 && options?.toolArguments === undefined) {
throw new UserError(
`Node "${nodeName}" takes its arguments from the agent, so a step run has to supply them. ` +
`Pass toolArguments with: ${expected.join(', ')}.`,
);
}
agentRequest = buildToolAgentRequest({ target, toolArguments: options?.toolArguments });
} else if (isToolNode(nodeTypes, target)) {
// A tool runs only through the node that owns it, and nothing owns
// this one: its tool output goes nowhere. `rewireGraph` then finds no
// node to stand in for, so the run would start, cost an execution and
// die on an error about a graph the caller never asked about.
throw new UserError(
`Node "${nodeName}" is a tool, and no node is connected to run it. ` +
'A tool runs only through the node that owns it, so connect it to an Agent ' +
'and run the step again.',
);
} else if (options?.toolArguments !== undefined) {
throw new UserError(
`toolArguments applies only to a tool node. "${nodeName}" runs in the main graph, ` +
'so its input comes from mockInput, reuseExecutionId, or the nodes above it.',
);
}
// A pinned node never executes, so the target's own pin — and any pin on
// a node whose output we just mocked — has to come off this run's
// copy. The saved workflow keeps them.
@ -2289,6 +2410,7 @@ export class InstanceAiAdapterService {
pinData: stepPinData,
runData: plan.runData,
dirtyNodeNames: plan.dirtyNodeNames,
agentRequest,
source: 'instance_ai',
};
@ -2315,6 +2437,9 @@ export class InstanceAiAdapterService {
userId: user.id,
dirtyNodeNames: plan.dirtyNodeNames,
triggerToStartFrom: runData.triggerToStartFrom,
// Without this a worker-run step on a tool loses the arguments and
// runs the tool on empty ones.
agentRequest,
source: 'instance_ai',
},
executionData: null,
@ -2370,6 +2495,12 @@ export class InstanceAiAdapterService {
nodeName,
inputMode: plan.inputMode,
mockedNodeNames: plan.mockedNodeNames,
// A sub-node only runs through the node that owns it, so the input
// the caller supplied fed that node, not the sub-node. Every owner is
// listed, not the one that ran: `rewireGraph` picks one and reports
// nothing about the choice, so naming a single owner here would be a
// guess.
...(plan.rootNodeNames ? { ranThroughNodeNames: plan.rootNodeNames } : {}),
...(replayedNodeNames.length > 0 ? { replayedNodeNames } : {}),
...(reusedFromExecutionId ? { reusedFromExecutionId } : {}),
// Report the pins that actually fed this run, not every pin the
@ -2403,6 +2534,9 @@ export class InstanceAiAdapterService {
executionId,
allowSendingParameterValues,
nodeTypes,
// A tool ran through the virtual Tool Executor. Report the run
// under the node the caller named, which is the one they can open.
plan.rootNodeNames ? nodeName : undefined,
);
trackStepRun(result.status, telemetryError);
return describe(result);
@ -4410,6 +4544,49 @@ export async function extractExecutionResult(
return (await extractExecutionOutcome(executionId, includeOutputData, nodeTypes)).result;
}
/** Output branches of a run that a non-main connection carried. */
function nonMainOutputs(run: ITaskData | undefined) {
return Object.entries(run?.data ?? {}).find(([type]) => type !== NodeConnectionTypes.Main)?.[1];
}
/** Whether this is the virtual node the engine runs a tool through. */
const isToolExecutor = (nodeName: string | undefined) => nodeName === TOOL_EXECUTOR_NODE_NAME;
/** Reports the node the caller asked for in place of the virtual one. */
function renameToolExecutor(nodeName: string | undefined, subNodeTarget?: string) {
return subNodeTarget && isToolExecutor(nodeName) ? subNodeTarget : nodeName;
}
/**
* Folds the virtual Tool Executor's run into the node the step run targeted.
*
* `rewireGraph` runs a tool through a node the workflow does not contain, and
* that node's name reaches the result in four places — the output data, the
* executed names, the last node, and a node error. None of them can be looked
* up or opened, because the workflow has no such node.
*
* The tool's own run is the better record: the Tool Executor re-serializes the
* result as one string, while the tool keeps its items. So the executor's run
* only stands in when the tool recorded none of its own.
*/
function foldToolExecutorRun(
runData: IRunData | undefined,
subNodeTarget?: string,
): IRunData | undefined {
if (!runData || !subNodeTarget || !runData[TOOL_EXECUTOR_NODE_NAME]) return runData;
const folded: IRunData = {};
for (const [nodeName, nodeRuns] of Object.entries(runData)) {
if (isToolExecutor(nodeName)) {
folded[subNodeTarget] ??= nodeRuns;
continue;
}
folded[nodeName] = nodeRuns;
}
return folded;
}
/**
* The execution result the agent sees, plus the error string telemetry may use.
* They differ when `N8N_AI_ALLOW_SENDING_PARAMETER_VALUES` is on: that setting
@ -4422,6 +4599,11 @@ export async function extractExecutionOutcome(
executionId: string,
includeOutputData = true,
nodeTypes?: NodeTypes,
/**
* Sub-node a step run targeted. The engine ran it through a virtual node, so
* the result has to be told which node the caller actually asked for.
*/
subNodeTarget?: string,
): Promise<{ result: ExecutionResult; telemetryError?: string }> {
const execution = await Container.get(ExecutionPersistence).findSingleExecution(executionId, {
includeData: true,
@ -4448,7 +4630,7 @@ export async function extractExecutionOutcome(
// omits. Verification uses this to tell "ran and returned nothing" apart
// from "never reached". Node names only, so it is safe regardless of the
// parameter-values privacy setting.
const runData = execution.data?.resultData?.runData;
const runData = foldToolExecutorRun(execution.data?.resultData?.runData, subNodeTarget);
const executedNodeNames = Object.keys(runData ?? {});
if (includeOutputData && runData) {
const workflow = buildExecutionWorkflow(execution.workflowData, nodeTypes);
@ -4456,8 +4638,15 @@ export async function extractExecutionOutcome(
try {
for (const [nodeName, nodeRuns] of Object.entries(runData)) {
const lastRun = nodeRuns[nodeRuns.length - 1];
if (!lastRun?.data?.main) continue;
const branches = lastRun.data.main.map((items) => (items ?? []).map((item) => item.json));
// A sub-node records its run under the connection type that carried it,
// never `main`. Only the step run's own target is read that way: every
// other sub-node (a model, a memory) would otherwise put its whole
// exchange into the result of every ordinary run.
const outputs =
lastRun?.data?.[NodeConnectionTypes.Main] ??
(nodeName === subNodeTarget ? nonMainOutputs(lastRun) : undefined);
if (!outputs) continue;
const branches = outputs.map((items) => (items ?? []).map((item) => item.json));
const totalItems = branches.reduce((sum, items) => sum + items.length, 0);
if (totalItems === 0) continue;
if (branches.length === 1) {
@ -4496,7 +4685,10 @@ export async function extractExecutionOutcome(
: undefined,
executedNodeNames: executedNodeNames.length > 0 ? executedNodeNames : undefined,
nodeErrors: nodeErrors.length > 0 ? nodeErrors : undefined,
lastNodeExecuted: execution.data?.resultData?.lastNodeExecuted,
lastNodeExecuted: renameToolExecutor(
execution.data?.resultData?.lastNodeExecuted,
subNodeTarget,
),
workflowVersionId: execution.workflowVersionId,
error: errorMessage,
startedAt: execution.startedAt?.toISOString(),
@ -4748,41 +4940,72 @@ export async function extractNodeOutput(
await workflow?.expression.releaseIsolate();
}
// A sub-node (a model, a memory, a tool) records its run under the connection
// type that carried it, never `main`, and it records one run for each call
// its owner made. A step run on one is refused and the caller is sent here to
// read those calls, so all of them are read. A node in the main graph keeps
// to its last run: one run for each iteration of a loop is the common case
// there, and merging them would change what every caller already gets.
//
// The test is a non-main output on *some* run, not the absence of a main one
// on the last. A failed run carries no data at all — `createTaskData` sets
// none and only a rewired tool has it filled on the error path — so "no main
// output" also describes a node that failed on the last iteration of a loop.
// Reading every run of it would answer a question about the failed run with
// the output of earlier ones. A sub-node that failed on its last call is the
// mirror case: the calls that did return are the answer, and they are only
// reachable by looking past the last run.
const readsEveryRun = nodeRuns.some((run) => nonMainOutputs(run) !== undefined);
const runsRead = readsEveryRun ? nodeRuns : [lastRun];
const outputsPerRun = runsRead.map(
(run) => run?.data?.[NodeConnectionTypes.Main] ?? nonMainOutputs(run) ?? [],
);
const outputCount = Math.max(0, ...outputsPerRun.map((runOutputs) => runOutputs.length));
// One page over the items of all outputs (first output first), reported per
// output so a Filter's Kept and Discarded items never read as one list.
// Only the requested slice is materialized — avoids OOM on huge result sets.
// A label names the call when several were read, so two items that differ
// only by call are still told apart.
let index = 0;
let returnedCount = 0;
const outputs = (lastRun?.data?.main ?? []).map((output, outputIndex) => {
const items = output ?? [];
const firstInPage = Math.max(startIndex - index, 0);
const outputs: NodeOutputBranch[] = [];
for (let outputIndex = 0; outputIndex < outputCount; outputIndex++) {
const collected: unknown[] = [];
for (const item of items) {
if (index >= startIndex && returnedCount < maxItems) {
collected.push(item.json);
returnedCount++;
let itemsOnOutput = 0;
outputsPerRun.forEach((runOutputs, runIndex) => {
for (const item of runOutputs[outputIndex] ?? []) {
if (index >= startIndex && returnedCount < maxItems) {
const call = runsRead.length > 1 ? `[call ${runIndex + 1}]` : '';
collected.push(
wrapUntrustedData(
JSON.stringify(capItem(item.json), null, 2),
'execution-output',
`node:${nodeName}${call}[${outputIndex}][${itemsOnOutput}]`,
),
);
returnedCount++;
}
index++;
itemsOnOutput++;
}
index++;
}
return {
});
outputs.push({
index: outputIndex,
...(names[outputIndex] ? { name: names[outputIndex] } : {}),
totalItems: items.length,
items: collected.map((item, i) =>
wrapUntrustedData(
JSON.stringify(capItem(item), null, 2),
'execution-output',
`node:${nodeName}[${outputIndex}][${firstInPage + i}]`,
),
),
};
});
totalItems: itemsOnOutput,
items: collected,
});
}
return {
nodeName,
outputs,
totalItems: index,
returned: { from: startIndex, to: startIndex + returnedCount },
...(nodeRuns.length > 1 ? { totalRuns: nodeRuns.length } : {}),
};
}

View File

@ -321,6 +321,7 @@ export class JobProcessor {
dirtyNodeNames: manualData?.dirtyNodeNames,
triggerToStartFrom: manualData?.triggerToStartFrom,
userId: manualData?.userId,
agentRequest: manualData?.agentRequest,
};
try {

View File

@ -500,6 +500,9 @@ export class WorkflowExecutionService {
userId: data.userId,
dirtyNodeNames: data.dirtyNodeNames,
triggerToStartFrom: data.triggerToStartFrom,
// A run of a single tool node needs this to reach the worker, or the
// tool runs on empty arguments.
agentRequest: data.agentRequest,
source: data.source,
},
// Set this to null so `createRunExecutionData` doesn't initialize it.

View File

@ -1,3 +1,4 @@
import { TOOL_EXECUTOR_NODE_NAME } from '@n8n/constants';
import { type INode, NodeConnectionTypes } from 'n8n-workflow';
import { createNodeData } from './helpers';
@ -226,4 +227,40 @@ describe('rewireGraph()', () => {
expect(rewiredGraph.hasNode('trigger')).toBe(true);
expect(rewiredGraph.hasNode('leafTool')).toBe(true);
});
it('stands in for the agent, not for a node downstream of it', () => {
// trigger -> upstream -> agent -> downstream -> final, with a tool on the agent.
// The Tool Executor has to inherit the *agent's* main parents. Inheriting a
// downstream node's parents would put the agent and everything between it and
// that node back in the run.
const trigger = createNodeData({ name: 'trigger' });
const upstream = createNodeData({ name: 'upstream' });
const agent = createNodeData({ name: 'agent', type: 'n8n-nodes-base.ai-agent' });
const downstream = createNodeData({ name: 'downstream' });
const final = createNodeData({ name: 'final' });
const tool = createNodeData({ name: 'tool', type: 'n8n-nodes-base.ai-tool' });
const graph = new DirectedGraph();
graph.addNodes(trigger, upstream, agent, downstream, final, tool);
graph.addConnections(
{ from: trigger, to: upstream, type: NodeConnectionTypes.Main },
{ from: upstream, to: agent, type: NodeConnectionTypes.Main },
{ from: agent, to: downstream, type: NodeConnectionTypes.Main },
{ from: downstream, to: final, type: NodeConnectionTypes.Main },
{ from: tool, to: agent, type: NodeConnectionTypes.AiTool },
);
const rewiredGraph = rewireGraph(tool, graph);
const executorNode = rewiredGraph.getNodesByNames([TOOL_EXECUTOR_NODE_NAME]).values().next()
.value as INode;
const parents = rewiredGraph.getDirectParentConnections(executorNode);
expect(
parents.filter((cn) => cn.type === NodeConnectionTypes.Main).map((cn) => cn.from.name),
).toEqual(['upstream']);
expect(rewiredGraph.hasNode('agent')).toBe(false);
// The nodes below the agent lose their only parent, so nothing reaches them.
expect(rewiredGraph.getDirectParentConnections(downstream)).toEqual([]);
});
});

View File

@ -1,25 +1,61 @@
import { TOOL_EXECUTOR_NODE_NAME } from '@n8n/constants';
import * as a from 'assert/strict';
import { type AiAgentRequest, type INode, NodeConnectionTypes } from 'n8n-workflow';
import { type DirectedGraph } from './directed-graph';
/**
* The node that runs `tool` — the one the Tool Executor stands in for.
*
* A tool reaches its root through non-main connections only, so the walk keeps
* to those: following main connections as well would leave the root behind and
* land on a node *downstream* of it, whose main parents are the nodes the run
* is supposed to skip.
*
* The walk stops at the first node that has main parents, because those are the
* connections the Tool Executor inherits. A node without them cannot supply
* them — an Agent Tool between the tool and the top Agent, for one — so the walk
* carries on past it. When no node on the way has main parents, the farthest one
* is the root and the Tool Executor starts the run on its own.
*/
function findRootNode(graph: DirectedGraph, tool: INode): INode | undefined {
const seen = new Set<INode>([tool]);
const queue = [tool];
let farthest: INode | undefined;
while (queue.length > 0) {
const current = queue.shift() as INode;
for (const connection of graph.getDirectChildConnections(current)) {
if (connection.type === NodeConnectionTypes.Main) continue;
if (seen.has(connection.to)) continue;
seen.add(connection.to);
const hasMainParents = graph
.getDirectParentConnections(connection.to)
.some((cn) => cn.type === NodeConnectionTypes.Main);
if (hasMainParents) return connection.to;
farthest = connection.to;
queue.push(connection.to);
}
}
return farthest;
}
export function rewireGraph(
tool: INode,
graph: DirectedGraph,
agentRequest?: AiAgentRequest,
): DirectedGraph {
const modifiedGraph = graph.clone();
const children = modifiedGraph.getChildren(tool);
const rootNode = findRootNode(modifiedGraph, tool);
if (children.size === 0) {
// Nothing runs this tool, so there is no node to stand in for.
if (!rootNode) {
return graph;
}
const rootNode = [...children][children.size - 1];
a.ok(rootNode);
const allIncomingConnection = modifiedGraph
.getDirectParentConnections(rootNode)
.filter((cn) => cn.type === NodeConnectionTypes.Main);

View File

@ -81,6 +81,9 @@ export interface IRunExecutionDataV1 {
| 'evaluationRunId'
| 'source'
| 'suppressErrorWorkflow'
// A run of a single tool node carries the agent request that supplies the
// tool's arguments. Without it a worker runs the tool on empty arguments.
| 'agentRequest'
>;
/** Metadata about whether and how this execution's data was redacted. */