github-mirror/n8n
1
0
Fork 0
mirror of https://github.com/n8n-io/n8n.git synced 2026-10-17 13:39:44 +02:00
Code Issues Actions 14 Packages Projects Releases Wiki Activity Graph

refactor(core): Simplify agent runtime structure (no-changelog) (#39507)

Browse Source
This commit is contained in:
Robin Braumann 2026-09-28 10:40:51 +00:00 committed by GitHub
parent a51434ff7d
commit 3584d15ddd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
25 changed files with 3335 additions and 2860 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

211
packages/@n8n/agents/src/runtime/__tests__/agent-runtime.test.ts
View File

@ -2604,6 +2604,36 @@ function makeGenerateWithToolCalls(
};
}
function makeStreamWithToolCalls(
toolCalls: Array<{ toolCallId: string; toolName: string; args: Record<string, unknown> }>,
) {
return {
stream: makeChunkStream([{ type: 'text-delta', id: 'text-1', text: 'calling tools...' }]),
finishReason: Promise.resolve('tool-calls'),
usage: Promise.resolve({ inputTokens: 10, outputTokens: 5, totalTokens: 15 }),
response: Promise.resolve({
messages: [
{
role: 'assistant',
content: toolCalls.map((tc) => ({
type: 'tool-call',
toolCallId: tc.toolCallId,
toolName: tc.toolName,
args: tc.args,
})),
},
],
}),
toolCalls: Promise.resolve(
toolCalls.map((tc) => ({
toolCallId: tc.toolCallId,
toolName: tc.toolName,
input: tc.args,
})),
),
};
}
describe('AgentRuntime — deferred tool loading', () => {
beforeEach(() => {
generateText.mockReset();
@ -5646,36 +5676,6 @@ describe('AgentRuntime — abort during a tool batch', () => {
{ toolCallId: 'tc-2', toolName: 'second_tool', args: {} },
];
function makeStreamWithToolCalls(
toolCalls: Array<{ toolCallId: string; toolName: string; args: Record<string, unknown> }>,
) {
return {
stream: makeChunkStream([{ type: 'text-delta', id: 'text-1', text: 'calling tools...' }]),
finishReason: Promise.resolve('tool-calls'),
usage: Promise.resolve({ inputTokens: 10, outputTokens: 5, totalTokens: 15 }),
response: Promise.resolve({
messages: [
{
role: 'assistant',
content: toolCalls.map((tc) => ({
type: 'tool-call',
toolCallId: tc.toolCallId,
toolName: tc.toolName,
args: tc.args,
})),
},
],
}),
toolCalls: Promise.resolve(
toolCalls.map((tc) => ({
toolCallId: tc.toolCallId,
toolName: tc.toolName,
input: tc.args,
})),
),
};
}
it('stream: resolves to "cancelled" (not "failed") and emits no Error event', async () => {
const bus = new AgentEventBus();
const { tools, secondHandler } = makeAbortingToolPair(bus);
@ -8499,30 +8499,58 @@ describe('AgentRuntime.resume() with createCancellation() — manual handling (h
});
// ---------------------------------------------------------------------------
// toModelOutput error resilience — processToolCall must never re-throw
// Tool output transform failures must not stop the agent loop.
// ---------------------------------------------------------------------------
describe('AgentRuntime — toModelOutput error resilience', () => {
describe.each([
{
name: 'toModelOutput',
transform: {
toModelOutput: () => {
throw new Error('Transform failed');
},
},
},
{
name: 'toMessage',
transform: {
toMessage: () => {
throw new Error('Transform failed');
},
},
},
{
name: 'async toMessage',
transform: {
toMessage: async () => await Promise.reject(new Error('Transform failed')),
},
},
])('AgentRuntime — $name error resilience', ({ transform }) => {
let events: AgentEventData[];
let eventBus: AgentEventBus;
beforeEach(() => {
vi.clearAllMocks();
generateText.mockReset();
streamText.mockReset();
events = [];
eventBus = new AgentEventBus();
eventBus.on(AgentEvent.ToolExecutionEnd, (event) => events.push(event));
});
function makeTransformErrorTool(name = 'transform_tool'): BuiltTool {
return {
name,
description: 'A tool whose toModelOutput always throws',
description: 'A tool whose output transform fails',
inputSchema: z.object({ value: z.string().optional() }),
handler: async () => await Promise.resolve({ raw: 'data' }),
toModelOutput: () => {
throw new Error('toModelOutput failed');
},
...transform,
};
}
function makeSuspendingTransformErrorTool(name = 'suspend_tool'): BuiltTool {
return {
name,
description: 'A suspending tool whose toModelOutput throws on the resumed call',
description: 'A tool whose output transform fails after resume',
inputSchema: z.object({ value: z.string().optional() }),
suspendSchema: z.object({ reason: z.string() }),
resumeSchema: z.object({ approved: z.boolean() }),
@ -8533,15 +8561,13 @@ describe('AgentRuntime — toModelOutput error resilience', () => {
}
return { done: true };
},
toModelOutput: () => {
throw new Error('toModelOutput failed on resume');
},
...transform,
};
}
it('generate(): toModelOutput throwing is treated as a tool error — loop continues', async () => {
it('generate(): records one failed tool completion and continues', async () => {
const tool = makeTransformErrorTool();
const { runtime } = createRuntimeWithTools([tool], 1);
const { runtime } = createRuntimeWithTools([tool], 1, eventBus);
generateText
.mockResolvedValueOnce(
@ -8555,34 +8581,17 @@ describe('AgentRuntime — toModelOutput error resilience', () => {
expect(result.finishReason).toBe('stop');
// Two LLM calls: tool-call turn + follow-up after error
expect(generateText).toHaveBeenCalledTimes(2);
expect(events).toEqual([expect.objectContaining({ toolCallId: 'tc-1', isError: true })]);
});
it('stream(): toModelOutput throwing surfaces as isError tool-result — loop continues', async () => {
it('stream(): records one failed tool completion and continues', async () => {
const tool = makeTransformErrorTool();
const { runtime } = createRuntimeWithTools([tool], 1);
const { runtime } = createRuntimeWithTools([tool], 1, eventBus);
streamText
.mockReturnValueOnce({
stream: makeChunkStream([{ type: 'text-delta', id: 'text-1', text: 'thinking...' }]),
finishReason: Promise.resolve('tool-calls'),
usage: Promise.resolve({ inputTokens: 10, outputTokens: 5, totalTokens: 15 }),
response: Promise.resolve({
messages: [
{
role: 'assistant',
content: [
{
type: 'tool-call',
toolCallId: 'tc-1',
toolName: 'transform_tool',
args: {},
},
],
},
],
}),
toolCalls: Promise.resolve([{ toolCallId: 'tc-1', toolName: 'transform_tool', input: {} }]),
})
.mockReturnValueOnce(
makeStreamWithToolCalls([{ toolCallId: 'tc-1', toolName: 'transform_tool', args: {} }]),
)
.mockReturnValueOnce(makeStreamSuccess('I see the tool errored'));
const { stream } = await runtime.stream('run the tool');
@ -8597,9 +8606,10 @@ describe('AgentRuntime — toModelOutput error resilience', () => {
(c) => c.type === 'tool-result' && c.toolCallId === 'tc-1',
) as (StreamChunk & { type: 'tool-result' }) | undefined;
expect(toolResultChunk?.isError).toBe(true);
expect(events).toEqual([expect.objectContaining({ toolCallId: 'tc-1', isError: true })]);
});
it('generate() resume: toModelOutput throwing in resumed tool call is captured — loop continues', async () => {
it('generate() resume: records the transform failure and continues', async () => {
const tool = makeSuspendingTransformErrorTool();
const { runtime } = createRuntimeWithTools([tool], 1);
@ -8616,9 +8626,7 @@ describe('AgentRuntime — toModelOutput error resilience', () => {
const { runId, toolCallId } = firstResult.pendingSuspend![0];
// Resume: tool returns a result but toModelOutput throws
// Bug: without fix this propagates out of iteratePendingToolCallsConcurrent and
// causes generate() to return with finishReason 'error' instead of 'stop'.
// The tool returns a result, but its output transform fails.
const resumeResult = await runtime.resume(
'generate',
{ approved: true },
@ -8633,32 +8641,14 @@ describe('AgentRuntime — toModelOutput error resilience', () => {
expect(generateText).toHaveBeenCalledTimes(2);
});
it('stream() resume: toModelOutput throwing in resumed tool call does not close the stream with error', async () => {
it('stream() resume: records the transform failure and continues', async () => {
const tool = makeSuspendingTransformErrorTool();
const { runtime } = createRuntimeWithTools([tool], 1);
// First stream: agent calls the tool and suspends
streamText.mockReturnValueOnce({
stream: makeChunkStream([{ type: 'text-delta', id: 'text-1', text: 'thinking...' }]),
finishReason: Promise.resolve('tool-calls'),
usage: Promise.resolve({ inputTokens: 10, outputTokens: 5, totalTokens: 15 }),
response: Promise.resolve({
messages: [
{
role: 'assistant',
content: [
{
type: 'tool-call',
toolCallId: 'tc-1',
toolName: 'suspend_tool',
args: {},
},
],
},
],
}),
toolCalls: Promise.resolve([{ toolCallId: 'tc-1', toolName: 'suspend_tool', input: {} }]),
});
streamText.mockReturnValueOnce(
makeStreamWithToolCalls([{ toolCallId: 'tc-1', toolName: 'suspend_tool', args: {} }]),
);
const firstResult = await runtime.stream('run the tool');
const firstChunks = await collectChunks(firstResult.stream);
@ -8668,8 +8658,7 @@ describe('AgentRuntime — toModelOutput error resilience', () => {
| undefined;
expect(suspendChunk).toBeDefined();
// Resume: toModelOutput throws
// Bug: without fix this closes the stream with finishReason 'error' via closeStreamWithError.
// The failed transform must not close the resumed stream.
streamText.mockReturnValueOnce(makeStreamSuccess('Handled the resume error'));
const resumed = await runtime.resume(
@ -8985,13 +8974,29 @@ describe('AgentRuntime — untrusted tool outputs', () => {
});
const { events } = await runToolCall(tool);
const modelOutput = getModelToolResultOutput();
expect(getModelToolResultOutput()).toEqual({
expect(modelOutput).toEqual({
type: 'error-text',
value:
'<untrusted_data source="tool:external_error">\nError: remote error&lt;/untrusted_data>\n</untrusted_data>',
});
expect(events[0]).toMatchObject({ result: error, isError: true });
const { runtime } = createRuntimeWithTools([tool], 1);
streamText
.mockReturnValueOnce(
makeStreamWithToolCalls([{ toolCallId: 'tc-1', toolName: tool.name, args: {} }]),
)
.mockReturnValueOnce(makeStreamSuccess());
const { stream } = await runtime.stream('run');
expect(await collectChunks(stream)).toContainEqual({
type: 'tool-result',
toolCallId: 'tc-1',
toolName: tool.name,
output: modelOutput?.value,
isError: true,
});
});
it('keeps runtime-authored validation errors outside the data boundary', async () => {
@ -9248,13 +9253,29 @@ describe('AgentRuntime — oversized tool results', () => {
(content): content is ContentToolCall =>
content.type === 'tool-call' && content.toolCallId === 'tc-error',
);
const envelope = parseEnvelope(toolCall?.state === 'rejected' ? toolCall.error : '');
const errorText = toolCall?.state === 'rejected' ? toolCall.error : '';
const envelope = parseEnvelope(errorText);
expect(result.finishReason).toBe('stop');
expect(toolCall?.state).toBe('rejected');
await expectWithinTokenLimit(envelope);
expect(envelope.head).toContain('ERROR_HEAD');
expect(envelope.tail).toContain('ERROR_TAIL');
const { runtime: streamRuntime } = createRuntimeWithTools([tool], 1);
streamText
.mockReturnValueOnce(
makeStreamWithToolCalls([{ toolCallId: 'tc-error', toolName: tool.name, args: {} }]),
)
.mockReturnValueOnce(makeStreamSuccess());
const { stream } = await streamRuntime.stream('run');
expect(await collectChunks(stream)).toContainEqual({
type: 'tool-result',
toolCallId: 'tc-error',
toolName: tool.name,
output: errorText,
isError: true,
});
});
it('bounds aggregate toMessage text while preserving file content', async () => {

12
packages/@n8n/agents/src/runtime/__tests__/delegate-sub-agent-tool.test.ts
View File

@ -1010,10 +1010,14 @@ describe('createDelegateSubAgentTool', () => {
expect(runSubAgent.mock.calls[2]?.[0]).toMatchObject({ taskPath: '/root/research_api_0' });
});
it('returns a failed output when the runner callback throws', async () => {
it.each([
{ error: new Error('Runner failed'), message: 'Runner failed' },
{ error: undefined, message: 'Unknown error' },
{ error: Symbol('failure'), message: 'Unknown error' },
])('reports the runner error: $error', async ({ error, message }) => {
const events: AgentEventData[] = [];
const tool = createDelegateSubAgentTool({
runSubAgent: async () => await Promise.reject(new Error('Runner failed')),
runSubAgent: vi.fn<DelegateSubAgentRunner>().mockRejectedValue(error),
});
await expect(
@ -1025,12 +1029,12 @@ describe('createDelegateSubAgentTool', () => {
status: 'failed',
taskPath: '/root/research_api_0',
answer: '',
error: 'Runner failed',
error: message,
});
expect(events[events.length - 1]).toMatchObject({
type: AgentEvent.SubAgentCompleted,
status: 'failed',
error: 'Runner failed',
error: message,
});
});

1091
packages/@n8n/agents/src/runtime/loop/agent-runtime.ts
View File

File diff suppressed because it is too large Load Diff

94
packages/@n8n/agents/src/runtime/loop/checkpoint-data.ts Normal file
View File

@ -0,0 +1,94 @@
import { isCancellation } from '../../sdk/cancellation';
import type { BuiltTool, PendingToolCall, SerializableAgentState } from '../../types';
import type {
AgentPersistenceOptions,
ExecutionOptions,
PersistedExecutionOptions,
ResumeOptions,
} from '../../types/sdk/agent';
import { parseWithSchema } from '../../utils/parse';
import type { AgentMessageList } from '../model/message-list';
export async function parseResumeData(
data: unknown,
resumeSchema: BuiltTool['resumeSchema'],
): Promise<unknown> {
if (isCancellation(data) || !resumeSchema) return data;
const result = await parseWithSchema(resumeSchema, data, { stripUnknown: true });
if (!result.success) throw new Error(`Invalid resume payload: ${result.error}`);
return result.data;
}
export function mergeResumeExecutionOptions(
state: SerializableAgentState,
callerExecOptions: ExecutionOptions,
): ExecutionOptions & { iterationCount?: number } {
const persisted = state.executionOptions ?? {};
const persistedMaxIterations = persisted.maxIterations;
const callerMaxIterations = callerExecOptions.maxIterations;
if (
callerMaxIterations !== undefined &&
persistedMaxIterations !== undefined &&
callerMaxIterations < persistedMaxIterations
) {
throw new Error(
`Cannot decrease maxIterations when resuming a run. Expected >= ${persistedMaxIterations}, received ${callerMaxIterations}.`,
);
}
const mergedMaxIterations = callerMaxIterations ?? persistedMaxIterations;
return {
...callerExecOptions,
...(mergedMaxIterations !== undefined ? { maxIterations: mergedMaxIterations } : {}),
...(state.iterationCount !== undefined ? { iterationCount: state.iterationCount } : {}),
};
}
export function mergeResumePersistence(
persistence: AgentPersistenceOptions | undefined,
hostMetadata: ResumeOptions['hostMetadata'],
): AgentPersistenceOptions | undefined {
if (!persistence) return undefined;
const merged = { ...persistence };
if (persistence.hostMetadata || hostMetadata) {
merged.hostMetadata = { ...persistence.hostMetadata, ...hostMetadata };
}
return merged;
}
export function markSuspendedToolCalls(
list: AgentMessageList,
pendingToolCalls: Record<string, PendingToolCall>,
): void {
// Record what confirmation each suspended call showed the user, so an
// abandoned suspension can be settled with that context on a later
// history load instead of vanishing from the transcript.
for (const pending of Object.values(pendingToolCalls)) {
if (!pending.suspended) continue;
const payload =
typeof pending.suspendPayload === 'object' && pending.suspendPayload !== null
? (pending.suspendPayload as { message?: unknown; requestId?: unknown })
: undefined;
list.markToolCallSuspended(pending.toolCallId, {
...(typeof payload?.message === 'string' ? { message: payload.message } : {}),
...(typeof payload?.requestId === 'string' ? { requestId: payload.requestId } : {}),
});
}
}
export function buildCheckpointOptions(
options: (ExecutionOptions & { iterationCount?: number }) | undefined,
maxIterations?: number,
iterationCount?: number,
): Pick<SerializableAgentState, 'executionOptions' | 'iterationCount'> {
// Persist loop controls only. providerOptions are intentionally excluded
// because they may contain sensitive data (API keys, auth headers).
const resolvedMaxIterations = maxIterations ?? options?.maxIterations;
const resolvedIterationCount = iterationCount ?? options?.iterationCount;
const executionOptions: PersistedExecutionOptions | undefined =
resolvedMaxIterations !== undefined ? { maxIterations: resolvedMaxIterations } : undefined;
return {
executionOptions,
...(resolvedIterationCount !== undefined ? { iterationCount: resolvedIterationCount } : {}),
};
}

6
packages/@n8n/agents/src/runtime/loop/execution-counter.ts
View File

@ -51,9 +51,5 @@ export function incrementTokenCountFromUsage(
usage.totalTokens ?? usage.tokens ?? (usage.inputTokens ?? 0) + (usage.outputTokens ?? 0);
if (tokenCount <= 0) return;
try {
counter.incrementTokenCount(tokenCount);
} catch {
// Aggregate counters are best-effort and must never affect agent execution.
}
recordExecutionCounter(() => counter.incrementTokenCount(tokenCount));
}

12
packages/@n8n/agents/src/runtime/loop/generate-sink.ts
View File

@ -1,3 +1,4 @@
import { finalizeRun } from './run-output-sink';
import type {
CompleteEmission,
ModelCallContext,
@ -5,7 +6,7 @@ import type {
RunOutputSink,
RunServices,
SuspendEmission,
} from './run-output-sink';
} from '../../types/runtime/agent-loop';
import { classifyModelTurnError } from './runtime-helpers';
import type { GenerateResult } from '../../types';
import type { ToolResultEntry } from '../../types/sdk/agent';
@ -13,7 +14,7 @@ import { isAttachmentValidationError } from '../model/attachment-validation-erro
import { loadAi } from '../model/lazy-ai';
import { fromAiFinishReason, fromAiMessages } from '../model/messages';
import { toTokenUsage } from '../streaming/stream';
import type { ToolCallBatchResult } from '../tools/tool-call-executor';
import type { ToolCallBatchResult } from '../../types/runtime/tool-execution';
/**
* Non-streaming output sink: drives the loop with `generateText`, accumulates a
@ -90,11 +91,8 @@ export class GenerateSink implements RunOutputSink<GenerateResult> {
}
async finishComplete(emission: CompleteEmission): Promise<GenerateResult> {
const { list, options, finishReason, usage, structuredOutput } = emission;
await this.services.saveToMemory(list, options);
await this.services.maybeGenerateTitle(list, options);
await this.services.cleanupRun();
await this.services.flushTelemetry(options);
const { list, finishReason, usage, structuredOutput } = emission;
await finalizeRun(this.services, emission);
return {
runId: this.services.runId,

159
packages/@n8n/agents/src/runtime/loop/run-output-sink.ts
View File

@ -1,142 +1,23 @@
import type {
LanguageModel,
LanguageModelCallOptions,
ModelMessage,
Output,
SystemModelMessage,
TelemetryOptions,
ToolCallRepairFunction,
ToolSet,
} from 'ai';
import type { CompleteEmission, RunServices } from '../../types/runtime/agent-loop';
import type { FinishReason, SerializableAgentState, TokenUsage } from '../../types';
import type { ExecutionOptions, RunOptions } from '../../types/sdk/agent';
import type { AgentMessage } from '../../types/sdk/message';
import type { JSONObject } from '../../types/utils/json';
import type { AgentMessageList } from '../model/message-list';
import type { ToolCallBatchResult, ToolCallSuspension } from '../tools/tool-call-executor';
export type {
ModelTurnErrorType,
ModelTurnError,
ModelTurnResult,
ModelCallContext,
SuspendEmission,
CompleteEmission,
RunServices,
RunOutputSink,
} from '../../types/runtime/agent-loop';
type RunCallOptions = (RunOptions & ExecutionOptions) | undefined;
/**
* Recognized ways a model turn can fail while the provider still returns
* "successfully" (no thrown SDK error). Extend this as new failure modes are
* recognized so consumers switch on `type` instead of probing separate fields.
* - `prompt_blocked`: the provider rejected the prompt (e.g. a safety block).
* - `no_output`: the model returned nothing with a non-stop finish reason.
*/
export type ModelTurnErrorType = 'prompt_blocked' | 'no_output';
/** A recognized model-turn failure: a `type` to branch on and a user-facing `message`. */
export interface ModelTurnError {
type: ModelTurnErrorType;
message: string;
}
/** Normalized result of a single LLM turn, independent of generate vs stream. */
export interface ModelTurnResult {
/** Raw AI SDK finish reason (used to detect the `tool-calls` continuation). */
aiFinishReason: string;
finishReason: FinishReason;
/** Normalized usage, including any cache-token details available from the provider. */
usage: TokenUsage | undefined;
newMessages: AgentMessage[];
toolCalls: Array<{
toolCallId: string;
toolName: string;
input: unknown;
providerExecuted?: boolean;
}>;
/** Resolved structured output, only when an output spec is set and the turn finished. */
structuredOutput: unknown;
/**
* Set when the turn failed in a recognized way despite the provider returning
* without a thrown error (e.g. a prompt safety block, or empty output). The
* loop surfaces this instead of ending the run silently. Check this single
* field rather than probing per-failure-mode properties.
*/
errorReason?: ModelTurnError;
}
/** Per-iteration inputs for the LLM call, assembled by the shared loop. */
export interface ModelCallContext {
model: LanguageModel;
system: SystemModelMessage | SystemModelMessage[];
messages: ModelMessage[];
abortSignal: AbortSignal;
hasTools: boolean;
aiTools: ToolSet;
reasoning?: LanguageModelCallOptions['reasoning'];
providerOptions?: Record<string, JSONObject>;
outputSpec?: ReturnType<typeof Output.object>;
maxOutputTokens?: number;
onInputRejected?: (error: unknown) => Promise<void>;
aiSdkOptions: {
telemetry?: TelemetryOptions;
repairToolCall?: ToolCallRepairFunction<NoInfer<ToolSet>>;
};
}
/** Data needed to emit a terminal suspension result. */
export interface SuspendEmission {
suspendRunId: string;
list: AgentMessageList;
usage: TokenUsage | undefined;
suspensions: ToolCallSuspension[];
}
/** Data needed to emit a terminal completion result. */
export interface CompleteEmission {
list: AgentMessageList;
options: RunCallOptions;
finishReason: FinishReason;
usage: TokenUsage | undefined;
structuredOutput: unknown;
}
/**
* Runtime services the sinks call into for shared terminal concerns (cost,
* memory persistence, title generation, telemetry flush, state). Implemented by
* `AgentRuntime` and passed to each sink.
*/
export interface RunServices {
readonly runId: string;
readonly modelId: string;
applyCost(usage: TokenUsage | undefined): TokenUsage | undefined;
saveToMemory(list: AgentMessageList, options: RunCallOptions): Promise<void>;
maybeGenerateTitle(list: AgentMessageList, options: RunCallOptions): Promise<void>;
flushTelemetry(options: RunCallOptions): Promise<void>;
cleanupRun(): Promise<void>;
updateState(patch: Partial<SerializableAgentState>): void;
emitAgentEnd(messages: AgentMessage[]): void;
getState(): SerializableAgentState;
}
/**
* Strategy that adapts the shared agentic loop to a concrete output channel:
* `GenerateSink` accumulates a `GenerateResult`; `StreamSink` writes
* `StreamChunk`s through the stream writer guard.
*/
export interface RunOutputSink<TResult> {
/** Run one LLM turn. Streaming implementations also emit text/tool chunks here. */
callModel(ctx: ModelCallContext): Promise<ModelTurnResult>;
/**
* Report the run's cumulative token usage after each turn. Streaming
* implementations keep it so an aborted run can still emit a terminal
* finish chunk carrying the tokens consumed before the stop.
*/
reportUsage(usage: TokenUsage | undefined): void;
/**
* Signal that the just-returned turn's messages have been folded into the list.
* Streaming implementations drop the retained streamed text here so a later abort
* can't duplicate a turn that is already in the list — while a stop landing before
* this call still recovers the turn from the retained text.
*/
onTurnFolded?(): void;
/** Emit the results/errors of a completed tool-call batch. */
emitToolBatch(batch: ToolCallBatchResult): Promise<void>;
/** Produce the terminal result when the run suspends for HITL / suspend-resume. */
finishSuspended(emission: SuspendEmission): Promise<TResult>;
/** Produce the terminal result when the run completes normally. */
finishComplete(emission: CompleteEmission): Promise<TResult>;
/** Persist the turn before checkpoint cleanup and telemetry flush. */
export async function finalizeRun(
services: RunServices,
{ list, options }: Pick<CompleteEmission, 'list' | 'options'>,
): Promise<void> {
await services.saveToMemory(list, options);
await services.maybeGenerateTitle(list, options);
await services.cleanupRun();
await services.flushTelemetry(options);
}

116
packages/@n8n/agents/src/runtime/loop/runtime-context.ts
View File

@ -2,7 +2,7 @@ import type { ProviderOptions } from '@ai-sdk/provider-utils';
import { getProviderPrefix } from '@n8n/ai-utilities/agent-config';
import type { LanguageModel, Output } from 'ai';
import type { AgentRuntimeConfig } from './agent-runtime';
import type { AgentRuntimeConfig } from '../../types/runtime/agent-runtime';
import { UNTRUSTED_OUTPUT_DOCTRINE } from '../../sdk/untrusted-content';
import type { AgentExecutionCounter, BuiltTool, JSONObject } from '../../types';
import type { AgentPersistenceOptions, ExecutionOptions } from '../../types/sdk/agent';
@ -23,13 +23,19 @@ import {
} from '../memory/episodic-memory-capture';
import { loadAi } from '../model/lazy-ai';
import type { AgentMessageList } from '../model/message-list';
import { createModel } from '../model/model-factory';
import { buildCallPromptCacheOptions, mergeProviderOptions } from '../model/prompt-cache';
import { createModel, supportsSplitSystemMessages } from '../model/model-factory';
import {
applyRuntimeCacheBreakpoints,
buildCallPromptCacheOptions,
buildInstructionPromptCacheOptions,
mergeProviderOptions,
} from '../model/prompt-cache';
import {
getProviderQuirks,
PROVIDER_QUIRKS,
resolveDefaultMaxOutputTokens,
} from '../model/provider-quirks';
import type { ActiveSkills } from '../skills/active-skills';
import type { DeferredToolManager } from '../tools/deferred-tool-manager';
import { buildToolMap, toAiSdkProviderTools, toAiSdkTools } from '../tools/tool-adapter';
@ -39,6 +45,13 @@ function wrapBuiltInRules(fragments: string[]): string | undefined {
return `<built_in_rules>\n${fragments.map((f) => `- ${f}`).join('\n')}\n</built_in_rules>`;
}
function prependBuiltInRules(fragments: string[], instructions: string): string {
const rules = wrapBuiltInRules(fragments);
if (!rules) return instructions;
if (!instructions) return rules;
return `${rules}\n\n${instructions}`;
}
export interface StaticLoopContext {
model: LanguageModel;
aiProviderTools: ReturnType<typeof toAiSdkProviderTools>;
@ -68,7 +81,7 @@ export class RuntimeContextBuilder {
buildStaticLoopContext(
execOptions?: ExecutionOptions & { persistence?: AgentPersistenceOptions },
): StaticLoopContext {
const { Output, jsonSchema } = loadAi();
const ai = loadAi();
const aiProviderTools = toAiSdkProviderTools(this.config.providerTools);
const model = createModel(this.config.model, this.config.modelFetch);
const outputSchema = this.config.structuredOutput;
@ -78,16 +91,7 @@ export class RuntimeContextBuilder {
isRawJsonSchemaOutput,
);
const outputSpec = outputSchema
? Output.object({
// Zod schemas pass through directly; a raw JSON Schema gets
// `additionalProperties: false` locked onto every object (required by Anthropic)
// and wrapped with the AI SDK's `jsonSchema()` helper.
schema: isZodSchema(outputSchema)
? outputSchema
: jsonSchema(lockAdditionalProperties(outputSchema)),
})
: undefined;
const outputSpec = this.buildOutputSpec(outputSchema, ai);
return {
model,
@ -99,6 +103,26 @@ export class RuntimeContextBuilder {
};
}
private buildOutputSpec(
outputSchema: AgentRuntimeConfig['structuredOutput'],
{ Output, jsonSchema }: ReturnType<typeof loadAi>,
): StaticLoopContext['outputSpec'] {
if (!outputSchema) return undefined;
// Anthropic requires additionalProperties: false on each raw schema object.
const schema = isZodSchema(outputSchema)
? outputSchema
: jsonSchema(lockAdditionalProperties(outputSchema));
return Output.object({ schema });
}
buildInstructionProviderOptions(): ProviderOptions | undefined {
// Explicit instruction options take precedence over cache defaults.
return mergeProviderOptions(
buildInstructionPromptCacheOptions(this.config.promptCaching, this.modelId),
this.config.instructionProviderOptions,
);
}
/** Build the current local tool view; deferred loads can change this between iterations. */
buildToolLoopContext(
aiProviderTools: ReturnType<typeof toAiSdkProviderTools>,
@ -124,6 +148,44 @@ export class RuntimeContextBuilder {
};
}
buildModelPrompt({
list,
tools,
instructionProviderOptions,
hostVolatileInstructions,
activeSkills,
}: {
list: AgentMessageList;
tools: ReturnType<RuntimeContextBuilder['buildToolLoopContext']>;
instructionProviderOptions: ProviderOptions | undefined;
hostVolatileInstructions: string | undefined;
activeSkills: ActiveSkills | undefined;
}) {
const combinedVolatileInstructions = [tools.volatileInstructions, hostVolatileInstructions]
.map((value) => value?.trim())
.filter((value): value is string => Boolean(value))
.join('\n\n');
const { system, messages } = list.forLlm(
// Skill content changes only on activation. Keep it cached when memory compacts.
[tools.effectiveInstructions, activeSkills?.instructions()]
.filter(Boolean)
.join('\n\n'),
instructionProviderOptions,
combinedVolatileInstructions || undefined,
supportsSplitSystemMessages(this.config.model),
);
// Cache breakpoints apply to this call only. Do not change stored messages or tools.
const cached = applyRuntimeCacheBreakpoints({
system,
messages: activeSkills?.modelMessages(messages, list) ?? messages,
aiTools: tools.aiTools,
promptCaching: this.config.promptCaching,
modelId: this.modelId,
staticToolCacheName: tools.staticToolCacheName,
});
return { system, ...cached };
}
/**
* Name of the tool eligible for an Anthropic tool-definitions cache
* breakpoint, or `undefined` if the tool set isn't fully static. Deferred
@ -260,6 +322,14 @@ export class RuntimeContextBuilder {
instructions: string;
volatileInstructions: string | undefined;
} {
const { stableFragments, volatileFragments } = this.collectInstructionFragments(tools);
return {
instructions: prependBuiltInRules(stableFragments, this.config.instructions),
volatileInstructions: wrapBuiltInRules(volatileFragments),
};
}
private collectInstructionFragments(tools: BuiltTool[]) {
const loadedToolNames = new Set(
this.deferredToolManager?.getLoadedTools().map((t) => t.name) ?? [],
);
@ -272,9 +342,8 @@ export class RuntimeContextBuilder {
) {
continue;
}
(loadedToolNames.has(tool.name) ? volatileFragments : stableFragments).push(
tool.systemInstruction,
);
const fragments = loadedToolNames.has(tool.name) ? volatileFragments : stableFragments;
fragments.push(tool.systemInstruction);
}
// Define the untrusted-data boundary ahead of the first wrapped result.
@ -288,18 +357,7 @@ export class RuntimeContextBuilder {
target.unshift(UNTRUSTED_OUTPUT_DOCTRINE);
}
const userInstructions = this.config.instructions;
const stableBlock = wrapBuiltInRules(stableFragments);
const instructions = stableBlock
? userInstructions
? `${stableBlock}\n\n${userInstructions}`
: stableBlock
: userInstructions;
return {
instructions,
volatileInstructions: wrapBuiltInRules(volatileFragments),
};
return { stableFragments, volatileFragments };
}
/** Build the providerOptions object for provider-specific thinking config. */

2
packages/@n8n/agents/src/runtime/loop/runtime-helpers.ts
View File

@ -2,7 +2,7 @@
* Pure utility functions used by AgentRuntime that require no class context.
* These are extracted here to keep agent-runtime.ts focused on orchestration logic.
*/
import type { ModelTurnError } from './run-output-sink';
import type { ModelTurnError } from '../../types/runtime/agent-loop';
import { stripInvisibleUnicode, wrapUntrustedData } from '../../sdk/untrusted-content';
import type { StreamChunk, TokenUsage, McpConnectionFailedEvent } from '../../types';
import type { AgentMessage, ContentToolCall } from '../../types/sdk/message';

226
packages/@n8n/agents/src/runtime/loop/stream-sink.ts
View File

@ -1,5 +1,6 @@
import type { StreamTextTransform, ToolSet } from 'ai';
import type { StreamTextTransform, TextStreamPart, ToolSet } from 'ai';
import { finalizeRun } from './run-output-sink';
import type {
CompleteEmission,
ModelCallContext,
@ -7,7 +8,7 @@ import type {
RunOutputSink,
RunServices,
SuspendEmission,
} from './run-output-sink';
} from '../../types/runtime/agent-loop';
import { classifyModelTurnError, mergeUsage } from './runtime-helpers';
import type { ExecutionOptions, TokenUsage } from '../../types/sdk/agent';
import type { AgentMessage } from '../../types/sdk/message';
@ -27,7 +28,7 @@ import {
withChunkIdleTimeout,
} from '../streaming/stream-stall';
import type { StreamWriterGuard } from '../streaming/stream-writer-guard';
import type { ToolCallBatchResult } from '../tools/tool-call-executor';
import type { ToolCallBatchResult } from '../../types/runtime/tool-execution';
/**
* Chunk types that are pure transport bookkeeping: an attempt that stalled
@ -160,7 +161,7 @@ export class StreamSink implements RunOutputSink<void> {
return transforms.length > 0 ? { experimental_transform: transforms } : {};
}
async callModel(ctx: ModelCallContext): Promise<ModelTurnResult> {
private getStreamDeadlines(): { idleMs: number; firstOutputMs: number } {
// Clamped to MAX_MODEL_STREAM_TIMEOUT_MS: above it setTimeout fires after
// 1ms, turning an "effectively disable" override into instant stalls.
const idleMs = Math.min(
@ -177,48 +178,55 @@ export class StreamSink implements RunOutputSink<void> {
),
MAX_MODEL_STREAM_TIMEOUT_MS,
);
return { idleMs, firstOutputMs };
}
async callModel(ctx: ModelCallContext): Promise<ModelTurnResult> {
const deadlines = this.getStreamDeadlines();
for (let attempt = 0; ; attempt++) {
// Opt-in: only build the reader when the host bills stopped runs, and only
// for providers with a reader. Re-created per attempt so a retried turn
// starts from a clean raw capture.
this.rawUsageReader = this.options?.recoverUsageOnAbort
? createRawUsageReader(this.services.modelId)
: undefined;
// Some providers report failures (e.g. prompt safety blocks) only on the
// raw stream — no error, no content — so raw chunks are required to
// explain an otherwise silent empty response.
this.rawErrorReader = createRawErrorReader(this.services.modelId);
this.resetRawReaders();
// Per-attempt controller: a stall must be able to cancel this attempt's
// fetch (releasing the socket the 1h network timeout would otherwise
// hold) without touching the run-level signal.
const turnAbort = new AbortController();
const attemptState = { streamedContent: false };
try {
return await this.streamModelTurn(ctx, turnAbort, { idleMs, firstOutputMs }, attemptState);
return await this.streamModelTurn(ctx, turnAbort, deadlines, attemptState);
} catch (error) {
if (isAttachmentValidationError(error)) {
if (!attemptState.streamedContent) await ctx.onInputRejected?.(error);
throw error;
}
// A stalled or empty stream before any content is invisible to the user
// (and to the host's persistence) — re-issue the request instead of
// failing the run for what is usually a dead connection at request time.
// Once content has streamed, a retry would duplicate segments and orphan
// already-persisted tool-call facts, so the error surfaces instead.
// The abandoned attempt's prompt processing goes unbilled — accepted:
// it only has usage at all when the stream died between message_start
// and the first content chunk.
const retryable =
(error instanceof ModelStreamStallError ||
loadAi().NoOutputGeneratedError.isInstance(error)) &&
attempt < MAX_MODEL_STREAM_STALL_RETRIES &&
!attemptState.streamedContent &&
!ctx.abortSignal.aborted;
if (!retryable) throw error;
if (!this.canRetryStream(error, attempt, attemptState, ctx.abortSignal)) throw error;
}
}
}
private resetRawReaders(): void {
// Each attempt needs fresh readers so a retry does not reuse captured usage.
this.rawUsageReader = this.options?.recoverUsageOnAbort
? createRawUsageReader(this.services.modelId)
: undefined;
// Some providers report a blocked prompt only in raw chunks.
this.rawErrorReader = createRawErrorReader(this.services.modelId);
}
private canRetryStream(
error: unknown,
attempt: number,
attemptState: { streamedContent: boolean },
abortSignal: AbortSignal,
): boolean {
// Retry only before content is visible or persisted. The abandoned attempt stays unbilled.
return (
(error instanceof ModelStreamStallError ||
loadAi().NoOutputGeneratedError.isInstance(error)) &&
attempt < MAX_MODEL_STREAM_STALL_RETRIES &&
!attemptState.streamedContent &&
!abortSignal.aborted
);
}
private async streamModelTurn(
ctx: ModelCallContext,
turnAbort: AbortController,
@ -227,6 +235,36 @@ export class StreamSink implements RunOutputSink<void> {
): Promise<ModelTurnResult> {
const { idleMs, firstOutputMs } = deadlines;
const { NoOutputGeneratedError, streamText } = loadAi();
const { result, activity } = this.startModelStream(ctx, turnAbort, idleMs, streamText);
// Use the longer deadline before the first output to allow prompt processing.
const chunkStream =
idleMs > 0
? withChunkIdleTimeout(
result.stream,
() => (attemptState.streamedContent ? idleMs : firstOutputMs),
() => turnAbort.abort(),
() => activity.lastAt,
)
: result.stream;
for await (const chunk of chunkStream) {
if (chunk.type === 'error' && isAttachmentValidationError(chunk.error)) throw chunk.error;
// Result promises report this error. Do not forward it or count it as content before a retry.
if (chunk.type === 'error' && NoOutputGeneratedError.isInstance(chunk.error)) continue;
if (!this.captureStreamChunk(chunk, attemptState)) continue;
await this.writeStreamChunk(chunk);
}
return await this.resolveModelTurn(result, ctx, idleMs, turnAbort);
}
private startModelStream(
ctx: ModelCallContext,
turnAbort: AbortController,
idleMs: number,
streamText: ReturnType<typeof loadAi>['streamText'],
) {
// Raw chunks serve two consumers: the usage/error readers parse them, and
// the stall watchdog counts them as liveness — keepalive events the SDK
// otherwise drops (e.g. Anthropic `ping`, a gateway's empty-delta frames)
@ -264,83 +302,58 @@ export class StreamSink implements RunOutputSink<void> {
: undefined,
),
});
return { result, activity };
}
// A healthy streaming response emits chunks continuously (raw provider
// events included), so prolonged silence means the connection or provider
// is wedged — fail the turn instead of hanging until the network timeout.
// Pre-first-output silence gets the longer deadline (prompt processing on
// large cache-miss prompts sends nothing for minutes); once content has
// streamed, the tighter idle limit applies.
const chunkStream =
idleMs > 0
? withChunkIdleTimeout(
result.stream,
() => (attemptState.streamedContent ? idleMs : firstOutputMs),
() => turnAbort.abort(),
() => activity.lastAt,
)
: result.stream;
private captureStreamChunk(
chunk: TextStreamPart<ToolSet>,
attemptState: { streamedContent: boolean },
): boolean {
// Capture raw chunks here when a stream does not use the tap, such as a test stream.
if (chunk.type === 'raw') {
this.rawUsageReader?.capture(chunk.rawValue);
this.rawErrorReader?.capture(chunk.rawValue);
return false;
}
if (!STALL_RETRY_SAFE_CHUNK_TYPES.has(chunk.type)) attemptState.streamedContent = true;
// The runtime writes its own finish chunk after the loop completes.
if (chunk.type === 'finish') return false;
// Retain partial text so cancellation can save an unfinished response.
if (chunk.type === 'text-delta') this.partialText += chunk.text ?? '';
return true;
}
// Consume the stream. When the AbortSignal fires mid-stream the AI SDK
// cancels the underlying fetch and the async iterator throws; the error
// propagates to the StreamSession which closes the consumer stream.
for await (const chunk of chunkStream) {
if (chunk.type === 'error' && isAttachmentValidationError(chunk.error)) throw chunk.error;
// The rejected result promises surface this error after the stream
// closes. Forwarding it here would mark a successful retry as failed.
// Runs before the `streamedContent` check below so a skipped chunk
// cannot make a stalled attempt look unretryable.
if (chunk.type === 'error' && NoOutputGeneratedError.isInstance(chunk.error)) continue;
// Raw chunks are normally consumed by the raw-chunk tap inside the SDK
// pipeline and never reach this loop; this branch is the fallback for
// streams the tap was not applied to (e.g. mocked streams in tests) and
// is why 'raw' stays in the retry-safe set. Capturing usage here also
// lets an aborted turn still be billed via getTerminalFinish.
if (chunk.type === 'raw') {
this.rawUsageReader?.capture(chunk.rawValue);
this.rawErrorReader?.capture(chunk.rawValue);
continue;
}
// Anything beyond transport bookkeeping counts as content: once seen,
// a stalled attempt is no longer silently retryable (see callModel).
if (!STALL_RETRY_SAFE_CHUNK_TYPES.has(chunk.type)) attemptState.streamedContent = true;
// Filter only the SDK's terminal `finish` chunk — the runtime emits its
// own consolidated `finish` after the loop completes. `start-step` /
// `finish-step` are passed through as LLM-iteration boundaries.
if (chunk.type === 'finish') continue;
// Accumulate streamed text so an abort mid-response can persist it (the
// turn's `newMessages` are only built below, which the abort never reaches).
if (chunk.type === 'text-delta') this.partialText += chunk.text ?? '';
// Provider-executed tools (e.g. native web search) skip the local
// execution loop that emits tool-execution lifecycle events via the
// event bus. Stamp them here at chunk-arrival time so live chat and the
// persisted timeline both show a duration. A failed call arrives as a
// `tool-error` part (never a `tool-result`), so close its timing there.
if ((chunk.type === 'tool-result' || chunk.type === 'tool-error') && chunk.providerExecuted) {
await this.guard.write({
type: 'tool-execution-end',
toolCallId: chunk.toolCallId,
toolName: chunk.toolName ?? '',
isError: chunk.type === 'tool-error',
endTime: Date.now(),
});
}
const converted = convertChunk(chunk);
if (converted) await this.guard.write(converted);
if (chunk.type === 'tool-call' && chunk.providerExecuted) {
await this.guard.write({
type: 'tool-execution-start',
toolCallId: chunk.toolCallId,
toolName: chunk.toolName ?? '',
startTime: Date.now(),
});
}
private async writeStreamChunk(chunk: TextStreamPart<ToolSet>): Promise<void> {
// Provider tools skip the local executor. Record their timing when chunks arrive.
if ((chunk.type === 'tool-result' || chunk.type === 'tool-error') && chunk.providerExecuted) {
await this.guard.write({
type: 'tool-execution-end',
toolCallId: chunk.toolCallId,
toolName: chunk.toolName ?? '',
isError: chunk.type === 'tool-error',
endTime: Date.now(),
});
}
const converted = convertChunk(chunk);
if (converted) await this.guard.write(converted);
if (chunk.type === 'tool-call' && chunk.providerExecuted) {
await this.guard.write({
type: 'tool-execution-start',
toolCallId: chunk.toolCallId,
toolName: chunk.toolName ?? '',
startTime: Date.now(),
});
}
}
private async resolveModelTurn(
result: ReturnType<StreamSink['startModelStream']>['result'],
ctx: ModelCallContext,
idleMs: number,
turnAbort: AbortController,
): Promise<ModelTurnResult> {
// The result promises settle as part of stream close on a healthy turn;
// guard them with the same stall deadline so an SDK-internal promise that
// never settles cannot hang the turn after the chunk loop ended.
@ -426,13 +439,10 @@ export class StreamSink implements RunOutputSink<void> {
}
async finishComplete(emission: CompleteEmission): Promise<void> {
const { list, options, finishReason, usage, structuredOutput } = emission;
const { list, finishReason, usage, structuredOutput } = emission;
const costUsage = this.services.applyCost(usage);
await this.services.saveToMemory(list, options);
await this.services.maybeGenerateTitle(list, options);
await this.services.cleanupRun();
await this.services.flushTelemetry(options);
await finalizeRun(this.services, emission);
await this.guard.write({
type: 'finish',

4
packages/@n8n/agents/src/runtime/memory/memory-orchestrator.ts
View File

@ -33,7 +33,7 @@ import { AgentEvent } from '../../types/runtime/event';
import type { AgentPersistenceOptions, ExecutionOptions, RunOptions } from '../../types/sdk/agent';
import type { AgentDbMessage } from '../../types/sdk/message';
import type { ObservationLogScope, ObservationLogTaskKind } from '../../types/sdk/observation-log';
import type { AgentRuntimeConfig } from '../loop/agent-runtime';
import type { AgentRuntimeConfig } from '../../types/runtime/agent-runtime';
import type { AgentMessageList } from '../model/message-list';
import { estimateObservationTokens, type TokenCounter } from '../model/model-token-counter';
import type { BackgroundTaskTracker } from '../state/background-task-tracker';
@ -44,7 +44,7 @@ import {
type MemorySpanAttributes,
type RuntimeTelemetry,
} from '../telemetry/runtime-telemetry';
import { sanitizeOffloadedToolResultsForMemory } from '../tools/tool-result-guard';
import { sanitizeOffloadedToolResultsForMemory } from './tool-result-memory';
const DEFAULT_MEMORY_TASK_LOCK_TTL_MS = 30_000;
/** Fraction of observerThresholdTokens at which the mid-run observer starts in the background. */

74
packages/@n8n/agents/src/runtime/memory/tool-result-memory.ts Normal file
View File

@ -0,0 +1,74 @@
import { isRecord } from '@n8n/utils/is-record';
import { toJsonValue } from '@n8n/utils/json/to-json-value';
import type { AgentDbMessage, MessageContent } from '../../types/sdk/message';
import type { JSONObject, JSONValue } from '../../types/utils/json';
import { isToolResultPath } from '../../workspace/tool-result-storage';
import { isContentToolResultOutput, type ContentToolResultOutput } from '../model/messages';
export const EXPIRED_OFFLOADED_TOOL_RESULT = {
_offloaded: true,
expired: true,
message: 'The stored tool result expired with its originating run.',
} satisfies JSONObject;
const EXPIRED_OFFLOADED_TOOL_RESULT_JSON = JSON.stringify(EXPIRED_OFFLOADED_TOOL_RESULT);
function isOffloadedToolResult(value: unknown): boolean {
return (
isRecord(value) &&
value._offloaded === true &&
typeof value.path === 'string' &&
isToolResultPath(value.path)
);
}
function isSerializedOffloadedToolResult(value: string): boolean {
try {
const parsed: unknown = JSON.parse(value);
return isOffloadedToolResult(parsed);
} catch {
return false;
}
}
export function sanitizeOffloadedToolResultsForMemory(
messages: AgentDbMessage[],
): AgentDbMessage[] {
return messages.map((message) => {
if (!('content' in message)) return { ...message };
const content = message.content.map(sanitizeOffloadedContentBlock);
return { ...message, content };
});
}
function sanitizeOffloadedContentBlock(block: MessageContent): MessageContent {
if (block.type === 'tool-call') {
if (block.state === 'resolved' && isOffloadedToolResult(block.output)) {
return { ...block, output: { ...EXPIRED_OFFLOADED_TOOL_RESULT } };
}
if (block.state === 'resolved' && isContentToolResultOutput(block.output)) {
return { ...block, output: sanitizeOffloadedContentResult(block.output) };
}
if (block.state === 'rejected' && isSerializedOffloadedToolResult(block.error)) {
return { ...block, error: EXPIRED_OFFLOADED_TOOL_RESULT_JSON };
}
}
if (block.type === 'text' && isSerializedOffloadedToolResult(block.text)) {
return { ...block, text: EXPIRED_OFFLOADED_TOOL_RESULT_JSON };
}
return { ...block };
}
function sanitizeOffloadedContentResult(output: ContentToolResultOutput): JSONValue {
return toJsonValue({
type: 'content',
value: output.value.map((part) => {
if (part.type !== 'text' || !isSerializedOffloadedToolResult(part.text)) return part;
return { ...part, text: EXPIRED_OFFLOADED_TOOL_RESULT_JSON };
}),
});
}

2
packages/@n8n/agents/src/runtime/model/raw-error.ts
View File

@ -2,7 +2,7 @@ import { getProviderPrefix } from '@n8n/ai-utilities/agent-config';
import { isRecord } from '@n8n/utils/is-record';
import { type ProviderId } from './provider-credentials';
import type { ModelTurnErrorType } from '../loop/run-output-sink';
import type { ModelTurnErrorType } from '../../types/runtime/agent-loop';
/** A recognized failure signal read from the provider's raw stream events. */
export interface RawProviderError {

2
packages/@n8n/agents/src/runtime/telemetry/runtime-telemetry.ts
View File

@ -14,7 +14,7 @@ import type { ExecutionOptions } from '../../types/sdk/agent';
import type { OpaqueSpanLink } from '../../types/telemetry';
import type { JSONValue } from '../../types/utils/json';
import { isZodSchema } from '../../utils/zod';
import type { AgentRuntimeConfig } from '../loop/agent-runtime';
import type { AgentRuntimeConfig } from '../../types/runtime/agent-runtime';
interface TelemetrySpan {
end(): void;

553
packages/@n8n/agents/src/runtime/tools/delegate-sub-agent-runner.ts Normal file
View File

@ -0,0 +1,553 @@
import { isRecord } from '@n8n/utils/is-record';
import type { JSONSchema7 } from 'json-schema';
import {
assertSubAgentTaskPath,
createChildSubAgentTaskPath,
type SubAgentTaskPath,
} from './sub-agent-task-path';
import { isAbortError } from '../../sdk/abort';
import { filterLlmMessages } from '../../sdk/message';
import {
DELEGATE_SUB_AGENT_TOOL_NAME,
DELEGATED_CHILD_SUSPEND_UNSUPPORTED_MESSAGE,
parseDelegateSubAgentContinuation,
type CreateDelegateSubAgentToolOptions,
type DelegateSubAgentCheckpointTarget,
type DelegateSubAgentContinuation,
type DelegateSubAgentInput,
type DelegateSubAgentPolicy,
type DelegateSubAgentRequest,
type DelegateSubAgentResumeRunner,
type DelegateSubAgentRunnerHelpers,
type DelegateSubAgentToolOutput,
} from '../../types/runtime/delegation';
import { AgentEvent, type ForwardedChildChunk } from '../../types/runtime/event';
import type { GenerateResult, StreamChunk, TokenUsage } from '../../types/sdk/agent';
import type { AgentMessage } from '../../types/sdk/message';
import type {
InterruptibleToolContext,
ToolCancellationContext,
ToolContext,
} from '../../types/sdk/tool';
import type { JSONValue } from '../../types/utils/json';
import { withoutMessageCount } from '../loop/execution-counter';
/** Cap on forwarded text+reasoning characters per delegation. */
const SUB_AGENT_FORWARD_CHAR_BUDGET = 20_000;
const FORWARDED_CHILD_CHUNK_TYPES = new Set<ForwardedChildChunk['type']>([
'text-delta',
'reasoning-start',
'reasoning-delta',
'reasoning-end',
'tool-input-start',
'tool-execution-start',
'tool-execution-end',
]);
function isForwardedChildChunk(chunk: StreamChunk): chunk is ForwardedChildChunk {
return FORWARDED_CHILD_CHUNK_TYPES.has(chunk.type as ForwardedChildChunk['type']);
}
function createEmitChunkHelper(
ctx: ToolContext,
request: DelegateSubAgentRequest,
): (chunk: StreamChunk) => void {
let forwardedChars = 0;
return (chunk) => {
if (request.parentToolCallId === undefined) return;
if (!isForwardedChildChunk(chunk)) return;
let forwarded: ForwardedChildChunk = chunk;
if (chunk.type === 'text-delta' || chunk.type === 'reasoning-delta') {
// Trim rather than pass through whole: a single provider delta can be
// arbitrarily large, and forwarding it intact would blow the budget.
const remaining = SUB_AGENT_FORWARD_CHAR_BUDGET - forwardedChars;
if (remaining <= 0) return;
const delta = chunk.delta.slice(0, remaining);
forwardedChars += delta.length;
forwarded = { ...chunk, delta };
}
ctx.emitEvent?.({
type: AgentEvent.SubAgentChunk,
...subAgentLifecycleBase(request),
chunk: forwarded,
});
};
}
/**
* Tool handler: assign the child's task path,
* assemble the {@link DelegateSubAgentRequest} from the model input plus the
* parent tool context, then run the child via the host `runSubAgent` callback
* while emitting started/progress/completed lifecycle events. A failure is
* converted into a `status: 'failed'` output (never thrown) so one failed
* delegation can't abort the parent's run; an abort is rethrown, because a
* cancelled run must end the parent too.
*/
export async function handleDelegateSubAgent(
input: DelegateSubAgentInput,
ctx: ToolContext | InterruptibleToolContext,
options: CreateDelegateSubAgentToolOptions,
childPathIndexes: Map<string, number>,
): Promise<DelegateSubAgentToolOutput> {
let taskPath: SubAgentTaskPath | undefined;
let request: DelegateSubAgentRequest | undefined;
let startedAt: number | undefined;
try {
if (
options.resumeSubAgent !== undefined &&
isInterruptibleToolContext(ctx) &&
ctx.continuation !== undefined
) {
const restored = restoreDelegateRequest(
input,
ctx,
ctx.continuation,
options.policy,
childPathIndexes,
);
const { checkpoint } = restored;
taskPath = checkpoint.taskPath;
request = restored.request;
startedAt = Date.now();
emitSubAgentStarted(ctx, request, startedAt);
return await resumeDelegatedChild(ctx, restored, startedAt, options, options.resumeSubAgent);
}
const childPathIndexKey = getChildPathIndexKey(ctx);
const childPathIndex = childPathIndexes.get(childPathIndexKey) ?? 0;
taskPath = createChildSubAgentTaskPath(input.taskName, childPathIndex);
childPathIndexes.set(childPathIndexKey, childPathIndex + 1);
request = createDelegateSubAgentRequest(input, ctx, taskPath, childPathIndex, options.policy);
startedAt = Date.now();
emitSubAgentStarted(ctx, request, startedAt);
return await runDelegatedChild(ctx, request, startedAt, options);
} catch (error) {
return handleDelegateFailure(error, ctx, taskPath, request, startedAt);
}
}
async function resumeDelegatedChild(
ctx: InterruptibleToolContext,
{ request, checkpoint }: ReturnType<typeof restoreDelegateRequest>,
startedAt: number,
options: CreateDelegateSubAgentToolOptions,
resumeSubAgent: DelegateSubAgentResumeRunner,
): Promise<DelegateSubAgentToolOutput> {
let output: DelegateSubAgentToolOutput;
try {
output = await resumeSubAgent(
{
...request,
...getChildCheckpointTarget(checkpoint),
resumeData: ctx.resumeData,
},
createRunnerHelpers(ctx, request, options.name),
);
} catch (error) {
if (ctx.abortSignal?.aborted || isAbortError(error)) throw error;
if (options.shouldRetrySubAgentResumeError?.(error) === false) throw error;
return await ctx.suspend(ctx.suspendPayload);
}
emitSubAgentCompleted(ctx, request, output, startedAt);
if (output.status === 'suspended') {
return await cascadeChildSuspension(ctx, request, output);
}
return output;
}
async function runDelegatedChild(
ctx: ToolContext | InterruptibleToolContext,
request: DelegateSubAgentRequest,
startedAt: number,
options: CreateDelegateSubAgentToolOptions,
): Promise<DelegateSubAgentToolOutput> {
const toolName = options.name ?? DELEGATE_SUB_AGENT_TOOL_NAME;
if (!options.runSubAgent) {
throw new Error(
`${toolName} was registered without a runSubAgent callback, and no host runner was provided. Register it on an Agent (for inline delegation) or pass runSubAgent.`,
);
}
const output = await options.runSubAgent(
request,
createRunnerHelpers(ctx, request, options.name),
);
emitSubAgentCompleted(ctx, request, output, startedAt);
if (
output.status === 'suspended' &&
options.resumeSubAgent !== undefined &&
isInterruptibleToolContext(ctx)
) {
return await cascadeChildSuspension(ctx, request, output);
}
return output;
}
function handleDelegateFailure(
error: unknown,
ctx: ToolContext,
taskPath: SubAgentTaskPath | undefined,
request: DelegateSubAgentRequest | undefined,
startedAt: number | undefined,
): DelegateSubAgentToolOutput {
// When the parent has a signal it is the authority: `isAbortError` also
// matches by message text, and an unrelated child error must not be
// mistaken for a cancellation and kill the parent run.
const aborted = ctx.abortSignal ? ctx.abortSignal.aborted : isAbortError(error);
const output: DelegateSubAgentToolOutput = {
status: aborted ? 'cancelled' : 'failed',
...(taskPath !== undefined ? { taskPath } : {}),
answer: '',
...(aborted ? {} : { error: stringifyUnknown(error) }),
};
if (request !== undefined && startedAt !== undefined) {
emitSubAgentCompleted(ctx, request, output, startedAt);
}
if (aborted) throw error;
return output;
}
function createDelegateSubAgentRequest(
input: DelegateSubAgentInput,
ctx: ToolContext,
taskPath: SubAgentTaskPath,
childCount: number,
policy?: DelegateSubAgentPolicy,
): DelegateSubAgentRequest {
return {
...input,
taskPath,
childCount,
...(ctx.runId !== undefined ? { parentRunId: ctx.runId } : {}),
...(ctx.persistence?.threadId !== undefined
? { parentThreadId: ctx.persistence.threadId }
: {}),
...(ctx.persistence?.resourceId !== undefined
? { parentResourceId: ctx.persistence.resourceId }
: {}),
...(ctx.persistence?.hostMetadata !== undefined
? { parentHostMetadata: ctx.persistence.hostMetadata }
: {}),
...(ctx.abortSignal !== undefined ? { parentAbortSignal: ctx.abortSignal } : {}),
...(ctx.toolCallId !== undefined ? { parentToolCallId: ctx.toolCallId } : {}),
...(ctx.executionCounter !== undefined
? { parentExecutionCounter: withoutMessageCount(ctx.executionCounter) }
: {}),
...(ctx.parentTelemetry !== undefined ? { parentTelemetry: ctx.parentTelemetry } : {}),
...(policy !== undefined ? { policy } : {}),
};
}
function isJsonSchemaObject(value: unknown): value is JSONSchema7 {
return isRecord(value);
}
export async function cancelDelegatedSubAgent(
input: DelegateSubAgentInput,
ctx: ToolCancellationContext,
options: CreateDelegateSubAgentToolOptions,
childPathIndexes: Map<string, number>,
): Promise<void> {
if (options.cancelSubAgent === undefined || ctx.continuation === undefined) return;
const { checkpoint, request } = restoreDelegateRequest(
input,
ctx,
ctx.continuation,
options.policy,
childPathIndexes,
);
await options.cancelSubAgent(
{
...request,
...getChildCheckpointTarget(checkpoint),
reason: ctx.cancellation.message,
},
createRunnerHelpers(ctx, request, options.name),
);
}
function restoreDelegateRequest(
input: DelegateSubAgentInput,
ctx: ToolContext,
continuation: JSONValue,
policy: DelegateSubAgentPolicy | undefined,
childPathIndexes: Map<string, number>,
): {
checkpoint: DelegateSubAgentContinuation & { taskPath: SubAgentTaskPath };
request: DelegateSubAgentRequest;
} {
const checkpoint = parseDelegateSubAgentContinuation(continuation);
if (!checkpoint) {
throw new Error('Delegated child checkpoint metadata is missing or invalid');
}
if (checkpoint.subAgentId !== input.subAgentId) {
throw new Error('Delegated child checkpoint does not match the selected sub-agent');
}
const { taskPath } = checkpoint;
assertSubAgentTaskPath(taskPath);
const key = getChildPathIndexKey(ctx);
childPathIndexes.set(key, Math.max(childPathIndexes.get(key) ?? 0, checkpoint.childCount + 1));
const request = createDelegateSubAgentRequest(
input,
ctx,
taskPath,
checkpoint.childCount,
policy,
);
return { checkpoint: { ...checkpoint, taskPath }, request };
}
function getChildCheckpointTarget(
checkpoint: DelegateSubAgentContinuation,
): DelegateSubAgentCheckpointTarget {
return {
childRunId: checkpoint.runId,
childToolCallId: checkpoint.toolCallId,
...(checkpoint.threadId !== undefined ? { childThreadId: checkpoint.threadId } : {}),
...(checkpoint.resumeContext !== undefined ? { resumeContext: checkpoint.resumeContext } : {}),
};
}
function createRunnerHelpers(
ctx: ToolContext | InterruptibleToolContext,
request: DelegateSubAgentRequest,
configuredToolName?: string,
): DelegateSubAgentRunnerHelpers {
const toolName = configuredToolName ?? DELEGATE_SUB_AGENT_TOOL_NAME;
return {
runInlineSubAgent: () => {
throw new Error(
`${toolName} host runner does not support inline delegation without helpers.runInlineSubAgent from an Agent build.`,
);
},
emitChunk: createEmitChunkHelper(ctx, request),
};
}
function isInterruptibleToolContext(
ctx: ToolContext | InterruptibleToolContext,
): ctx is InterruptibleToolContext {
return 'suspend' in ctx;
}
async function cascadeChildSuspension(
ctx: InterruptibleToolContext,
request: DelegateSubAgentRequest,
output: DelegateSubAgentToolOutput,
): Promise<never> {
const suspension = output.pendingSuspend?.[0];
if (!suspension || !isJsonSchemaObject(suspension.resumeSchema)) {
throw new Error(DELEGATED_CHILD_SUSPEND_UNSUPPORTED_MESSAGE);
}
return await ctx.suspend(suspension.suspendPayload, {
resumeSchema: suspension.resumeSchema,
continuation: {
runId: suspension.runId,
toolCallId: suspension.toolCallId,
taskPath: request.taskPath,
subAgentId: request.subAgentId,
childCount: request.childCount,
...(output.threadId !== undefined ? { threadId: output.threadId } : {}),
...(output.resumeContext !== undefined ? { resumeContext: output.resumeContext } : {}),
},
});
}
function emitSubAgentStarted(
ctx: ToolContext,
request: DelegateSubAgentRequest,
startedAt: number,
): void {
ctx.emitEvent?.({
type: AgentEvent.SubAgentStarted,
...subAgentLifecycleBase(request),
startedAt,
});
}
function emitSubAgentCompleted(
ctx: ToolContext,
request: DelegateSubAgentRequest,
output: DelegateSubAgentToolOutput,
startedAt: number,
): void {
const finishedAt = Date.now();
ctx.emitEvent?.({
type: AgentEvent.SubAgentCompleted,
...subAgentLifecycleBase(request),
status: output.status,
startedAt,
finishedAt,
durationMs: finishedAt - startedAt,
...(output.runId !== undefined ? { runId: output.runId } : {}),
...(output.threadId !== undefined ? { threadId: output.threadId } : {}),
...(output.usage !== undefined ? { usage: output.usage } : {}),
...(output.finishReason !== undefined ? { finishReason: output.finishReason } : {}),
...(output.model !== undefined ? { model: output.model } : {}),
...(output.error !== undefined ? { error: output.error } : {}),
});
}
function subAgentLifecycleBase(request: DelegateSubAgentRequest) {
return {
taskName: request.taskName,
taskPath: request.taskPath,
...(request.parentRunId !== undefined ? { parentRunId: request.parentRunId } : {}),
...(request.parentToolCallId !== undefined
? { parentToolCallId: request.parentToolCallId }
: {}),
...(request.subAgentId !== undefined ? { subAgentId: request.subAgentId } : {}),
};
}
function getChildPathIndexKey(ctx: ToolContext): string {
return ctx.runId ?? ctx.persistence?.threadId ?? ctx.persistence?.resourceId ?? 'adhoc';
}
function stringifyUnknown(value: unknown): string {
if (value instanceof Error) return value.message;
if (typeof value === 'string') return value;
if (typeof value === 'number' || typeof value === 'boolean' || value === null) {
return String(value);
}
try {
return JSON.stringify(value) ?? 'Unknown error';
} catch {
return 'Unknown error';
}
}
/**
* Optional helpers for a `runSubAgent` implementation.
*
* A host that runs the child by calling `agent.generate(...)`/`stream(...)` can
* reuse these instead of hand-rolling the delegation prompt and the result
* mapping. They are NOT wired into the tool — call them from your `runSubAgent`
* (the n8n CLI runner does).
*/
/** Render the default delegation prompt from a request's goal / context / expectedOutput. */
export function renderDelegateSubAgentPrompt(request: {
goal: string;
context?: string;
expectedOutput?: string;
}): string {
const sections = [
'You are a focused subagent working on a specific delegated task.',
`YOUR TASK:\n${request.goal}`,
];
if (request.context) {
sections.push(`CONTEXT:\n${request.context}`);
}
if (request.expectedOutput) {
sections.push(`EXPECTED OUTPUT:\n${request.expectedOutput}`);
}
sections.push(
[
'Complete this task using the tools available to you. When finished, provide a clear, concise summary of:',
'- What you did',
'- What you found or accomplished',
'- Important outputs, decisions, or evidence',
'- Any issues, assumptions, or limitations',
'',
'If the information above is insufficient, do your best with explicitly stated assumptions and note what was missing, rather than stopping to ask.',
'',
'Be thorough but concise -- your response is returned to the parent agent as a summary.',
].join('\n'),
);
return sections.join('\n\n');
}
function resolveDelegateSubAgentStatus(
result: GenerateResult,
): DelegateSubAgentToolOutput['status'] {
if (result.finishReason === 'error' || result.error !== undefined) {
// An aborted child sets its state to `cancelled` but still reports an
// error, so the state decides which of the two this really was.
return result.getState().status === 'cancelled' ? 'cancelled' : 'failed';
}
if (result.pendingSuspend !== undefined && result.pendingSuspend.length > 0) {
return 'suspended';
}
return 'completed';
}
/** Failed delegate output when a child run suspends for user input (not yet resumable). */
export function failedDelegatedChildSuspendOutput(
taskPath: SubAgentTaskPath,
model?: string,
): DelegateSubAgentToolOutput {
return {
status: 'failed',
taskPath,
answer: '',
error: DELEGATED_CHILD_SUSPEND_UNSUPPORTED_MESSAGE,
...(model !== undefined ? { model } : {}),
};
}
/** Map an agent {@link GenerateResult} into the delegate tool's output shape. */
export function generateResultToDelegateSubAgentOutput(
taskPath: SubAgentTaskPath,
result: GenerateResult,
threadId?: string,
): DelegateSubAgentToolOutput {
const status = resolveDelegateSubAgentStatus(result);
return {
status,
taskPath,
runId: result.runId,
...(threadId !== undefined ? { threadId } : {}),
...(result.model !== undefined ? { model: result.model } : {}),
answer: lastText(result.messages),
...(result.structuredOutput !== undefined ? { structuredOutput: result.structuredOutput } : {}),
...(result.usage !== undefined
? {
usage: toDelegateUsage(result.usage),
}
: {}),
...(result.finishReason !== undefined ? { finishReason: result.finishReason } : {}),
...(result.error !== undefined ? { error: stringifyUnknown(result.error) } : {}),
...(status === 'suspended' && result.pendingSuspend !== undefined
? { pendingSuspend: result.pendingSuspend }
: {}),
};
}
function toDelegateUsage(usage: TokenUsage): DelegateSubAgentToolOutput['usage'] {
return {
promptTokens: usage.promptTokens,
completionTokens: usage.completionTokens,
totalTokens: usage.totalTokens,
...(usage.cost !== undefined ? { cost: usage.cost } : {}),
};
}
/** Last non-empty assistant text across the run's messages. */
function lastText(messages: AgentMessage[]): string {
const llmMessages = filterLlmMessages(messages);
for (let i = llmMessages.length - 1; i >= 0; i--) {
const message = llmMessages[i];
if (!message) continue;
const text = message.content
.filter((content) => content.type === 'text')
.map((content) => content.text)
.join('\n')
.trim();
if (text) return text;
}
return '';
}

994
packages/@n8n/agents/src/runtime/tools/delegate-sub-agent-tool.ts
View File

File diff suppressed because it is too large Load Diff

82
packages/@n8n/agents/src/runtime/tools/tool-adapter.ts
View File

@ -117,42 +117,14 @@ export async function executeTool(
throw new Error(`No handler found for tool "${builtTool.name}"`);
}
// Anchor programmatic skill activations to the calling tool's result by
// default, so the skill body rides on that result (kept out of `system`)
// instead of forcing a full prompt-prefix rewrite. An explicit anchor wins.
const baseLoadSkill = executionContext.loadSkill;
const loadSkill: ToolExecutionContext['loadSkill'] = baseLoadSkill
? async (skillId, anchor) =>
await baseLoadSkill(skillId, anchor ?? (toolCallId ? { toolCallId } : undefined))
: undefined;
const loadSkill = anchorSkillLoader(executionContext.loadSkill, toolCallId);
if (builtTool.suspendSchema) {
const isCancelled = isCancellation(resumeData);
const ctx: InterruptibleToolContext = {
suspend: async (payload: unknown, options?: ToolSuspendOptions): Promise<never> => {
const resolvedOptions: ToolSuspendOptions = {
continuation: executionContext.continuation,
...options,
resumeSchema: options?.resumeSchema ?? executionContext.resumeSchema,
};
await executionContext.onSuspend?.(payload, resolvedOptions);
return await Promise.resolve({
[SUSPEND_BRAND]: true,
payload,
...resolvedOptions,
} as never);
},
suspend: createSuspendHandler(executionContext),
resumeData: isCancelled ? undefined : resumeData,
cancellation: isCancelled ? { message: resumeData.message } : undefined,
parentTelemetry,
toolCallId,
toolName: builtTool.name,
runId: executionContext.runId,
...(loadSkill ? { loadSkill } : {}),
persistence: executionContext.persistence,
emitEvent: executionContext.emitEvent,
abortSignal: executionContext.abortSignal,
executionCounter: executionContext.executionCounter,
...createToolContext(builtTool, executionContext, parentTelemetry, toolCallId, loadSkill),
suspendPayload: executionContext.suspendPayload,
continuation: executionContext.continuation,
resumeSchema: executionContext.resumeSchema,
@ -160,7 +132,34 @@ export async function executeTool(
return await builtTool.handler(args, ctx);
}
const ctx: ToolContext = {
const ctx = createToolContext(
builtTool,
executionContext,
parentTelemetry,
toolCallId,
loadSkill,
);
return await builtTool.handler(args, ctx);
}
function anchorSkillLoader(
loadSkill: ToolExecutionContext['loadSkill'],
toolCallId: string | undefined,
): ToolExecutionContext['loadSkill'] {
if (!loadSkill) return undefined;
// Keep skill content on the tool result. An explicit anchor takes precedence.
return async (skillId, anchor) =>
await loadSkill(skillId, anchor ?? (toolCallId ? { toolCallId } : undefined));
}
function createToolContext(
builtTool: BuiltTool,
executionContext: ToolExecutionContext,
parentTelemetry: BuiltTelemetry | undefined,
toolCallId: string | undefined,
loadSkill: ToolExecutionContext['loadSkill'],
): ToolContext {
return {
parentTelemetry,
toolCallId,
toolName: builtTool.name,
@ -171,7 +170,24 @@ export async function executeTool(
abortSignal: executionContext.abortSignal,
executionCounter: executionContext.executionCounter,
};
return await builtTool.handler(args, ctx);
}
function createSuspendHandler(
executionContext: ToolExecutionContext,
): InterruptibleToolContext['suspend'] {
return async (payload: unknown, options?: ToolSuspendOptions): Promise<never> => {
const resolvedOptions: ToolSuspendOptions = {
continuation: executionContext.continuation,
...options,
resumeSchema: options?.resumeSchema ?? executionContext.resumeSchema,
};
await executionContext.onSuspend?.(payload, resolvedOptions);
return await Promise.resolve({
[SUSPEND_BRAND]: true,
payload,
...resolvedOptions,
} as never);
};
}
/**

1259
packages/@n8n/agents/src/runtime/tools/tool-call-executor.ts
View File

File diff suppressed because it is too large Load Diff

520
packages/@n8n/agents/src/runtime/tools/tool-call-runner.ts Normal file
View File

@ -0,0 +1,520 @@
import { zodSchemaToJsonSchema } from '@n8n/ai-utilities/json-schema';
import type { JSONSchema7 } from 'json-schema';
import { executeTool, isSuspendedToolResult, type SuspendedToolResult } from './tool-adapter';
import {
guardToolErrorForModel,
guardToolMessageForModel,
guardToolResultForModel,
type ToolResultGuardStorage,
} from './tool-result-guard';
import {
protectUntrustedToolError,
protectUntrustedToolMessage,
protectUntrustedToolResult,
} from './untrusted-tool-output';
import { isAbortError, raceWithAbort } from '../../sdk/abort';
import { isCancellation } from '../../sdk/cancellation';
import { isLlmMessage } from '../../sdk/message';
import type { BuiltTool, ToolSuspendOptions } from '../../types';
import { AgentEvent } from '../../types/runtime/event';
import type {
ProcessToolCallParams,
ToolCallExecutorDeps,
ToolCallOutcome,
} from '../../types/runtime/tool-execution';
import type { AgentMessage, ContentToolCall, Message } from '../../types/sdk/message';
import type { JSONValue } from '../../types/utils/json';
import { parseWithSchema } from '../../utils/parse';
import { isZodSchema } from '../../utils/zod';
import { incrementToolCallCount } from '../loop/execution-counter';
import { stringifyError } from '../loop/runtime-helpers';
import type { AgentMessageList } from '../model/message-list';
import type { AgentEventBus } from '../state/event-bus';
import type { RuntimeTelemetry } from '../telemetry/runtime-telemetry';
interface InterruptedToolSuspension {
didSuspend: boolean;
abortObserved: boolean;
payload: unknown;
options?: ToolSuspendOptions;
cleanup?: Promise<void>;
}
/** A tool-call content block that has already been settled by the AI SDK. */
type SettledToolCall = ContentToolCall & { state: 'resolved' | 'rejected' };
function isDeniedApprovalResumeData(value: unknown): boolean {
return value !== null && typeof value === 'object' && Reflect.get(value, 'approved') === false;
}
function shouldEmitToolExecutionStart(tool: BuiltTool, resumeData: unknown): boolean {
if (!tool.approval) return true;
if (!tool.approval.required && tool.approval.conditional !== true) return true;
if (resumeData === undefined) return false;
return !isDeniedApprovalResumeData(resumeData);
}
function getToolResumeJsonSchema(
tool: BuiltTool,
resumeSchemaOverride?: ToolSuspendOptions['resumeSchema'],
): JSONSchema7 | undefined {
const resolvedSchema = resumeSchemaOverride ?? tool.resumeSchema;
if (!resolvedSchema) return undefined;
return isZodSchema(resolvedSchema) ? zodSchemaToJsonSchema(resolvedSchema) : resolvedSchema;
}
/** Execute one tool call and keep its result, cancellation, and lifecycle events together. */
export class ToolCallRunner {
private offloadedToolResults = false;
constructor(private readonly deps: ToolCallExecutorDeps) {}
get hasOffloadedToolResults(): boolean {
return this.offloadedToolResults;
}
private get telemetry(): RuntimeTelemetry {
return this.deps.telemetry;
}
private get eventBus(): AgentEventBus {
return this.deps.eventBus;
}
/**
* Execute a single tool call, emit lifecycle events, record the result in the
* message list, and return the outcome. The caller is responsible for writing
* any stream chunks and for handling suspension (building pendingToolCalls and
* persisting state).
*
* On tool execution errors, emits ToolExecutionEnd with isError=true, adds an
* error tool-result message to the list so the LLM can self-correct, and returns
* `{ outcome: 'error' }` — never re-throws.
*/
async processToolCall(params: ProcessToolCallParams): Promise<ToolCallOutcome> {
const { toolName, toolMap, list, toolCallId, resumeData } = params;
const builtTool = toolMap.get(toolName);
if (!builtTool) {
return await this.toolError(params, new Error(`Tool ${toolName} not found`));
}
// Already settled by the AI SDK (e.g. provider-executed tools): emit the
// lifecycle end and skip re-execution.
const settledBlock = this.findSettledToolCall(list, toolCallId);
if (settledBlock) {
return this.completeSettledToolCall(params, settledBlock);
}
if (isCancellation(resumeData) && !builtTool.handleCancellation) {
try {
await this.runCancellationCleanup(params, builtTool, resumeData.message);
} catch {
return { outcome: 'retryable-error' };
}
return this.buildCancelledOutcome(params, resumeData.message);
}
if (params.countToolCall ?? true) {
incrementToolCallCount(params.executionCounter);
}
const validation = await this.validateToolInput(params, builtTool);
if (!validation.ok) return validation.outcome;
const input = validation.input;
if (shouldEmitToolExecutionStart(builtTool, resumeData)) {
this.eventBus.emit({
type: AgentEvent.ToolExecutionStart,
toolCallId,
toolName,
args: input,
});
}
return await this.executeValidatedToolCall(params, builtTool, input);
}
private async executeValidatedToolCall(
params: ProcessToolCallParams,
builtTool: BuiltTool,
input: JSONValue,
): Promise<ToolCallOutcome> {
let toolResult: unknown;
const suspension: InterruptedToolSuspension = {
didSuspend: false,
abortObserved: false,
payload: undefined,
};
try {
toolResult = await this.runToolHandler(params, builtTool, input, async (payload, options) => {
suspension.didSuspend = true;
suspension.payload = payload;
suspension.options = options;
if (suspension.abortObserved || params.abortSignal?.aborted) {
await this.cleanupInterruptedSuspension(params, builtTool, input, suspension);
}
});
} catch (error) {
if (isAbortError(error) || params.abortSignal?.aborted) {
suspension.abortObserved = true;
await this.cleanupAbortedTool(params, builtTool, input, suspension);
this.deps.onCancelled();
return this.buildCancelledOutcome(params, 'Run aborted');
}
return await this.toolError(params, error, builtTool);
}
if (isSuspendedToolResult(toolResult)) {
return await this.buildSuspendedOutcome(params, builtTool, toolResult);
}
return await this.buildSuccessOutcome(params, builtTool, input, toolResult);
}
private async cleanupInterruptedSuspension(
params: ProcessToolCallParams,
builtTool: BuiltTool,
input: JSONValue,
suspension: InterruptedToolSuspension,
): Promise<void> {
// Both abort and late suspension can reach this path. Run cleanup only once.
suspension.cleanup ??= this.runCancellationCleanup(
{
...params,
input,
suspendPayload: suspension.payload,
continuation: suspension.options?.continuation,
resumeSchema: getToolResumeJsonSchema(builtTool, suspension.options?.resumeSchema),
},
builtTool,
'Run aborted',
).catch(() => undefined);
await suspension.cleanup;
}
private async cleanupAbortedTool(
params: ProcessToolCallParams,
builtTool: BuiltTool,
input: JSONValue,
suspension: InterruptedToolSuspension,
): Promise<void> {
if (suspension.didSuspend) {
await this.cleanupInterruptedSuspension(params, builtTool, input, suspension);
} else if (params.suspendPayload !== undefined || params.continuation !== undefined) {
try {
await this.runCancellationCleanup({ ...params, input }, builtTool, 'Run aborted');
} catch {
// Parent shutdown must continue; persistent stores will prune stale checkpoints.
}
}
}
async runCancellationCleanup(
params: ProcessToolCallParams,
builtTool: BuiltTool,
message: string,
): Promise<void> {
if (!builtTool.onCancellation) return;
await builtTool.onCancellation(params.input, {
cancellation: { message },
toolCallId: params.toolCallId,
toolName: params.toolName,
runId: params.runId,
persistence: params.persistence,
parentTelemetry: params.resolvedTelemetry,
emitEvent: (event) => this.eventBus.emit(event),
abortSignal: params.abortSignal,
executionCounter: params.executionCounter,
suspendPayload: params.suspendPayload,
continuation: params.continuation,
resumeSchema: params.resumeSchema,
});
}
/**
* Emit a failed ToolExecutionEnd, record the error on the list, return an
* error outcome. Pass `builtTool` only for errors authored by the tool
* itself (handler or transform failures) so runtime-authored errors such as
* input validation stay plain; tool-authored text from untrusted tools is
* wrapped before the size guard so any offloaded copy stays protected.
*/
private async toolError(
params: ProcessToolCallParams,
error: unknown,
builtTool?: BuiltTool,
): Promise<ToolCallOutcome> {
this.eventBus.emit({
type: AgentEvent.ToolExecutionEnd,
toolCallId: params.toolCallId,
toolName: params.toolName,
result: error,
isError: true,
});
const errorText = stringifyError(error);
const guardedError = await guardToolErrorForModel(
builtTool?.outputTrust === 'untrusted'
? protectUntrustedToolError(errorText, builtTool)
: errorText,
this.deps.tokenCounter,
this.getResultStorage(params),
);
params.list.setToolCallError(params.toolCallId, guardedError);
return { outcome: 'error', error: guardedError };
}
/** Find an already-settled (resolved/rejected) tool-call block for this id, if any. */
private findSettledToolCall(
list: AgentMessageList,
toolCallId: string,
): SettledToolCall | undefined {
return list
.responseDelta()
.flatMap((m) => (isLlmMessage(m) && 'content' in m ? (m as Message).content : []))
.find(
(c): c is SettledToolCall =>
c.type === 'tool-call' && c.toolCallId === toolCallId && c.state !== 'pending',
);
}
/** Emit ToolExecutionEnd for a block the AI SDK already settled; the result was written earlier. */
private completeSettledToolCall(
params: ProcessToolCallParams,
settledBlock: SettledToolCall,
): ToolCallOutcome {
const settledResult =
settledBlock.state === 'resolved' ? settledBlock.output : settledBlock.error;
this.eventBus.emit({
type: AgentEvent.ToolExecutionEnd,
toolCallId: params.toolCallId,
toolName: params.toolName,
result: settledResult,
isError: settledBlock.state === 'rejected',
});
return { outcome: 'noop' };
}
/** Record a cancelled tool call (user declined / cancelled) and build its outcome. */
private buildCancelledOutcome(
params: ProcessToolCallParams,
userMessage: string,
): ToolCallOutcome {
const { toolCallId, toolName, input, list } = params;
const modelOutput = `[Tool call cancelled. User said: "${userMessage}"]`;
this.eventBus.emit({
type: AgentEvent.ToolExecutionEnd,
toolCallId,
toolName,
result: modelOutput,
isError: false,
});
list.setToolCallResult(toolCallId, modelOutput, { canceled: true });
return {
outcome: 'cancelled',
toolEntry: { tool: toolName, input, output: modelOutput, transformed: false, canceled: true },
modelOutput,
userMessage,
canceled: true,
};
}
/** Validate input against the tool's input schema (if any). */
private async validateToolInput(
params: ProcessToolCallParams,
builtTool: BuiltTool,
): Promise<{ ok: true; input: JSONValue } | { ok: false; outcome: ToolCallOutcome }> {
if (!builtTool.inputSchema) return { ok: true, input: params.input };
const result = await parseWithSchema(builtTool.inputSchema, params.input);
if (!result.success) {
const reason = result.schemaInvalid
? `Tool ${params.toolName} has an input schema that could not be compiled: ${result.error}`
: `Invalid tool input: ${result.error}`;
return { ok: false, outcome: await this.toolError(params, new Error(reason)) };
}
return { ok: true, input: result.data as JSONValue };
}
/** Execute the tool handler inside a telemetry span. Throws on handler failure. */
private async runToolHandler(
params: ProcessToolCallParams,
builtTool: BuiltTool,
input: JSONValue,
onSuspend: (payload: unknown, options?: ToolSuspendOptions) => void,
): Promise<unknown> {
const {
toolCallId,
toolName,
runId,
persistence,
resumeData,
resolvedTelemetry,
executionCounter,
abortSignal,
suspendPayload,
continuation,
resumeSchema,
} = params;
return await this.telemetry.withToolSpan(
toolCallId,
toolName,
input,
resolvedTelemetry,
async () =>
await raceWithAbort(
async () =>
await executeTool(input, builtTool, resumeData, resolvedTelemetry, toolCallId, {
runId,
persistence,
...(this.deps.loadSkill ? { loadSkill: this.deps.loadSkill } : {}),
emitEvent: (event) => this.eventBus.emit(event),
abortSignal,
executionCounter,
suspendPayload,
continuation,
resumeSchema,
onSuspend,
}),
abortSignal,
),
);
}
/** Validate a suspend payload + resume schema and build the suspended outcome. */
private async buildSuspendedOutcome(
params: ProcessToolCallParams,
builtTool: BuiltTool,
toolResult: SuspendedToolResult,
): Promise<ToolCallOutcome> {
if (builtTool.suspendSchema) {
const parseResult = await parseWithSchema(builtTool.suspendSchema, toolResult.payload);
if (!parseResult.success) {
return await this.toolError(
params,
new Error(`Invalid suspend payload: ${parseResult.error}`),
);
}
toolResult.payload = parseResult.data;
}
const resumeSchema = getToolResumeJsonSchema(builtTool, toolResult.resumeSchema);
if (!resumeSchema) {
return await this.toolError(
params,
new Error(`Tool ${params.toolName} has no resume schema`),
);
}
return {
outcome: 'suspended',
payload: toolResult.payload,
resumeSchema,
...(toolResult.continuation !== undefined ? { continuation: toolResult.continuation } : {}),
};
}
/** Convert the tool output before recording success. */
private async buildSuccessOutcome(
params: ProcessToolCallParams,
builtTool: BuiltTool,
input: JSONValue,
toolResult: unknown,
): Promise<ToolCallOutcome> {
const { toolCallId, toolName, list } = params;
// Apply toModelOutput transform before emitting the success event.
// If the transform throws, treat it as a tool error so processToolCall
// never re-throws (preserving the "never re-throws" contract).
// Untrusted results are wrapped before the size guard so any offloaded
// or truncated copy stays protected while the guard's own envelope
// remains plain runtime text.
let modelResult: unknown;
try {
modelResult = builtTool.toModelOutput ? builtTool.toModelOutput(toolResult) : toolResult;
if (builtTool.outputTrust === 'untrusted') {
modelResult = protectUntrustedToolResult(modelResult, builtTool);
}
} catch (error) {
return await this.toolError(params, error, builtTool);
}
const storage = this.getResultStorage(params);
const guardedResult = await guardToolResultForModel(
modelResult,
this.deps.tokenCounter,
storage,
);
let guardedCustomMessage: AgentMessage | undefined;
try {
guardedCustomMessage = await this.prepareCustomToolMessage(
params,
builtTool,
toolResult,
storage,
);
} catch (error) {
return await this.toolError(params, error, builtTool);
}
this.eventBus.emit({
type: AgentEvent.ToolExecutionEnd,
toolCallId,
toolName,
result: toolResult,
isError: false,
});
list.setToolCallResult(toolCallId, guardedResult.historyOutput);
if (guardedCustomMessage) {
list.addResponse([guardedCustomMessage]);
}
return {
outcome: 'success',
toolEntry: {
tool: toolName,
input,
output: toolResult,
transformed: !!builtTool.toModelOutput,
},
modelOutput: guardedResult.wireOutput,
customMessage: guardedCustomMessage,
...(builtTool.mcpServerName !== undefined ? { mcpServerName: builtTool.mcpServerName } : {}),
};
}
private async prepareCustomToolMessage(
params: ProcessToolCallParams,
builtTool: BuiltTool,
toolResult: unknown,
storage: ToolResultGuardStorage | undefined,
): Promise<AgentMessage | undefined> {
let customMessage = await builtTool.toMessage?.(toolResult);
if (customMessage && builtTool.outputTrust === 'untrusted') {
customMessage = protectUntrustedToolMessage(customMessage, builtTool);
}
let guardedCustomMessage = customMessage
? await guardToolMessageForModel(customMessage, this.deps.tokenCounter, storage)
: undefined;
// Stamp tool provenance so derived transcripts (e.g. the observation
// log observer) can keep this content inside untrusted-data boundaries.
if (guardedCustomMessage && 'role' in guardedCustomMessage) {
guardedCustomMessage = {
...guardedCustomMessage,
origin: { kind: 'tool', toolName: params.toolName },
};
}
return guardedCustomMessage;
}
private getResultStorage(params: ProcessToolCallParams): ToolResultGuardStorage | undefined {
const filesystem = this.deps.workspaceFilesystem;
if (!filesystem) return undefined;
return {
filesystem,
runId: params.runId,
toolCallId: params.toolCallId,
onOffloaded: () => {
this.offloadedToolResults = true;
},
...(params.abortSignal ? { abortSignal: params.abortSignal } : {}),
};
}
}

96
packages/@n8n/agents/src/runtime/tools/tool-result-guard.ts
View File

@ -1,10 +1,8 @@
import { isRecord } from '@n8n/utils/is-record';
import { toJsonValue } from '@n8n/utils/json/to-json-value';
import type { AgentDbMessage, AgentMessage, MessageContent } from '../../types/sdk/message';
import type { AgentMessage } from '../../types/sdk/message';
import type { JSONObject, JSONValue } from '../../types/utils/json';
import {
isToolResultPath,
storeToolResult,
type ToolResultKind,
type ToolResultStorageScope,
@ -13,6 +11,11 @@ import type { WorkspaceFilesystem } from '../../workspace/types';
import { isContentToolResultOutput, type ContentToolResultOutput } from '../model/messages';
import { estimateObservationTokens, type TokenCounter } from '../model/model-token-counter';
export {
EXPIRED_OFFLOADED_TOOL_RESULT,
sanitizeOffloadedToolResultsForMemory,
} from '../memory/tool-result-memory';
export const MAX_MODEL_TOOL_RESULT_TOKENS = 50_000;
// BPE tokens cannot exceed UTF-8 bytes; reserve room for provider message framing.
const SMALL_PAYLOAD_OVERHEAD_BYTES = 1_024;
@ -40,14 +43,6 @@ interface OffloadedToolResult extends JSONObject {
message: string;
}
export const EXPIRED_OFFLOADED_TOOL_RESULT = {
_offloaded: true,
expired: true,
message: 'The stored tool result expired with its originating run.',
} satisfies JSONObject;
const EXPIRED_OFFLOADED_TOOL_RESULT_JSON = JSON.stringify(EXPIRED_OFFLOADED_TOOL_RESULT);
export interface ToolResultGuardStorage extends ToolResultStorageScope {
filesystem: WorkspaceFilesystem;
/** Invoked after a result was successfully offloaded to the workspace filesystem. */
@ -120,14 +115,7 @@ async function guardContentToolResultForModel(
}
const replacement = JSON.stringify(guardedText.historyOutput);
let replacedText = false;
const value = output.value.flatMap((part): ContentToolResultOutput['value'] => {
if (part.type !== 'text') return [part];
if (replacedText) return [];
replacedText = true;
return [{ ...part, text: replacement }];
});
const value = replaceTextParts(output.value, replacement);
const wireOutput: ContentToolResultOutput = { ...output, value };
return {
@ -161,72 +149,18 @@ export async function guardToolMessageForModel(
if (!guarded.truncated && !guarded.offloaded) return message;
const replacement = JSON.stringify(guarded.historyOutput);
let replacedText = false;
const content = message.content.flatMap((block): MessageContent[] => {
if (block.type !== 'text') return [block];
if (replacedText) return [];
replacedText = true;
return [{ ...block, text: replacement }];
});
const content = replaceTextParts(message.content, replacement);
return { ...message, content };
}
function isOffloadedToolResult(value: unknown): boolean {
return (
isRecord(value) &&
value._offloaded === true &&
typeof value.path === 'string' &&
isToolResultPath(value.path)
);
}
function isSerializedOffloadedToolResult(value: string): boolean {
try {
const parsed: unknown = JSON.parse(value);
return isOffloadedToolResult(parsed);
} catch {
return false;
}
}
export function sanitizeOffloadedToolResultsForMemory(
messages: AgentDbMessage[],
): AgentDbMessage[] {
return messages.map((message) => {
if (!('content' in message)) return { ...message };
const content = message.content.map((block): MessageContent => {
if (block.type === 'tool-call') {
if (block.state === 'resolved' && isOffloadedToolResult(block.output)) {
return { ...block, output: { ...EXPIRED_OFFLOADED_TOOL_RESULT } };
}
if (block.state === 'resolved' && isContentToolResultOutput(block.output)) {
return {
...block,
output: toJsonValue({
type: 'content',
value: block.output.value.map((part) =>
part.type === 'text' && isSerializedOffloadedToolResult(part.text)
? { ...part, text: EXPIRED_OFFLOADED_TOOL_RESULT_JSON }
: part,
),
}),
};
}
if (block.state === 'rejected' && isSerializedOffloadedToolResult(block.error)) {
return { ...block, error: EXPIRED_OFFLOADED_TOOL_RESULT_JSON };
}
}
if (block.type === 'text' && isSerializedOffloadedToolResult(block.text)) {
return { ...block, text: EXPIRED_OFFLOADED_TOOL_RESULT_JSON };
}
return { ...block };
});
return { ...message, content };
function replaceTextParts<T extends { type: string }>(parts: T[], replacement: string): T[] {
let replacedText = false;
return parts.flatMap((part) => {
if (part.type !== 'text') return [part];
if (replacedText) return [];
replacedText = true;
return [{ ...part, text: replacement }];
});
}

137
packages/@n8n/agents/src/types/runtime/agent-loop.ts Normal file
View File

@ -0,0 +1,137 @@
import type {
LanguageModel,
LanguageModelCallOptions,
ModelMessage,
Output,
SystemModelMessage,
TelemetryOptions,
ToolCallRepairFunction,
ToolSet,
} from 'ai';
import type { FinishReason, SerializableAgentState, TokenUsage } from '../index';
import type { ExecutionOptions, RunOptions } from '../sdk/agent';
import type { AgentMessage } from '../sdk/message';
import type { JSONObject } from '../utils/json';
import type { AgentMessageList } from '../../runtime/model/message-list';
import type { ToolCallBatchResult, ToolCallInput, ToolCallSuspension } from './tool-execution';
type RunCallOptions = (RunOptions & ExecutionOptions) | undefined;
/**
* Recognized ways a model turn can fail while the provider still returns
* "successfully" (no thrown SDK error). Extend this as new failure modes are
* recognized so consumers switch on `type` instead of probing separate fields.
* - `prompt_blocked`: the provider rejected the prompt (e.g. a safety block).
* - `no_output`: the model returned nothing with a non-stop finish reason.
*/
export type ModelTurnErrorType = 'prompt_blocked' | 'no_output';
/** A recognized model-turn failure: a `type` to branch on and a user-facing `message`. */
export interface ModelTurnError {
type: ModelTurnErrorType;
message: string;
}
/** Normalized result of a single LLM turn, independent of generate vs stream. */
export interface ModelTurnResult {
/** Raw AI SDK finish reason (used to detect the `tool-calls` continuation). */
aiFinishReason: string;
finishReason: FinishReason;
/** Normalized usage, including any cache-token details available from the provider. */
usage: TokenUsage | undefined;
newMessages: AgentMessage[];
toolCalls: ToolCallInput[];
/** Resolved structured output, only when an output spec is set and the turn finished. */
structuredOutput: unknown;
/**
* Set when the turn failed in a recognized way despite the provider returning
* without a thrown error (e.g. a prompt safety block, or empty output). The
* loop surfaces this instead of ending the run silently. Check this single
* field rather than probing per-failure-mode properties.
*/
errorReason?: ModelTurnError;
}
/** Per-iteration inputs for the LLM call, assembled by the shared loop. */
export interface ModelCallContext {
model: LanguageModel;
system: SystemModelMessage | SystemModelMessage[];
messages: ModelMessage[];
abortSignal: AbortSignal;
hasTools: boolean;
aiTools: ToolSet;
reasoning?: LanguageModelCallOptions['reasoning'];
providerOptions?: Record<string, JSONObject>;
outputSpec?: ReturnType<typeof Output.object>;
maxOutputTokens?: number;
onInputRejected?: (error: unknown) => Promise<void>;
aiSdkOptions: {
telemetry?: TelemetryOptions;
repairToolCall?: ToolCallRepairFunction<NoInfer<ToolSet>>;
};
}
/** Data needed to emit a terminal suspension result. */
export interface SuspendEmission {
suspendRunId: string;
list: AgentMessageList;
usage: TokenUsage | undefined;
suspensions: ToolCallSuspension[];
}
/** Data needed to emit a terminal completion result. */
export interface CompleteEmission {
list: AgentMessageList;
options: RunCallOptions;
finishReason: FinishReason;
usage: TokenUsage | undefined;
structuredOutput: unknown;
}
/**
* Runtime services the sinks call into for shared terminal concerns (cost,
* memory persistence, title generation, telemetry flush, state). Implemented by
* `AgentRuntime` and passed to each sink.
*/
export interface RunServices {
readonly runId: string;
readonly modelId: string;
applyCost(usage: TokenUsage | undefined): TokenUsage | undefined;
saveToMemory(list: AgentMessageList, options: RunCallOptions): Promise<void>;
maybeGenerateTitle(list: AgentMessageList, options: RunCallOptions): Promise<void>;
flushTelemetry(options: RunCallOptions): Promise<void>;
cleanupRun(): Promise<void>;
updateState(patch: Partial<SerializableAgentState>): void;
emitAgentEnd(messages: AgentMessage[]): void;
getState(): SerializableAgentState;
}
/**
* Strategy that adapts the shared agentic loop to a concrete output channel:
* `GenerateSink` accumulates a `GenerateResult`; `StreamSink` writes
* `StreamChunk`s through the stream writer guard.
*/
export interface RunOutputSink<TResult> {
/** Run one LLM turn. Streaming implementations also emit text/tool chunks here. */
callModel(ctx: ModelCallContext): Promise<ModelTurnResult>;
/**
* Report the run's cumulative token usage after each turn. Streaming
* implementations keep it so an aborted run can still emit a terminal
* finish chunk carrying the tokens consumed before the stop.
*/
reportUsage(usage: TokenUsage | undefined): void;
/**
* Signal that the just-returned turn's messages have been folded into the list.
* Streaming implementations drop the retained streamed text here so a later abort
* can't duplicate a turn that is already in the list — while a stop landing before
* this call still recovers the turn from the retained text.
*/
onTurnFolded?(): void;
/** Emit the results/errors of a completed tool-call batch. */
emitToolBatch(batch: ToolCallBatchResult): Promise<void>;
/** Produce the terminal result when the run suspends for HITL / suspend-resume. */
finishSuspended(emission: SuspendEmission): Promise<TResult>;
/** Produce the terminal result when the run completes normally. */
finishComplete(emission: CompleteEmission): Promise<TResult>;
}

96
packages/@n8n/agents/src/types/runtime/agent-runtime.ts Normal file
View File

@ -0,0 +1,96 @@
import type { ProviderOptions } from '@ai-sdk/provider-utils';
import type { JSONSchema7 } from 'json-schema';
import type { z } from 'zod';
import type { FetchFn } from '../../runtime/model/model-factory';
import type { ScopedMemoryTaskEvent } from '../../runtime/memory/scoped-memory-task-runner';
import type { AgentEventBus } from '../../runtime/state/event-bus';
import type { RunStateManager } from '../../runtime/state/run-state';
import type { ModelCost } from '../../sdk/catalog';
import type { RuntimeSkillSource } from '../../skills/types';
import type { WorkspaceFilesystem } from '../../workspace';
import type {
BuiltFileStore,
BuiltMemory,
BuiltProviderTool,
BuiltTelemetry,
BuiltTool,
CheckpointStore,
EpisodicMemoryConfig,
McpConnectionFailedEvent,
ObservationalMemoryConfig,
ObservationLogMemoryConfig,
ReasoningLevel,
ThinkingConfig,
TitleGenerationConfig,
} from '../index';
import type { AgentPersistenceOptions, ModelConfig, PromptCachingConfig } from '../sdk/agent';
export interface VolatileInstructionsContext {
persistence?: AgentPersistenceOptions;
}
export type VolatileInstructionsProvider = (
context: VolatileInstructionsContext,
) => Promise<string | undefined>;
export interface AgentRuntimeConfig {
name: string;
model: ModelConfig;
/**
* Proxy-aware `fetch` used for all model calls in this runtime (main model and
* title generation). When unset, model construction falls back to the ambient
* HTTP_PROXY resolver.
*/
modelFetch?: FetchFn;
instructions: string;
skillSource?: RuntimeSkillSource;
instructionProviderOptions?: ProviderOptions;
tools?: BuiltTool[];
deferredTools?: BuiltTool[];
workspaceFilesystem?: WorkspaceFilesystem;
toolSearch?: {
topK?: number;
};
providerTools?: BuiltProviderTool[];
memory?: BuiltMemory;
/** Host store resolving file-reference content parts to bytes before LLM calls. */
fileStore?: BuiltFileStore;
observationLog?: ObservationLogMemoryConfig;
observationalMemory?: ObservationalMemoryConfig;
episodicMemory?: EpisodicMemoryConfig;
structuredOutput?: z.ZodType | JSONSchema7;
checkpointStorage?: 'memory' | CheckpointStore;
thinking?: ThinkingConfig;
reasoning?: ReasoningLevel;
promptCaching?: PromptCachingConfig;
eventBus?: AgentEventBus;
/** Number of tool calls to execute concurrently. Default `1` (sequential). */
toolCallConcurrency?: number;
titleGeneration?: TitleGenerationConfig;
telemetry?: BuiltTelemetry;
/** Existing run id to continue, used when resuming a suspended run. */
runId?: string;
/**
* Pre-fetched model cost from the catalog. When provided, skips the per-run
* catalog fetch. Set once during Agent.build() and shared across per-run runtimes.
*/
modelCost?: ModelCost;
/**
* Shared RunStateManager for suspend/resume. When provided, per-run runtimes
* use the same store so resume() can find state from a prior run.
*/
runState?: RunStateManager;
/** Host callback for observational-memory background task lifecycle events. */
onMemoryTaskEvent?: (event: ScopedMemoryTaskEvent) => void;
/**
* Per-server MCP connection failures recorded during `Agent.build()` when
* resolving MCP tools. Tools from these servers were skipped; the runtime
* surfaces each as a non-fatal `warning` stream chunk at the start of a
* stream so hosts can tell the user an MCP server was unavailable without
* aborting the run.
*/
mcpConnectionFailures?: McpConnectionFailedEvent[];
/** The runtime loads these host instructions before each model call but does not save them. */
volatileInstructionsProvider?: VolatileInstructionsProvider;
}

301
packages/@n8n/agents/src/types/runtime/delegation.ts Normal file
View File

@ -0,0 +1,301 @@
import { z } from 'zod';
import type {
SubAgentTaskPath,
SubAgentTaskPathPolicy,
} from '../../runtime/tools/sub-agent-task-path';
import type {
AgentExecutionCounter,
FinishReason,
GenerateResult,
ModelConfig,
StreamChunk,
TokenUsage,
} from '../sdk/agent';
import type { BuiltProviderTool } from '../sdk/tool';
import type { BuiltTelemetry } from '../telemetry';
import type { JSONObject, JSONValue } from '../utils/json';
export const DELEGATE_SUB_AGENT_TOOL_NAME = 'delegate_subagent';
export const INLINE_SUB_AGENT_ID = 'inline';
/** i18n key — localized in the agent chat UI; see `agents.chat.delegate.childSuspendUnsupported`. */
export const DELEGATED_CHILD_SUSPEND_UNSUPPORTED_MESSAGE =
'agents.chat.delegate.childSuspendUnsupported';
export const INLINE_DELEGATE_SUB_AGENT_TOOL_METADATA_KEY = 'inlineDelegateSubAgent';
export const SUB_AGENT_TASK_DIFFICULTIES = ['low', 'medium', 'high'] as const;
const SubAgentTaskDifficultySchema = z.enum(SUB_AGENT_TASK_DIFFICULTIES);
export type SubAgentTaskDifficulty = z.infer<typeof SubAgentTaskDifficultySchema>;
const jsonValueSchema: z.ZodType<JSONValue> = z.lazy(() =>
z.union([
z.string(),
z.number(),
z.boolean(),
z.null(),
z.array(jsonValueSchema),
z.record(z.string(), jsonValueSchema),
]),
);
// Model-facing input: the arguments the LLM fills in when it calls the tool.
// The `.describe(...)` text is what the model reads, so keep it task-oriented.
export const delegateSubAgentInputSchema = z.object({
subAgentId: z
.string()
.min(1)
.describe(
'Required. Use "inline" for a one-off inline sub-agent. Use an exact configured sub-agent ID only when one is listed and fits the task.',
),
taskName: z
.string()
.min(1)
.describe('Short human-readable name for this delegated task, e.g. "research_api".'),
goal: z.string().min(1).describe('The concrete goal the sub-agent should accomplish.'),
context: z
.string()
.optional()
.describe(
'All details the child needs, since it sees nothing else: constraints, paths, data, prior decisions, acceptance criteria, and what you have already tried or ruled out.',
),
expectedOutput: z.string().optional().describe('The expected shape or contents of the answer.'),
difficulty: SubAgentTaskDifficultySchema.optional().describe(
'Optional difficulty estimate for this delegated task. Use low for simple bounded work, medium for moderate analysis or implementation, and high for complex research, architecture, or multi-step reasoning.',
),
});
// Documents the tool result shape for typing/introspection. Note: the handler's
// returned object (not this schema) is what is actually sent back to the model,
// so this is kept in sync with DelegateSubAgentToolOutput by hand.
export const delegateSubAgentOutputSchema = z.object({
status: z.enum(['completed', 'failed', 'suspended', 'cancelled']),
taskPath: z.string().optional(),
runId: z.string().optional(),
threadId: z.string().optional(),
model: z.string().optional(),
answer: z.string(),
structuredOutput: z.unknown().optional(),
usage: z
.object({
promptTokens: z.number().optional(),
completionTokens: z.number().optional(),
totalTokens: z.number().optional(),
cost: z.number().optional(),
})
.optional(),
finishReason: z.string().optional(),
error: z.string().optional(),
resumeContext: jsonValueSchema.optional(),
pendingSuspend: z
.array(
z.object({
runId: z.string(),
toolCallId: z.string(),
toolName: z.string(),
input: z.unknown(),
suspendPayload: z.unknown(),
resumeSchema: z.unknown().optional(),
}),
)
.optional(),
});
const delegateSubAgentContinuationSchema = z.object({
runId: z.string(),
toolCallId: z.string(),
taskPath: z.string(),
subAgentId: z.string(),
childCount: z.number().int().nonnegative(),
threadId: z.string().optional(),
resumeContext: jsonValueSchema.optional(),
});
export const delegateSubAgentSuspendSchema = z.unknown();
export const delegateSubAgentResumeSchema = z.unknown();
export type DelegateSubAgentContinuation = z.infer<typeof delegateSubAgentContinuationSchema>;
export function parseDelegateSubAgentContinuation(
value: unknown,
): DelegateSubAgentContinuation | undefined {
const parsed = delegateSubAgentContinuationSchema.safeParse(value);
return parsed.success ? parsed.data : undefined;
}
/** The arguments the LLM provides when calling delegate_subagent. */
export type DelegateSubAgentInput = z.infer<typeof delegateSubAgentInputSchema>;
/**
* Limits the delegate tool enforces structurally for a delegation: fan-out
* and the on/off switch (see {@link SubAgentTaskPathPolicy}).
*
* Per-run runtime constraints (e.g. a wall-clock timeout) are intentionally not
* here — they're a host concern, enforced inside the `runSubAgent` callback (as
* the n8n CLI runner does).
*/
export type DelegateSubAgentPolicy = SubAgentTaskPathPolicy;
/**
* What a host's `runSubAgent` callback receives: the model's
* {@link DelegateSubAgentInput} plus runtime-derived context the host needs to
* run and link the child. All `parent*` fields come from the parent's tool
* execution context and are used for tracing/linkage, not required to run.
*/
export interface DelegateSubAgentRequest extends DelegateSubAgentInput {
/** Direct child path assigned to this delegation (e.g. `/root/research_api_0`). */
taskPath: SubAgentTaskPath;
/** Parent run id (`ctx.runId`), e.g. for memory scoping / correlation. */
parentRunId?: string;
/** Parent's persisted memory thread id (`ctx.persistence.threadId`). */
parentThreadId?: string;
/** Parent's episodic-memory resource id (`ctx.persistence.resourceId`). */
parentResourceId?: string;
/** Opaque host metadata from the parent's persistence scope. */
parentHostMetadata?: JSONObject;
/** Parent's tool-call id that triggered this delegation. */
parentToolCallId?: string;
/**
* Parent run's abort signal (`ctx.abortSignal`). Forward it to the child so
* cancelling the parent run also cancels the delegated work.
*/
parentAbortSignal?: AbortSignal;
/** Parent aggregate execution counter (`ctx.executionCounter`) for inline child accounting. */
parentExecutionCounter?: AgentExecutionCounter;
/**
* Parent's live, resolved telemetry (`ctx.parentTelemetry`). Hosts derive the
* child's own telemetry from this (see `deriveSubAgentTelemetry`) so the
* sub-agent shares the parent's tracer and nests under the parent's
* delegate-tool-call span instead of starting a separate root trace.
*/
parentTelemetry?: BuiltTelemetry;
/** How many siblings the parent already spawned before this one (0-based). */
childCount: number;
/** Effective policy for this delegation. */
policy?: DelegateSubAgentPolicy;
}
/** The result a delegation returns to the parent model and to lifecycle events. */
export interface DelegateSubAgentToolOutput {
status: 'completed' | 'failed' | 'suspended' | 'cancelled';
/** Echoed back so consumers can correlate the result with the delegation. */
taskPath?: SubAgentTaskPath;
/** The child run's id, when the executor produced one. */
runId?: string;
/**
* The child run's memory thread id (`persistence.threadId`), when the
* executor used one. Surfaced so a consumer can correlate the child run or
* re-supply it to continue the same thread on a later delegation.
*/
threadId?: string;
/** Effective child model id used for this delegation (e.g. `anthropic/claude-haiku-4-5`). */
model?: string;
/** The child's answer — the main payload the parent acts on. */
answer: string;
structuredOutput?: unknown;
/** Child token usage + cost, surfaced so the parent can account for it. */
usage?: Pick<TokenUsage, 'promptTokens' | 'completionTokens' | 'totalTokens' | 'cost'>;
finishReason?: FinishReason;
/** Present when status is 'failed'. */
error?: string;
/** Host-owned, serializable context required to reconstruct this exact child. */
resumeContext?: JSONValue;
/** Present when status is 'suspended' — child run paused awaiting tool resume. */
pendingSuspend?: GenerateResult['pendingSuspend'];
}
/**
* Helpers passed to a host `runSubAgent` callback so the host can route
* `subAgentId: "inline"` while reusing the SDK inline child runner implementation.
*/
export interface DelegateSubAgentRunnerHelpers {
/** Run a one-off inline child using the parent agent's inherited local/deferred tool set. */
runInlineSubAgent: (request: DelegateSubAgentRequest) => Promise<DelegateSubAgentToolOutput>;
/**
* Forward a child stream chunk into the parent run's event bus (allowlisted
* types only, with a per-delegation character budget). No-op when the
* parent tool call id is missing.
*/
emitChunk: (chunk: StreamChunk) => void;
}
export type InlineSubAgentProviderToolsResolver = (
modelConfig: ModelConfig,
) => BuiltProviderTool[] | Promise<BuiltProviderTool[]>;
export type DelegateSubAgentRunner = (
request: DelegateSubAgentRequest,
helpers: DelegateSubAgentRunnerHelpers,
) => Promise<DelegateSubAgentToolOutput>;
export interface DelegateSubAgentCheckpointTarget {
childRunId: string;
childToolCallId: string;
childThreadId?: string;
resumeContext?: JSONValue;
}
export interface DelegateSubAgentResumeRequest
extends DelegateSubAgentRequest,
DelegateSubAgentCheckpointTarget {
resumeData: unknown;
}
export type DelegateSubAgentResumeRunner = (
request: DelegateSubAgentResumeRequest,
helpers: DelegateSubAgentRunnerHelpers,
) => Promise<DelegateSubAgentToolOutput>;
export interface DelegateSubAgentCancelRequest
extends DelegateSubAgentRequest,
DelegateSubAgentCheckpointTarget {
reason: string;
}
export type DelegateSubAgentCancelRunner = (
request: DelegateSubAgentCancelRequest,
helpers: DelegateSubAgentRunnerHelpers,
) => Promise<void>;
export interface CreateDelegateSubAgentToolOptions {
/** Model-facing tool name. Defaults to {@link DELEGATE_SUB_AGENT_TOOL_NAME}; the default description and system instruction adapt to it. */
name?: string;
/** Model-facing tool description. Anything other than a non-blank string falls back to the built-in text. */
description?: string | null;
/** System-prompt delegation guidance. Anything other than a non-blank string falls back to the built-in guidance. */
systemInstruction?: string | null;
/**
* Sub-agents the model may choose between. Listed in the system prompt; the
* model selects one by passing its id as `subAgentId`.
*/
availableSubAgents?: Array<{ id: string; name: string; useWhen?: string }>;
/** Parallelism limit for delegated child runs (also used as delegate_subagent batch size). */
policy?: DelegateSubAgentPolicy;
/** Additional local/deferred tool names the host removes from inline children. */
inlineSubAgentBlockedTools?: string[];
/**
* Resolved inline sub-agent models by task difficulty. Hosts map persisted config
* to runtime {@link ModelConfig} values before registering the delegate tool.
*/
inlineSubAgentModelsByDifficulty?: Partial<Record<SubAgentTaskDifficulty, ModelConfig>>;
/**
* Resolve provider-defined tools for an inline child after its model has been chosen.
* Inline children do not inherit parent provider tools.
*/
resolveInlineSubAgentProviderTools?: InlineSubAgentProviderToolsResolver;
/**
* Run the child for this delegation and return its result. When provided, the
* host receives every `subAgentId` (including `"inline"`) and may call
* `helpers.runInlineSubAgent` for inline work.
*/
runSubAgent?: DelegateSubAgentRunner;
/** Resume a child checkpoint previously cascaded through this delegate tool. */
resumeSubAgent?: DelegateSubAgentResumeRunner;
/** Return true only when a thrown child-resume error is safe to retry. */
shouldRetrySubAgentResumeError?: (error: unknown) => boolean;
/** Clean up a child checkpoint when the parent cancels the suspended delegation. */
cancelSubAgent?: DelegateSubAgentCancelRunner;
toModelOutput?: (output: z.infer<typeof delegateSubAgentOutputSchema>) => unknown;
}
export type DelegateSubAgentToolMetadata = CreateDelegateSubAgentToolOptions;

144
packages/@n8n/agents/src/types/runtime/tool-execution.ts Normal file
View File

@ -0,0 +1,144 @@
import type { JSONSchema7 } from 'json-schema';
import type { AgentMessageList } from '../../runtime/model/message-list';
import type { TokenCounter } from '../../runtime/model/model-token-counter';
import type { AgentEventBus } from '../../runtime/state/event-bus';
import type { RuntimeTelemetry } from '../../runtime/telemetry/runtime-telemetry';
import type { RuntimeSkillLoader } from '../../skills/types';
import type { WorkspaceFilesystem } from '../../workspace/types';
import type {
AgentExecutionCounter,
BuiltTelemetry,
BuiltTool,
PendingToolCall,
ToolSuspendOptions,
} from '../index';
import type { AgentPersistenceOptions, ToolResultEntry } from '../sdk/agent';
import type { AgentMessage } from '../sdk/message';
import type { JSONValue } from '../utils/json';
/** Pending tool calls from a suspended run, passed into the loop to execute before the first LLM call. */
export interface PendingResume {
pendingToolCalls: Record<string, PendingToolCall>;
/** The tool call being resumed with new data. */
resumeToolCallId: string;
resumeData: unknown;
}
export type ToolCallOutcome =
| {
outcome: 'success';
toolEntry: ToolResultEntry;
/**
* Output as the LLM sees it (after `toModelOutput`). Same as
* `toolEntry.output` when no `toModelOutput` transform is configured.
* Surfaced on the `tool-result` wire chunk so consumers see what the
* LLM saw (rather than the larger raw output).
*/
modelOutput: unknown;
customMessage?: AgentMessage;
mcpServerName?: string;
}
| {
outcome: 'suspended';
payload: unknown;
resumeSchema: JSONSchema7;
continuation?: JSONValue;
}
| {
outcome: 'cancelled';
toolEntry: ToolResultEntry;
modelOutput: string;
userMessage: string;
canceled: true;
}
| { outcome: 'retryable-error' }
| { outcome: 'error'; error: unknown }
| { outcome: 'noop' }; // tool call shouldn't be saved or logged anywhere, usually means that if was executed by AI SDK
export type ToolCallIdentity = Pick<PendingToolCall, 'toolCallId' | 'toolName' | 'input'>;
/** A tool call that completed successfully. */
export interface ToolCallSuccess extends ToolCallIdentity {
toolEntry: ToolResultEntry;
modelOutput: unknown;
customMessage?: AgentMessage;
/** Set when the tool belongs to an MCP server, so hosts can attribute the result to it. */
mcpServerName?: string;
}
/** Info about a tool call that suspended (before persistence — no runId yet). */
export interface ToolCallSuspension extends ToolCallIdentity {
payload: unknown;
/** JSON Schema describing the shape of resume data, derived from the tool's resumeSchema. */
resumeSchema: JSONSchema7;
}
/** Info about a tool call that failed — carries enough data for stream chunks. */
export interface ToolCallError extends ToolCallIdentity {
error: unknown;
}
/** Result of executing a batch of tool calls (before persistence). */
export interface ToolCallBatchResult {
results: ToolCallSuccess[];
suspensions: ToolCallSuspension[];
errors: ToolCallError[];
/** All items to persist: suspended tools (with suspendPayload) + unexecuted tools (without). */
pending: Record<string, PendingToolCall>;
}
export interface ToolCallInput {
toolCallId: string;
toolName: string;
input: unknown;
providerExecuted?: boolean;
}
/** Shared input for the tool-call batch iterators. */
export interface ToolBatchContext {
toolMap: Map<string, BuiltTool>;
list: AgentMessageList;
runId: string;
persistence?: AgentPersistenceOptions;
telemetry?: BuiltTelemetry;
executionCounter?: AgentExecutionCounter;
abortSignal: AbortSignal;
isAborted: () => boolean;
}
export interface ResumeToolBatchContext extends ToolBatchContext {
pendingResume: PendingResume;
}
/** Inputs for executing a single tool call. */
export interface ProcessToolCallParams extends ToolCallIdentity {
toolMap: Map<string, BuiltTool>;
list: AgentMessageList;
runId: string;
persistence?: AgentPersistenceOptions;
resumeData?: unknown;
resolvedTelemetry?: BuiltTelemetry;
executionCounter?: AgentExecutionCounter;
abortSignal?: AbortSignal;
/** Whether this counts as a new tool-call invocation. Default `true`; `false` on resume. */
countToolCall?: boolean;
/** Checkpointed suspend payload of the tool call being resumed. */
suspendPayload?: unknown;
/** Checkpointed private continuation of the tool call being resumed. */
continuation?: JSONValue;
/** Checkpointed resume schema of the tool call being resumed. */
resumeSchema?: ToolSuspendOptions['resumeSchema'];
}
export interface ToolCallExecutorDeps {
loadSkill?: RuntimeSkillLoader;
telemetry: RuntimeTelemetry;
eventBus: AgentEventBus;
/** Effective tool-call concurrency (default 1 = sequential). */
concurrency: number;
/** Invoked when a run is aborted mid-batch so the runtime can set cancelled state. */
onCancelled: () => void;
tokenCounter: TokenCounter;
workspaceFilesystem?: WorkspaceFilesystem;
}

2
packages/@n8n/agents/src/types/sdk/agent-builder.ts
View File

@ -4,7 +4,7 @@ import type { BuiltGuardrail } from './guardrail';
import type { CheckpointStore } from './memory';
import type { ReasoningLevel } from './reasoning';
import type { BuiltProviderTool, BuiltTool } from './tool';
import type { VolatileInstructionsProvider } from '../../runtime/loop/agent-runtime';
import type { VolatileInstructionsProvider } from '../runtime/agent-runtime';
import type { ScopedMemoryTaskEvent } from '../../runtime/memory/scoped-memory-task-runner';
import type { RuntimeSkill, RuntimeSkillSource } from '../../skills';