diff --git a/packages/@n8n/instance-ai/docs/tools.md b/packages/@n8n/instance-ai/docs/tools.md index 70015659f6f..4bdf7e9a278 100644 --- a/packages/@n8n/instance-ai/docs/tools.md +++ b/packages/@n8n/instance-ai/docs/tools.md @@ -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.` — 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. diff --git a/packages/@n8n/instance-ai/skills/debugging-executions/SKILL.md b/packages/@n8n/instance-ai/skills/debugging-executions/SKILL.md index ba6e170e476..04ab80722e1 100644 --- a/packages/@n8n/instance-ai/skills/debugging-executions/SKILL.md +++ b/packages/@n8n/instance-ai/skills/debugging-executions/SKILL.md @@ -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=, + 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.`. 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 diff --git a/packages/@n8n/instance-ai/src/tools/__tests__/executions.tool.test.ts b/packages/@n8n/instance-ai/src/tools/__tests__/executions.tool.test.ts index 9fdd2f67591..124c8af9dd3 100644 --- a/packages/@n8n/instance-ai/src/tools/__tests__/executions.tool.test.ts +++ b/packages/@n8n/instance-ai/src/tools/__tests__/executions.tool.test.ts @@ -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 }); diff --git a/packages/@n8n/instance-ai/src/tools/executions.tool.ts b/packages/@n8n/instance-ai/src/tools/executions.tool.ts index 4fa4d34c282..9ce3913c671 100644 --- a/packages/@n8n/instance-ai/src/tools/executions.tool.ts +++ b/packages/@n8n/instance-ai/src/tools/executions.tool.ts @@ -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, diff --git a/packages/@n8n/instance-ai/src/types.ts b/packages/@n8n/instance-ai/src/types.ts index 3bd5ecfbef8..22da4fedbda 100644 --- a/packages/@n8n/instance-ai/src/types.ts +++ b/packages/@n8n/instance-ai/src/types.ts @@ -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>; + /** + * 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; /** Run a past version's graph instead of the current draft. */ versionId?: string; timeout?: number; diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-step-run.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-step-run.test.ts index f299ceba64a..064f6eb2617 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai-step-run.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai-step-run.test.ts @@ -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 }>>): ITaskData { + return { + startTime: 0, + executionTime: 0, + executionIndex: 0, + source: [], + data: { [NodeConnectionTypes.Main]: outputs }, + }; +} + function taskData(items: Array<{ json: Record }>): 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[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.`. 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: {}, + }); + }); + }); +}); diff --git a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts index b1c3f12da0b..77aa03915f7 100644 --- a/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts +++ b/packages/cli/src/modules/instance-ai/__tests__/instance-ai.adapter.service.test.ts @@ -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; 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.`) and says + * "tool" nowhere, yet it describes an `ai_tool` output all the same. + */ +function runNodeTypesStub(): NodeTypes { + const nodeTypes = mock(); + 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, 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[10], mockWorkflowRunner as unknown as ConstructorParameters[11], {} as unknown as ConstructorParameters[12], - {} as unknown as ConstructorParameters[13], + (options?.nodeTypes ?? runNodeTypesStub()) as unknown as ConstructorParameters< + typeof InstanceAiAdapterService + >[13], { n8nFolder: '/tmp' } as unknown as ConstructorParameters[14], {} as unknown as ConstructorParameters[15], {} as unknown as ConstructorParameters[16], @@ -5520,6 +5781,7 @@ describe('createExecutionAdapter runStep()', () => { type StepOptions = { reuseExecutionId?: string; mockInput?: Array>; + toolArguments?: Record | 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; + + 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; + + // 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; + + 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; + + 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; + + // 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; + + 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) => ({ + 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; + + 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; + + 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; + + // 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.` 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; + + 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; + + // 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; + + 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', () => { diff --git a/packages/cli/src/modules/instance-ai/instance-ai-step-run.ts b/packages/cli/src/modules/instance-ai/instance-ai-step-run.ts index 5144c150ddb..880c9a20fdb 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai-step-run.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai-step-run.ts @@ -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, + nodeName: string, + outputIndex?: number, + seen: Set = 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, + nodeName: string, +): string[] { + const children = new Set(); + + 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(); + const seen = new Set([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([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(); - 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(); + const visited = new Set([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(); + 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, + nodeName: string, + avoid: Set, +): boolean { + const seen = new Set(); + 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(); + const seen = new Set(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([MCP_CLIENT_TOOL_NODE_TYPE]); + +/** + * The MCP registry saves a server as its own node type, `@n8n/mcp-registry.`, + * 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; +}): AiAgentRequest { + const { target } = args; + const toolArguments = args.toolArguments ?? {}; + + const names = new Set([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 diff --git a/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts b/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts index cbbac739699..81f0fe0c622 100644 --- a/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts +++ b/packages/cli/src/modules/instance-ai/instance-ai.adapter.service.ts @@ -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 } : {}), }; } diff --git a/packages/cli/src/scaling/job-processor.ts b/packages/cli/src/scaling/job-processor.ts index 8a0b81ec5cb..616a823ec96 100644 --- a/packages/cli/src/scaling/job-processor.ts +++ b/packages/cli/src/scaling/job-processor.ts @@ -321,6 +321,7 @@ export class JobProcessor { dirtyNodeNames: manualData?.dirtyNodeNames, triggerToStartFrom: manualData?.triggerToStartFrom, userId: manualData?.userId, + agentRequest: manualData?.agentRequest, }; try { diff --git a/packages/cli/src/workflows/workflow-execution.service.ts b/packages/cli/src/workflows/workflow-execution.service.ts index cc695e0ad32..af9b5add3b5 100644 --- a/packages/cli/src/workflows/workflow-execution.service.ts +++ b/packages/cli/src/workflows/workflow-execution.service.ts @@ -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. diff --git a/packages/core/src/execution-engine/partial-execution-utils/__tests__/rewire-graph.test.ts b/packages/core/src/execution-engine/partial-execution-utils/__tests__/rewire-graph.test.ts index f50c643cb68..c5ba8bc0077 100644 --- a/packages/core/src/execution-engine/partial-execution-utils/__tests__/rewire-graph.test.ts +++ b/packages/core/src/execution-engine/partial-execution-utils/__tests__/rewire-graph.test.ts @@ -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([]); + }); }); diff --git a/packages/core/src/execution-engine/partial-execution-utils/rewire-graph.ts b/packages/core/src/execution-engine/partial-execution-utils/rewire-graph.ts index 9e16dbb17f3..0e1fd234b97 100644 --- a/packages/core/src/execution-engine/partial-execution-utils/rewire-graph.ts +++ b/packages/core/src/execution-engine/partial-execution-utils/rewire-graph.ts @@ -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([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); diff --git a/packages/workflow/src/run-execution-data/run-execution-data.v1.ts b/packages/workflow/src/run-execution-data/run-execution-data.v1.ts index 0997f7ad2a4..e1bafdde233 100644 --- a/packages/workflow/src/run-execution-data/run-execution-data.v1.ts +++ b/packages/workflow/src/run-execution-data/run-execution-data.v1.ts @@ -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. */