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

feat: Add streaming to Message an Agent workflows (#36734)

Browse Source
This commit is contained in:
Michael Drury 2026-08-21 09:11:07 +00:00 committed by GitHub
parent 63d8747070
commit 4f7fbc14d3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
32 changed files with 1368 additions and 114 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
packages/@n8n/api-types/src/agents/__tests__/sanitize-agent-tool-name.test.ts Normal file
View File

@ -0,0 +1,15 @@
import { sanitizeAgentToolName } from '../sanitize-agent-tool-name';
describe('sanitizeAgentToolName', () => {
it('leaves valid runtime names unchanged', () => {
expect(sanitizeAgentToolName('Lookup_Orders')).toBe('Lookup_Orders');
});
it('converts workflow display names to runtime-safe names', () => {
expect(sanitizeAgentToolName('Lookup Orders & Returns')).toBe('lookup-orders-returns');
});
it.each(['', '!!!', '🛠️'])('falls back when %j has no valid characters', (name) => {
expect(sanitizeAgentToolName(name)).toBe('tool');
});
});

1
packages/@n8n/api-types/src/agents/index.ts
View File

@ -10,6 +10,7 @@ export * from './agent-skill.schema';
export * from './child-trace';
export * from './inline-agent-config.schema';
export * from './sanitize-agent-json-config';
export * from './sanitize-agent-tool-name';
export * from './slack';
export * from './agent-task.schema';
export * from './dto';

14
packages/@n8n/api-types/src/agents/sanitize-agent-tool-name.ts Normal file
View File

@ -0,0 +1,14 @@
const VALID_AGENT_TOOL_NAME = /^[a-zA-Z0-9_-]{1,128}$/;
/** Converts a persisted display name into the name exposed to the model at runtime. */
export function sanitizeAgentToolName(name: string): string {
if (VALID_AGENT_TOOL_NAME.test(name)) return name;
const sanitizedName = name
.toLowerCase()
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-|-$/g, '')
.slice(0, 128);
return sanitizedName || 'tool';
}

21
packages/@n8n/api-types/src/push/execution.ts
View File

@ -54,6 +54,26 @@ export type ExecutionRecovered = {
};
};
export type AgentNodeCapability =
| { kind: 'tool'; name: string }
| { kind: 'skill'; id: string; name?: string }
| { kind: 'skill'; id?: never; name: string };
export type AgentNodeProgress = {
type: 'agentNodeProgress';
data: {
executionId: string;
nodeId: string;
nodeName: string;
runIndex: number;
itemIndex: number;
sequenceNumber: number;
toolCallId: string;
capability: AgentNodeCapability;
status: 'running' | 'succeeded' | 'failed';
};
};
export type NodeExecuteBefore = {
type: 'nodeExecuteBefore';
data: {
@ -124,6 +144,7 @@ export type ExecutionPushMessage =
| ExecutionWaiting
| ExecutionFinished
| ExecutionRecovered
| AgentNodeProgress
| NodeExecuteBefore
| NodeExecuteAfter
| NodeExecuteAfterData;

2
packages/@n8n/api-types/src/push/index.ts
View File

@ -11,6 +11,8 @@ import type { WorkerPushMessage } from './worker';
import type { WorkflowPushMessage } from './workflow';
import type { WorkflowReviewPushMessage } from './workflow-review';
export type { AgentNodeCapability, AgentNodeProgress } from './execution';
export type PushMessage =
| AgentsPushMessage
| ExecutionPushMessage

55
packages/cli/src/__tests__/workflow-execute-additional-data.test.ts
View File

@ -1644,6 +1644,61 @@ describe('WorkflowExecuteAdditionalData', () => {
});
});
it('bridges the invocation context to response chunks and editor progress', async () => {
const sendDataToUI = vi.fn();
const sendResponseChunk = vi.fn().mockResolvedValue(undefined);
const additionalData = mock<IWorkflowExecuteAdditionalData>({
userId: 'user-1',
projectId: 'project-1',
workflowId: 'workflow-1',
sendDataToUI,
});
await executeAgent(
{ agentId: AGENT_ID },
MESSAGE,
EXEC_ID,
THREAD_ID,
additionalData,
'webhook',
undefined,
undefined,
{
nodeId: 'node-1',
nodeName: 'Message an Agent',
runIndex: 2,
itemIndex: 3,
sendResponseChunk,
},
);
expect(agentWorkflowExecutionService.executeForWorkflow.mock.calls[0]?.[9]).toEqual(
executionSandboxScope,
);
const streamObserver = agentWorkflowExecutionService.executeForWorkflow.mock.calls[0]?.[10];
expect(streamObserver).toEqual(expect.any(Function));
await streamObserver?.({ type: 'response-delta', delta: 'hello' });
await streamObserver?.({
type: 'capability-start',
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
});
expect(sendResponseChunk).toHaveBeenCalledWith('item', 'hello');
expect(sendDataToUI).toHaveBeenCalledWith('agentNodeProgress', {
executionId: EXEC_ID,
nodeId: 'node-1',
nodeName: 'Message an Agent',
runIndex: 2,
itemIndex: 3,
sequenceNumber: 0,
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
status: 'running',
});
});
it('backfills projectId from the workflow owner when missing', async () => {
const additionalData = mock<IWorkflowExecuteAdditionalData>({
userId: 'user-1',

72
packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts
View File

@ -23,6 +23,7 @@ import type { Agent } from '../entities/agent.entity';
import type { NodeToolAiGatewayService } from '../json-config/node-tool-ai-gateway.service';
import type { AgentRepository } from '../repositories/agent.repository';
import type { ToolRegistry } from '../tool-registry';
import type { WorkflowAgentStreamObserver } from '../workflow-agent-stream';
const agentId = 'agent-1';
const projectId = 'project-1';
@ -658,6 +659,7 @@ describe('AgentWorkflowExecutionService', () => {
runExecutionData: { resultData: { runData: {} } } as unknown as IRunExecutionData,
};
Object.assign(runtime.agent, { tool: vi.fn(), declaredTools: [] });
const streamObserver = vi.fn<WorkflowAgentStreamObserver>().mockResolvedValue(undefined);
const result = await service.executeInlineForWorkflow(
inlinePayload,
@ -669,6 +671,7 @@ describe('AgentWorkflowExecutionService', () => {
'production',
undefined,
workflowContext,
streamObserver,
);
// No entity lookup — the embedded config is the source of truth.
@ -697,6 +700,11 @@ describe('AgentWorkflowExecutionService', () => {
);
expect(result.response).toBe('answer');
expect(streamObserver.mock.calls.map(([event]) => event)).toEqual([
{ type: 'response-begin' },
{ type: 'response-delta', delta: 'answer' },
{ type: 'response-end' },
]);
// Inline runs have no persisted session.
expect(result.session).toBeNull();
expect(executionService.startExecutionRecording).not.toHaveBeenCalled();
@ -1023,20 +1031,32 @@ describe('AgentWorkflowExecutionService', () => {
it('tracks a failed inline turn before rethrowing a stream reader error', async () => {
const { service, reconstructionService, telemetry } = makeService();
const runtime = makeRuntime();
const streamObserver = vi.fn<WorkflowAgentStreamObserver>().mockResolvedValue(undefined);
runtime.agent.stream.mockResolvedValue({
stream: makeFailingStream(new Error('reader failed while consuming stream')),
});
reconstructionService.reconstructFromResolvedSource.mockResolvedValue(runtime);
await expect(
service.executeInlineForWorkflow(
inlinePayload,
'hello',
'execution-1',
'thread-1',
projectId,
),
).rejects.toThrow('reader failed while consuming stream');
const execution = service.executeInlineForWorkflow(
inlinePayload,
'hello',
'execution-1',
'thread-1',
projectId,
undefined,
undefined,
undefined,
undefined,
streamObserver,
);
await expect(execution).rejects.toMatchObject({
message: 'reader failed while consuming stream',
description: 'reader failed while consuming stream',
});
expect(streamObserver.mock.calls.map(([event]) => event)).toEqual([
{ type: 'response-begin' },
{ type: 'response-delta', delta: 'partial answer' },
]);
// Telemetry fires even for failed runs — the rethrow happens after.
expect(telemetry.trackAgentTurnFinished).toHaveBeenCalledWith(
@ -1044,6 +1064,40 @@ describe('AgentWorkflowExecutionService', () => {
);
});
it('exposes a runtime stream error through the engine-owned error chunk', async () => {
const { service, reconstructionService } = makeService();
const runtime = makeRuntime([
{ type: 'text-start', id: 'text-1' },
{ type: 'text-delta', id: 'text-1', delta: 'partial answer' },
{ type: 'error', error: new Error('runtime stream failed') },
{ type: 'finish', finishReason: 'error' },
]);
const streamObserver = vi.fn<WorkflowAgentStreamObserver>().mockResolvedValue(undefined);
reconstructionService.reconstructFromResolvedSource.mockResolvedValue(runtime);
const execution = service.executeInlineForWorkflow(
inlinePayload,
'hello',
'execution-1',
'thread-1',
projectId,
undefined,
undefined,
undefined,
undefined,
streamObserver,
);
await expect(execution).rejects.toMatchObject({
message: 'Agent execution failed: runtime stream failed',
description: 'Agent execution failed: runtime stream failed',
});
expect(streamObserver.mock.calls.map(([event]) => event)).toEqual([
{ type: 'response-begin' },
{ type: 'response-delta', delta: 'partial answer' },
]);
});
it('surfaces inline compile failures as an OperationalError', async () => {
const { service, reconstructionService } = makeService();
reconstructionService.reconstructFromResolvedSource.mockRejectedValue(new Error('boom'));

296
packages/cli/src/modules/agents/__tests__/workflow-agent-stream.test.ts Normal file
View File

@ -0,0 +1,296 @@
import { SKILL_LOAD_TOOL_NAME, type StreamChunk } from '@n8n/agents';
import type { IWorkflowExecuteAdditionalData } from 'n8n-workflow';
import { mock } from 'vitest-mock-extended';
import {
createWorkflowAgentStreamObserver,
WorkflowAgentStreamAdapter,
type WorkflowAgentStreamObserver,
} from '../workflow-agent-stream';
function observerMock() {
return vi.fn<WorkflowAgentStreamObserver>().mockResolvedValue(undefined);
}
async function observeChunks(adapter: WorkflowAgentStreamAdapter, chunks: StreamChunk[]) {
for (const chunk of chunks) await adapter.observe(chunk);
}
describe('WorkflowAgentStreamAdapter', () => {
it('maps text segments and drops empty deltas', async () => {
const observer = observerMock();
const adapter = new WorkflowAgentStreamAdapter(observer);
await observeChunks(adapter, [
{ type: 'text-start', id: 'text-1' },
{ type: 'text-delta', id: 'text-1', delta: '' },
{ type: 'text-delta', id: 'text-1', delta: 'Hello' },
{ type: 'text-end', id: 'text-1' },
]);
expect(observer.mock.calls.map(([event]) => event)).toEqual([
{ type: 'response-begin' },
{ type: 'response-delta', delta: 'Hello' },
{ type: 'response-end' },
]);
});
it('synthesizes a begin and closes an open segment before a new text id', async () => {
const observer = observerMock();
const adapter = new WorkflowAgentStreamAdapter(observer);
await observeChunks(adapter, [
{ type: 'text-delta', id: 'text-1', delta: 'Before' },
{ type: 'text-delta', id: 'text-2', delta: 'After' },
{ type: 'finish', finishReason: 'stop' },
]);
expect(observer.mock.calls.map(([event]) => event)).toEqual([
{ type: 'response-begin' },
{ type: 'response-delta', delta: 'Before' },
{ type: 'response-end' },
{ type: 'response-begin' },
{ type: 'response-delta', delta: 'After' },
{ type: 'response-end' },
]);
});
it('abandons an open segment on failure without emitting a response error', async () => {
const observer = observerMock();
const adapter = new WorkflowAgentStreamAdapter(observer);
adapter.fail();
await adapter.observe({ type: 'text-start', id: 'text-1' });
adapter.fail();
adapter.fail();
await adapter.observe({ type: 'finish', finishReason: 'error' });
expect(observer.mock.calls.map(([event]) => event)).toEqual([{ type: 'response-begin' }]);
});
it('drops reasoning, tool-input, sub-agent, step, message, warning, and suspension events', async () => {
const observer = observerMock();
const adapter = new WorkflowAgentStreamAdapter(observer);
const ignoredChunks = [
{ type: 'reasoning-start', id: 'reason-1' },
{ type: 'reasoning-delta', id: 'reason-1', delta: 'private thought' },
{ type: 'reasoning-end', id: 'reason-1' },
{ type: 'tool-input-start', toolCallId: 'tc-1', toolName: 'lookup' },
{ type: 'tool-input-delta', toolCallId: 'tc-1', delta: 'secret input' },
{ type: 'subagent-started', taskName: 'child' },
{
type: 'subagent-chunk',
parentToolCallId: 'tc-child',
chunk: { type: 'text-delta', id: 'child-text', delta: 'child response' },
},
{ type: 'subagent-completed', taskName: 'child' },
{ type: 'start-step' },
{ type: 'finish-step' },
{ type: 'message', message: { role: 'assistant', content: 'private message' } },
{ type: 'warning', message: 'warning' },
{ type: 'tool-call-suspended', runId: 'run-1', toolCallId: 'tc-1' },
] as StreamChunk[];
await observeChunks(adapter, ignoredChunks);
expect(observer).not.toHaveBeenCalled();
});
it('tracks concurrent calls to the same tool by call id', async () => {
const observer = observerMock();
const adapter = new WorkflowAgentStreamAdapter(observer);
await observeChunks(adapter, [
{ type: 'tool-call', toolCallId: 'tc-1', toolName: 'lookup', input: { secret: 'one' } },
{ type: 'tool-call', toolCallId: 'tc-2', toolName: 'lookup', input: { secret: 'two' } },
{ type: 'tool-execution-start', toolCallId: 'tc-1', toolName: 'lookup', startTime: 1 },
{ type: 'tool-execution-start', toolCallId: 'tc-2', toolName: 'lookup', startTime: 2 },
{
type: 'tool-execution-end',
toolCallId: 'tc-1',
toolName: 'lookup',
isError: false,
endTime: 3,
},
{
type: 'tool-execution-end',
toolCallId: 'tc-2',
toolName: 'lookup',
isError: true,
endTime: 4,
},
]);
expect(observer.mock.calls.map(([event]) => event)).toEqual([
{
type: 'capability-start',
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
},
{
type: 'capability-start',
toolCallId: 'tc-2',
capability: { kind: 'tool', name: 'lookup' },
},
{
type: 'capability-end',
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
status: 'succeeded',
},
{
type: 'capability-end',
toolCallId: 'tc-2',
capability: { kind: 'tool', name: 'lookup' },
status: 'failed',
},
]);
});
it.each([
[
'by id',
{ skillId: 'skill-1', name: 'Research', filePath: 'references/api.md' },
{ kind: 'skill', id: 'skill-1', name: 'Research' },
],
['by name', { name: 'Research' }, { kind: 'skill', name: 'Research' }],
] as const)('resolves a loaded skill %s', async (_label, input, capability) => {
const observer = observerMock();
const adapter = new WorkflowAgentStreamAdapter(observer);
await observeChunks(adapter, [
{ type: 'tool-call', toolCallId: 'tc-skill', toolName: SKILL_LOAD_TOOL_NAME, input },
{
type: 'tool-execution-start',
toolCallId: 'tc-skill',
toolName: SKILL_LOAD_TOOL_NAME,
startTime: 1,
},
]);
expect(observer).toHaveBeenCalledWith({
type: 'capability-start',
toolCallId: 'tc-skill',
capability,
});
});
it('uses tool-result as a terminal fallback without emitting a duplicate end', async () => {
const observer = observerMock();
const adapter = new WorkflowAgentStreamAdapter(observer);
await observeChunks(adapter, [
{ type: 'tool-call', toolCallId: 'tc-1', toolName: 'lookup', input: {} },
{ type: 'tool-execution-start', toolCallId: 'tc-1', toolName: 'lookup', startTime: 1 },
{ type: 'tool-result', toolCallId: 'tc-1', toolName: 'lookup', output: { secret: true } },
{
type: 'tool-execution-end',
toolCallId: 'tc-1',
toolName: 'lookup',
isError: false,
endTime: 2,
},
]);
expect(observer.mock.calls.map(([event]) => event.type)).toEqual([
'capability-start',
'capability-end',
]);
expect(JSON.stringify(observer.mock.calls)).not.toContain('secret');
});
});
describe('createWorkflowAgentStreamObserver', () => {
const invocation = {
nodeId: 'node-1',
nodeName: 'Message an Agent',
runIndex: 2,
itemIndex: 3,
};
it('maps response events to the standard response chunk callback', async () => {
const sendResponseChunk = vi.fn().mockResolvedValue(undefined);
const observer = createWorkflowAgentStreamObserver({
additionalData: mock<IWorkflowExecuteAdditionalData>(),
executionId: 'exec-1',
invocation: { ...invocation, sendResponseChunk },
});
await observer({ type: 'response-begin' });
await observer({ type: 'response-delta', delta: 'hello' });
await observer({ type: 'response-end' });
expect(sendResponseChunk.mock.calls).toEqual([['begin'], ['item', 'hello'], ['end']]);
});
it('sends minimal ordered progress payloads without tool input or output', async () => {
const sendDataToUI = vi.fn();
const observer = createWorkflowAgentStreamObserver({
additionalData: mock<IWorkflowExecuteAdditionalData>({ sendDataToUI }),
executionId: 'exec-1',
invocation,
});
await observer({
type: 'capability-start',
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
});
await observer({
type: 'capability-end',
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
status: 'succeeded',
});
expect(sendDataToUI.mock.calls).toEqual([
[
'agentNodeProgress',
{
executionId: 'exec-1',
nodeId: 'node-1',
nodeName: 'Message an Agent',
runIndex: 2,
itemIndex: 3,
sequenceNumber: 0,
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
status: 'running',
},
],
[
'agentNodeProgress',
{
executionId: 'exec-1',
nodeId: 'node-1',
nodeName: 'Message an Agent',
runIndex: 2,
itemIndex: 3,
sequenceNumber: 1,
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
status: 'succeeded',
},
],
]);
expect(JSON.stringify(sendDataToUI.mock.calls)).not.toContain('input');
expect(JSON.stringify(sendDataToUI.mock.calls)).not.toContain('output');
});
it('does not require a response callback or an editor push connection', async () => {
const observer = createWorkflowAgentStreamObserver({
additionalData: mock<IWorkflowExecuteAdditionalData>(),
executionId: 'exec-1',
invocation,
});
await expect(observer({ type: 'response-delta', delta: 'ignored' })).resolves.toBeUndefined();
await expect(
observer({
type: 'capability-start',
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
}),
).resolves.toBeUndefined();
});
});

43
packages/cli/src/modules/agents/agent-workflow-execution.service.ts
View File

@ -47,6 +47,10 @@ import { getPublishedAgentSnapshot } from './utils/agent-published-snapshot';
import { streamAgentChunks } from './utils/agent-stream';
import { validateNodeToolConfigs, validateNodeToolExpressions } from './utils/node-tool-validation';
import { describeStructuredOutputError } from './utils/structured-output-error';
import {
WorkflowAgentStreamAdapter,
type WorkflowAgentStreamObserver,
} from './workflow-agent-stream';
interface WorkflowSandboxScope {
principalHash: AgentSandboxPrincipalHash;
@ -69,6 +73,13 @@ function getFinalWorkflowResponse(messageRecord: MessageRecord): string {
.join('');
}
function createWorkflowAgentExecutionError(message: string, cause?: Error): OperationalError {
return new OperationalError(message, {
description: message,
...(cause ? { cause } : {}),
});
}
/**
* Executes agents invoked from inside a workflow execution (the AI Agent node
* or a "Message an Agent" tool call): non-streaming runs against isolated
@ -91,12 +102,19 @@ export class AgentWorkflowExecutionService {
private normalizeWorkflowStreamError(error: unknown, outputSchema?: JSONSchema7): Error {
const normalizedError = error instanceof Error ? error : new Error(String(error));
if (!outputSchema || normalizedError instanceof OperationalError) return normalizedError;
if (!outputSchema || normalizedError instanceof OperationalError) {
if ('description' in normalizedError && typeof normalizedError.description === 'string') {
return normalizedError;
}
return createWorkflowAgentExecutionError(normalizedError.message, normalizedError);
}
const structuredOutputError = describeStructuredOutputError(normalizedError.message);
if (!structuredOutputError) return normalizedError;
if (!structuredOutputError) {
return createWorkflowAgentExecutionError(normalizedError.message, normalizedError);
}
return new OperationalError(structuredOutputError, { cause: normalizedError });
return createWorkflowAgentExecutionError(structuredOutputError, normalizedError);
}
/**
@ -249,6 +267,7 @@ export class AgentWorkflowExecutionService {
nodeName?: string;
};
recordingParams?: StartExecutionParams;
streamObserver?: WorkflowAgentStreamObserver;
sandboxScope?: { projectId: string; principalHash: AgentSandboxPrincipalHash };
}): Promise<WorkflowAgentRunOutcome> {
const {
@ -261,8 +280,10 @@ export class AgentWorkflowExecutionService {
outputSchema,
tracing,
recordingParams,
streamObserver,
sandboxScope,
} = params;
const streamAdapter = new WorkflowAgentStreamAdapter(streamObserver);
let agentExecutionId: string | undefined;
const recorder = new ExecutionRecorder(undefined, (timeline) => {
@ -347,6 +368,7 @@ export class AgentWorkflowExecutionService {
for await (const value of streamAgentChunks(resultStream.stream)) {
recorder.record(value);
await streamAdapter.observe(value);
if (value.type === 'tool-call') {
toolInputs.set(value.toolCallId, { toolName: value.toolName, input: value.input });
@ -367,6 +389,7 @@ export class AgentWorkflowExecutionService {
recorder.record({ type: 'error', error: normalizedError });
recorder.record({ type: 'finish', finishReason: 'error' });
streamError = normalizedError;
streamAdapter.fail();
}
};
@ -423,7 +446,7 @@ export class AgentWorkflowExecutionService {
}
if (recorder.suspended) {
throw new OperationalError(
throw createWorkflowAgentExecutionError(
'Agent execution suspended waiting for tool approval. ' +
'Suspend/resume is not supported in workflow execution context.',
);
@ -433,14 +456,14 @@ export class AgentWorkflowExecutionService {
if (outputSchema) {
const structuredOutputError = describeStructuredOutputError(messageRecord.error);
if (structuredOutputError) {
throw new OperationalError(structuredOutputError);
throw createWorkflowAgentExecutionError(structuredOutputError);
}
}
throw new OperationalError(`Agent execution failed: ${messageRecord.error}`);
throw createWorkflowAgentExecutionError(`Agent execution failed: ${messageRecord.error}`);
}
if (messageRecord.finishReason === 'error') {
throw new OperationalError(
throw createWorkflowAgentExecutionError(
outputSchema
? 'Agent execution finished with an error while producing structured output. ' +
"The agent's model or provider may not support JSON Schema structured output."
@ -475,6 +498,7 @@ export class AgentWorkflowExecutionService {
outputSchema?: JSONSchema7,
workflowContext?: ExecuteAgentWorkflowContext,
sandboxScope?: WorkflowSandboxScope,
streamObserver?: WorkflowAgentStreamObserver,
): Promise<ExecuteAgentData> {
return await this.executeForWorkflowInternal(
agentId,
@ -487,6 +511,7 @@ export class AgentWorkflowExecutionService {
outputSchema,
workflowContext,
sandboxScope,
streamObserver,
);
}
@ -501,6 +526,7 @@ export class AgentWorkflowExecutionService {
outputSchema?: JSONSchema7,
workflowContext?: ExecuteAgentWorkflowContext,
sandboxScope?: WorkflowSandboxScope,
streamObserver?: WorkflowAgentStreamObserver,
): Promise<ExecuteAgentData> {
const agentEntity = await this.agentRepository.findByIdAndProjectId(agentId, projectId);
if (!agentEntity) {
@ -558,6 +584,7 @@ export class AgentWorkflowExecutionService {
configuration: telemetryConfiguration,
},
},
streamObserver,
...(sandboxScope
? {
sandboxScope: {
@ -620,6 +647,7 @@ export class AgentWorkflowExecutionService {
runType: AgentRunTelemetryType = 'production',
outputSchema?: JSONSchema7,
workflowContext?: ExecuteAgentWorkflowContext,
streamObserver?: WorkflowAgentStreamObserver,
): Promise<ExecuteAgentData> {
const { config, skills } = await this.validateInlineAgentConfig(inlineAgent);
@ -694,6 +722,7 @@ export class AgentWorkflowExecutionService {
nodeId: workflowContext?.callingNodeId,
nodeName: workflowContext?.callingNodeName,
},
streamObserver,
});
// No `recordMessage` here: inline runs have no agent entity to attach a

17
packages/cli/src/modules/agents/json-config/agent-config-composition.ts
View File

@ -1,4 +1,5 @@
import type { AgentIntegrationConfig, AgentJsonConfig } from '@n8n/api-types';
export { sanitizeAgentToolName as sanitizeToolName } from '@n8n/api-types';
import type { Agent } from '../entities/agent.entity';
@ -26,19 +27,3 @@ export function decomposeJsonConfig(config: AgentJsonConfig): {
const { integrations, ...schemaConfig } = config;
return { schemaConfig, integrations: integrations ?? [] };
}
/**
* Coerce a display name into a tool name that satisfies the
* Anthropic/OpenAI constraint `^[a-zA-Z0-9_-]{1,128}$`. Applied at runtime
* only — the persisted `tool.name` keeps the user's original display string
* (e.g. "D&D Invite"), and the LLM-facing name is derived from it on the
* fly. Already-valid names pass through unchanged.
*/
export function sanitizeToolName(name: string): string {
if (/^[a-zA-Z0-9_-]{1,128}$/.test(name)) return name;
return name
.toLowerCase()
.replace(/[^a-z0-9]+/g, '-')
.replace(/^-|-$/g, '')
.slice(0, 128);
}

164
packages/cli/src/modules/agents/workflow-agent-stream.ts Normal file
View File

@ -0,0 +1,164 @@
import { SKILL_LOAD_TOOL_NAME, type StreamChunk } from '@n8n/agents';
import type { AgentNodeCapability, PushPayload } from '@n8n/api-types';
import type { ExecuteAgentInvocationContext, IWorkflowExecuteAdditionalData } from 'n8n-workflow';
type WorkflowAgentStreamEvent =
| { type: 'response-begin' }
| { type: 'response-delta'; delta: string }
| { type: 'response-end' }
| { type: 'capability-start'; toolCallId: string; capability: AgentNodeCapability }
| {
type: 'capability-end';
toolCallId: string;
capability: AgentNodeCapability;
status: 'succeeded' | 'failed';
};
export type WorkflowAgentStreamObserver = (event: WorkflowAgentStreamEvent) => Promise<void>;
type PendingTool = {
toolName: string;
input: unknown;
};
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === 'object' && value !== null && !Array.isArray(value);
}
function capabilityFor(toolName: string, input: unknown): AgentNodeCapability | undefined {
if (toolName !== SKILL_LOAD_TOOL_NAME) return { kind: 'tool', name: toolName };
if (!isRecord(input)) return undefined;
const id =
typeof input.skillId === 'string' && input.skillId.length > 0 ? input.skillId : undefined;
const name = typeof input.name === 'string' && input.name.length > 0 ? input.name : undefined;
if (id) return { kind: 'skill', id, ...(name ? { name } : {}) };
if (name) return { kind: 'skill', name };
return undefined;
}
export function createWorkflowAgentStreamObserver({
additionalData,
executionId,
invocation,
}: {
additionalData: IWorkflowExecuteAdditionalData;
executionId: string;
invocation: ExecuteAgentInvocationContext;
}): WorkflowAgentStreamObserver {
let sequenceNumber = 0;
return async (event) => {
switch (event.type) {
case 'response-begin':
await invocation.sendResponseChunk?.('begin');
return;
case 'response-delta':
await invocation.sendResponseChunk?.('item', event.delta);
return;
case 'response-end':
await invocation.sendResponseChunk?.('end');
return;
}
if (!additionalData.sendDataToUI) return;
const data: PushPayload<'agentNodeProgress'> = {
executionId,
nodeId: invocation.nodeId,
nodeName: invocation.nodeName,
runIndex: invocation.runIndex,
itemIndex: invocation.itemIndex,
sequenceNumber: sequenceNumber++,
toolCallId: event.toolCallId,
capability: event.capability,
status: event.type === 'capability-start' ? 'running' : event.status,
};
additionalData.sendDataToUI('agentNodeProgress', data);
};
}
export class WorkflowAgentStreamAdapter {
private openTextId: string | undefined;
private readonly pendingTools = new Map<string, PendingTool>();
private readonly runningCapabilities = new Map<string, AgentNodeCapability>();
constructor(private readonly observer?: WorkflowAgentStreamObserver) {}
async observe(chunk: StreamChunk): Promise<void> {
switch (chunk.type) {
case 'text-start':
await this.beginText(chunk.id);
return;
case 'text-delta':
if (chunk.delta.length === 0) return;
if (this.openTextId !== chunk.id) await this.beginText(chunk.id);
await this.emit({ type: 'response-delta', delta: chunk.delta });
return;
case 'text-end':
await this.endText();
return;
case 'finish':
await this.endText();
return;
case 'error':
this.fail();
return;
case 'tool-call':
this.pendingTools.set(chunk.toolCallId, {
toolName: chunk.toolName,
input: chunk.input,
});
return;
case 'tool-execution-start':
await this.startCapability(chunk.toolCallId, chunk.toolName);
return;
case 'tool-execution-end':
await this.endCapability(chunk.toolCallId, chunk.isError ? 'failed' : 'succeeded');
return;
case 'tool-result':
await this.endCapability(chunk.toolCallId, chunk.isError === true ? 'failed' : 'succeeded');
return;
}
}
fail(): void {
this.openTextId = undefined;
}
private async beginText(id: string): Promise<void> {
if (this.openTextId !== undefined) await this.endText();
this.openTextId = id;
await this.emit({ type: 'response-begin' });
}
private async endText(): Promise<void> {
if (this.openTextId === undefined) return;
this.openTextId = undefined;
await this.emit({ type: 'response-end' });
}
private async startCapability(toolCallId: string, toolName: string): Promise<void> {
const pending = this.pendingTools.get(toolCallId);
const capability = capabilityFor(toolName, pending?.input);
if (!capability) return;
this.runningCapabilities.set(toolCallId, capability);
await this.emit({ type: 'capability-start', toolCallId, capability });
}
private async endCapability(toolCallId: string, status: 'succeeded' | 'failed'): Promise<void> {
const capability = this.runningCapabilities.get(toolCallId);
if (!capability) return;
this.runningCapabilities.delete(toolCallId);
this.pendingTools.delete(toolCallId);
await this.emit({ type: 'capability-end', toolCallId, capability, status });
}
private async emit(event: WorkflowAgentStreamEvent): Promise<void> {
await this.observer?.(event);
}
}

19
packages/cli/src/workflow-execute-additional-data.ts
View File

@ -14,6 +14,7 @@ import type {
AiEvent,
EnvProviderState,
ExecuteAgentData,
ExecuteAgentInvocationContext,
ExecuteAgentSource,
ExecuteAgentWorkflowContext,
ExecuteWorkflowData,
@ -71,6 +72,10 @@ import { getWorkflowProjectDetailsSafe } from '@/workflows/utils';
import { WorkflowPublishedDataService } from '@/workflows/workflow-published-data.service';
import { RuntimeCredentialProxyService } from './services/runtime-credential-proxy.service';
import {
createWorkflowAgentStreamObserver,
type WorkflowAgentStreamObserver,
} from './modules/agents/workflow-agent-stream';
export function getRunData(
workflowData: IWorkflowBase,
@ -369,6 +374,7 @@ export async function executeAgent(
executionMode: WorkflowExecuteMode,
outputSchema?: JSONSchema7,
workflowContext?: ExecuteAgentWorkflowContext,
invocationContext?: ExecuteAgentInvocationContext,
): Promise<ExecuteAgentData> {
const telemetryUserId = additionalData.userId;
let projectId = additionalData.projectId;
@ -394,7 +400,16 @@ export async function executeAgent(
'@/modules/agents/agent-workflow-execution.service.js'
);
const agentWorkflowExecutionService = Container.get(AgentWorkflowExecutionService);
const streamObserver = invocationContext
? createWorkflowAgentStreamObserver({
additionalData,
executionId,
invocation: invocationContext,
})
: undefined;
const streamObserverArguments: [] | [WorkflowAgentStreamObserver] = streamObserver
? [streamObserver]
: [];
if (!additionalData.workflowId) {
throw new UnexpectedError('Cannot execute agent without a workflowId in additional data');
}
@ -412,6 +427,7 @@ export async function executeAgent(
isManualOrChatExecution(executionMode) ? 'test' : 'production',
outputSchema,
workflowContext,
...streamObserverArguments,
);
}
@ -445,6 +461,7 @@ export async function executeAgent(
outputSchema,
workflowContext,
sandboxScope,
...streamObserverArguments,
);
// Callers see the session id they supplied (or the derived per-call id), so

111
packages/core/src/execution-engine/node-execution-context/__tests__/execute-context.test.ts
View File

@ -11,7 +11,9 @@ import type {
INodeType,
INodeTypes,
ICredentialDataDecryptedObject,
ExecuteAgentInvocationContext,
WorkflowExpression,
IWorkflowBase,
} from 'n8n-workflow';
import {
UnexpectedError,
@ -21,7 +23,7 @@ import {
} from 'n8n-workflow';
import { mock } from 'vitest-mock-extended';
import type { ExecutionLifecycleHooks } from '@/execution-engine/execution-lifecycle-hooks';
import { ExecutionLifecycleHooks } from '@/execution-engine/execution-lifecycle-hooks';
import { describeCommonTests } from './shared-tests';
import { ExecuteContext } from '../execute-context';
@ -449,6 +451,40 @@ describe('ExecuteContext', () => {
[closeFn],
abortSignal,
);
const createStreamingAgentContext = (
contextRunIndex = runIndex,
contextInputData = inputData,
) => {
const hooks = new ExecutionLifecycleHooks('manual', 'exec-1', mock<IWorkflowBase>());
const sendChunkHandler = vi.fn();
hooks.addHandler('sendChunk', sendChunkHandler);
const executeAgentMock = vi.fn().mockResolvedValue({ response: 'ok' });
const additionalData = mock<IWorkflowExecuteAdditionalData>({
hooks,
rootExecutionMode: undefined,
streamingEnabled: true,
});
additionalData.executeAgent =
executeAgentMock as IWorkflowExecuteAdditionalData['executeAgent'];
return {
context: new ExecuteContext(
agentWorkflow,
node,
additionalData,
'manual',
runExecutionData,
contextRunIndex,
connectionInputData,
contextInputData,
executeData,
[closeFn],
abortSignal,
),
executeAgentMock,
sendChunkHandler,
};
};
it('passes the workflow context to additionalData.executeAgent', async () => {
agentAdditionalData.executeAgent = vi
@ -485,9 +521,77 @@ describe('ExecuteContext', () => {
],
runExecutionData,
},
{
nodeId: node.id,
nodeName: node.name,
runIndex,
itemIndex: 0,
},
);
});
it('passes a response chunk callback when workflow streaming is available', async () => {
const { context, executeAgentMock, sendChunkHandler } = createStreamingAgentContext(2, {
main: [[{ json: { idx: 0 } }, { json: { idx: 1 } }]],
});
await context.executeAgent({ agentId: 'agent-1' }, 'hello', 'exec-1', 1);
const invocationContext = executeAgentMock.mock
.calls[0]?.[8] as ExecuteAgentInvocationContext;
expect(invocationContext).toMatchObject({
nodeId: node.id,
nodeName: node.name,
runIndex: 2,
itemIndex: 1,
sendResponseChunk: expect.any(Function),
});
await invocationContext.sendResponseChunk?.('item', 'partial');
expect(sendChunkHandler).toHaveBeenCalledWith({
type: 'item',
content: 'partial',
metadata: {
nodeId: node.id,
nodeName: node.name,
runIndex: 2,
itemIndex: 1,
timestamp: expect.any(Number),
},
});
});
it('omits the response chunk callback when structured output is configured', async () => {
const { context, executeAgentMock } = createStreamingAgentContext();
await context.executeAgent(
{ agentId: 'agent-1', outputSchema: { type: 'object' } },
'hello',
'exec-1',
0,
);
const invocationContext = executeAgentMock.mock
.calls[0]?.[8] as ExecuteAgentInvocationContext;
expect(invocationContext).not.toHaveProperty('sendResponseChunk');
});
it('omits the response chunk callback when agent streaming is disabled', async () => {
const { context, executeAgentMock } = createStreamingAgentContext();
await context.executeAgent(
{ agentId: 'agent-1', enableStreaming: false },
'hello',
'exec-1',
0,
);
const invocationContext = executeAgentMock.mock
.calls[0]?.[8] as ExecuteAgentInvocationContext;
expect(invocationContext).not.toHaveProperty('sendResponseChunk');
});
it('passes all input items when inputDataScope is all', async () => {
agentAdditionalData.executeAgent = vi
.fn()
@ -513,6 +617,7 @@ describe('ExecuteContext', () => {
inputDataScope: 'all',
exposeWorkflowData: true,
}),
expect.objectContaining({ itemIndex: 0, nodeId: node.id, runIndex }),
);
});
@ -557,6 +662,7 @@ describe('ExecuteContext', () => {
'manual',
undefined,
expect.objectContaining({ inputData: [{ json: { idx: 1 } }], inputDataScope: 'item' }),
expect.objectContaining({ itemIndex: 1, nodeId: node.id, runIndex }),
);
});
@ -604,6 +710,7 @@ describe('ExecuteContext', () => {
inputData: [{ json: { idx: 0 } }, { json: { idx: 1 } }],
inputDataScope: 'all',
}),
expect.objectContaining({ itemIndex: 0, nodeId: node.id, runIndex }),
);
});
@ -651,6 +758,7 @@ describe('ExecuteContext', () => {
inputData: [{ json: { branch: 0 } }, { json: { branch: 2 } }],
inputDataScope: 'all',
}),
expect.objectContaining({ itemIndex: 0, nodeId: node.id, runIndex }),
);
});
@ -695,6 +803,7 @@ describe('ExecuteContext', () => {
'manual',
undefined,
expect.objectContaining({ inputData: [], inputDataScope: 'item' }),
expect.objectContaining({ itemIndex: 5, nodeId: node.id, runIndex }),
);
});
});

51
packages/core/src/execution-engine/node-execution-context/base-execute-context.ts
View File

@ -23,9 +23,12 @@ import type {
IWorkflowDataProxyData,
ISourceData,
AiEvent,
ChunkType,
NodeConnectionType,
IExecuteFunctions,
ExecuteAgentWorkflowContext,
IDataObject,
StructuredChunk,
} from 'n8n-workflow';
import {
UnexpectedError,
@ -241,6 +244,14 @@ export class BaseExecuteContext extends NodeExecutionContext {
runExecutionData: this.runExecutionData,
};
const sendResponseChunk =
agentInfo.enableStreaming !== false &&
agentInfo.outputSchema === undefined &&
this.isStreaming()
? async (type: 'begin' | 'item' | 'end' | 'error', content?: string) =>
await this.sendChunk(type, itemIndex, content)
: undefined;
return await this.additionalData.executeAgent(
source,
message,
@ -250,9 +261,49 @@ export class BaseExecuteContext extends NodeExecutionContext {
this.additionalData.rootExecutionMode ?? this.getMode(),
agentInfo.outputSchema,
workflowContext,
{
nodeId: this.node.id,
nodeName: this.node.name,
runIndex: this.runIndex,
itemIndex,
...(sendResponseChunk ? { sendResponseChunk } : {}),
},
);
}
isStreaming(): boolean {
const handlers = this.additionalData.hooks?.handlers?.sendChunk?.length;
const hasHandlers = handlers !== undefined && handlers > 0;
const streamingEnabled = this.additionalData.streamingEnabled === true;
const executionModeSupportsStreaming = ['manual', 'webhook', 'integrated', 'chat'];
return hasHandlers && executionModeSupportsStreaming.includes(this.mode) && streamingEnabled;
}
async sendChunk(
type: ChunkType,
itemIndex: number,
content?: IDataObject | string,
): Promise<void> {
const metadata = {
nodeId: this.node.id,
nodeName: this.node.name,
itemIndex,
runIndex: this.runIndex,
timestamp: Date.now(),
};
const parsedContent = typeof content === 'string' ? content : JSON.stringify(content);
const message: StructuredChunk = {
type,
content: parsedContent,
metadata,
};
await this.additionalData.hooks?.runHook('sendChunk', [message]);
}
async getExecutionDataById(executionId: string): Promise<IRunExecutionData | undefined> {
return await this.additionalData.getRunExecutionData(executionId);
}

42
packages/core/src/execution-engine/node-execution-context/execute-context.ts
View File

@ -2,7 +2,6 @@ import { createDeferredPromise } from '@n8n/utils/promise/deferred-promise';
import type {
AINodeConnectionType,
CallbackManager,
ChunkType,
CloseFunction,
IDataObject,
IExecuteData,
@ -15,7 +14,6 @@ import type {
ITaskDataConnections,
IWorkflowExecuteAdditionalData,
NodeExecutionHint,
StructuredChunk,
Workflow,
WorkflowExecuteMode,
EngineResponse,
@ -141,46 +139,6 @@ export class ExecuteContext extends BaseExecuteContext implements IExecuteFuncti
return await this.additionalData.getRuntimeCredential(this.runExecutionData, alias);
}
isStreaming(): boolean {
// Check if we have sendChunk handlers
const handlers = this.additionalData.hooks?.handlers?.sendChunk?.length;
const hasHandlers = handlers !== undefined && handlers > 0;
// Check if streaming was enabled for this execution
const streamingEnabled = this.additionalData.streamingEnabled === true;
// Check current execution mode supports streaming
const executionModeSupportsStreaming = ['manual', 'webhook', 'integrated', 'chat'];
const isStreamingMode = executionModeSupportsStreaming.includes(this.mode);
return hasHandlers && isStreamingMode && streamingEnabled;
}
async sendChunk(
type: ChunkType,
itemIndex: number,
content?: IDataObject | string,
): Promise<void> {
const node = this.getNode();
const metadata = {
nodeId: node.id,
nodeName: node.name,
itemIndex,
runIndex: this.runIndex,
timestamp: Date.now(),
};
const parsedContent = typeof content === 'string' ? content : JSON.stringify(content);
const message: StructuredChunk = {
type,
content: parsedContent,
metadata,
};
await this.additionalData.hooks?.runHook('sendChunk', [message]);
}
async getInputConnectionData(
connectionType: AINodeConnectionType,
itemIndex: number,

10
packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/agentNodeProgress.ts Normal file
View File

@ -0,0 +1,10 @@
import type { AgentNodeProgress } from '@n8n/api-types';
import { useWorkflowExecutionStateStore } from '@/app/stores/workflowExecutionState.store';
import type { PushHandlerOptions } from './types';
export async function agentNodeProgress(
event: AgentNodeProgress,
{ documentId }: PushHandlerOptions,
) {
useWorkflowExecutionStateStore(documentId).handleAgentNodeProgress(event);
}

1
packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/index.ts
View File

@ -1,3 +1,4 @@
export * from './agentNodeProgress';
export * from './builderCreditsUpdated';
export * from './executionFinished';
export * from './executionRecovered';

2
packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/nodeExecuteAfter.test.ts
View File

@ -59,6 +59,7 @@ describe('nodeExecuteAfter', () => {
it('should update node execution data with placeholder and remove executing node', async () => {
const assistantStore = useAssistantStore();
const clearAgentNodeProgress = vi.spyOn(workflowExecutionStateStore, 'clearAgentNodeProgress');
const event: NodeExecuteAfter = {
type: 'nodeExecuteAfter',
@ -82,6 +83,7 @@ describe('nodeExecuteAfter', () => {
expect(workflowExecutionStateStore.executingNode.removeExecutingNode).toHaveBeenCalledWith(
'Test Node',
);
expect(clearAgentNodeProgress).toHaveBeenCalledWith('Test Node');
expect(assistantStore.onNodeExecution).toHaveBeenCalledTimes(1);
expect(assistantStore.onNodeExecution).toHaveBeenCalledWith(event.data);

1
packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/nodeExecuteAfter.ts
View File

@ -66,6 +66,7 @@ export async function nodeExecuteAfter(
);
workflowExecutionStateStore.executingNode.removeExecutingNode(pushData.nodeName);
workflowExecutionStateStore.clearAgentNodeProgress(pushData.nodeName);
// Side effects
if (pushData.data.executionStatus === 'waiting' && pushData.data.metadata?.resumeFormUrl) {

28
packages/frontend/editor-ui/src/app/composables/usePushConnection/usePushConnection.test.ts
View File

@ -3,10 +3,11 @@ import {
testWebhookReceived,
builderCreditsUpdated,
executionStarted,
agentNodeProgress,
} from '@/app/composables/usePushConnection/handlers';
import type { TestWebhookReceived } from '@n8n/api-types/push/webhook';
import type { BuilderCreditsPushMessage } from '@n8n/api-types/push/builder-credits';
import type { PushMessage } from '@n8n/api-types';
import type { AgentNodeProgress, PushMessage } from '@n8n/api-types';
import { pushHandlerRegistry } from '@n8n/frontend-module-sdk';
import { useRouter } from 'vue-router';
import type { OnPushMessageHandler } from '@/app/stores/pushConnection.store';
@ -43,6 +44,7 @@ vi.mock('@/app/composables/usePushConnection/handlers', () => ({
workflowDeactivated: vi.fn(),
collaboratorsChanged: vi.fn(),
builderCreditsUpdated: vi.fn(),
agentNodeProgress: vi.fn(),
}));
vi.mock('vue-router', async (importOriginal) => {
@ -128,6 +130,30 @@ describe('usePushConnection composable', () => {
expect(removeEventListener).toHaveBeenCalledTimes(1);
});
it('routes agent capability progress to its execution-scoped handler', async () => {
pushConnection.initialize();
const handler = addEventListener.mock.calls[0][0];
const event: AgentNodeProgress = {
type: 'agentNodeProgress',
data: {
executionId: 'exec-1',
nodeId: 'node-1',
nodeName: 'Message an Agent',
runIndex: 0,
itemIndex: 0,
sequenceNumber: 0,
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
status: 'running',
},
};
handler(event);
await Promise.resolve();
expect(agentNodeProgress).toHaveBeenCalledWith(event, expect.any(Object));
});
it('should handle updateBuilderCredits event correctly', async () => {
pushConnection.initialize();

3
packages/frontend/editor-ui/src/app/composables/usePushConnection/usePushConnection.ts
View File

@ -24,6 +24,7 @@ import {
workflowDeactivated,
workflowAutoDeactivated,
workflowSettingsUpdated,
agentNodeProgress,
} from '@/app/composables/usePushConnection/handlers';
import type { PushHandlerOptions } from '@/app/composables/usePushConnection/handlers/types';
import { injectWorkflowDocumentStore } from '@/app/stores/workflowDocument.store';
@ -76,6 +77,8 @@ export function usePushConnection({ router }: { router: ReturnType<typeof useRou
}
switch (event.type) {
case 'agentNodeProgress':
return await agentNodeProgress(event, options);
case 'testWebhookDeleted':
return await testWebhookDeleted(event, options);
case 'testWebhookReceived':

3
packages/frontend/editor-ui/src/app/constants/durations.ts
View File

@ -44,6 +44,9 @@ export const AGENT_EVAL_RUN_POLL_MAX_ERRORS = 3;
export const LOADING_ANIMATION_MIN_DURATION = 1000;
/** Keeps short agent tool and skill calls perceptible without adding continuous motion. */
export const AGENT_CAPABILITY_ACTIVE_MIN_DURATION = 300;
/** Hover-intent delays for reveal-on-hover affordances (e.g. a collapsed group's description). */
export const HOVER_DELAY = {
/** Delay before a hovered affordance reveals its content. */

130
packages/frontend/editor-ui/src/app/stores/workflowExecutionState.store.test.ts
View File

@ -6,7 +6,8 @@
* executions filter+dedupe, last-successful-execution via id reference,
* dispose lifecycle.
*/
import { describe, it, expect, vi, beforeEach } from 'vitest';
import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
import type { AgentNodeProgress } from '@n8n/api-types';
import { setActivePinia, createPinia, getActivePinia } from 'pinia';
import { defineComponent, provide, shallowRef } from 'vue';
import { createComponentRenderer } from '@/__tests__/render';
@ -53,6 +54,24 @@ function makeExecutionSummary(overrides: Partial<ExecutionSummary> = {}): Execut
} as ExecutionSummary;
}
function makeAgentProgress(overrides: Partial<AgentNodeProgress['data']> = {}): AgentNodeProgress {
return {
type: 'agentNodeProgress',
data: {
executionId: 'exec-1',
nodeId: 'node-1',
nodeName: 'Message an Agent',
runIndex: 0,
itemIndex: 0,
sequenceNumber: 0,
toolCallId: 'tc-1',
capability: { kind: 'tool', name: 'lookup' },
status: 'running',
...overrides,
},
};
}
describe('workflowExecutionState.store', () => {
beforeEach(() => {
setActivePinia(createPinia());
@ -1558,6 +1577,115 @@ describe('workflowExecutionState.store', () => {
});
});
describe('agent capability progress', () => {
let store: ReturnType<typeof useWorkflowExecutionStateStore>;
beforeEach(() => {
vi.useFakeTimers();
vi.setSystemTime(new Date('2026-01-01T00:00:00Z'));
store = useWorkflowExecutionStateStore(createWorkflowDocumentId('wf-agent-progress'));
store.setActiveExecutionId('exec-1');
});
afterEach(() => {
vi.useRealTimers();
});
it('ignores events for another execution and stale sequence numbers', () => {
store.handleAgentNodeProgress(makeAgentProgress({ executionId: 'exec-other' }));
expect(store.activeAgentCapabilityKeysByNodeId.size).toBe(0);
store.handleAgentNodeProgress(makeAgentProgress({ sequenceNumber: 1 }));
store.handleAgentNodeProgress(makeAgentProgress({ sequenceNumber: 2, status: 'succeeded' }));
store.handleAgentNodeProgress(makeAgentProgress({ sequenceNumber: 1, status: 'running' }));
vi.advanceTimersByTime(300);
expect(store.activeAgentCapabilityKeysByNodeId.size).toBe(0);
});
it('keeps fast calls visible for at least 300 ms', () => {
store.handleAgentNodeProgress(makeAgentProgress());
vi.advanceTimersByTime(100);
store.handleAgentNodeProgress(makeAgentProgress({ sequenceNumber: 1, status: 'succeeded' }));
vi.advanceTimersByTime(199);
expect(store.activeAgentCapabilityKeysByNodeId.get('node-1')).toEqual(
new Set(['tool:lookup']),
);
vi.advanceTimersByTime(1);
expect(store.activeAgentCapabilityKeysByNodeId.size).toBe(0);
});
it('keeps a tool active until every concurrent call finishes', () => {
store.handleAgentNodeProgress(makeAgentProgress({ toolCallId: 'tc-1' }));
store.handleAgentNodeProgress(makeAgentProgress({ toolCallId: 'tc-2', sequenceNumber: 1 }));
store.handleAgentNodeProgress(
makeAgentProgress({ toolCallId: 'tc-1', sequenceNumber: 2, status: 'succeeded' }),
);
vi.advanceTimersByTime(300);
expect(store.activeAgentCapabilityKeysByNodeId.get('node-1')).toEqual(
new Set(['tool:lookup']),
);
store.handleAgentNodeProgress(
makeAgentProgress({ toolCallId: 'tc-2', sequenceNumber: 3, status: 'failed' }),
);
expect(store.activeAgentCapabilityKeysByNodeId.size).toBe(0);
});
it('accepts reordered terminal events for different concurrent calls', () => {
store.handleAgentNodeProgress(makeAgentProgress({ toolCallId: 'tc-1' }));
store.handleAgentNodeProgress(makeAgentProgress({ toolCallId: 'tc-2', sequenceNumber: 1 }));
store.handleAgentNodeProgress(
makeAgentProgress({ toolCallId: 'tc-2', sequenceNumber: 3, status: 'succeeded' }),
);
store.handleAgentNodeProgress(
makeAgentProgress({ toolCallId: 'tc-1', sequenceNumber: 2, status: 'succeeded' }),
);
vi.advanceTimersByTime(300);
expect(store.activeAgentCapabilityKeysByNodeId.size).toBe(0);
});
it('prefers skill id and falls back to skill name', () => {
store.handleAgentNodeProgress(
makeAgentProgress({
toolCallId: 'tc-id',
capability: { kind: 'skill', id: 'skill-1', name: 'Research' },
}),
);
store.handleAgentNodeProgress(
makeAgentProgress({
toolCallId: 'tc-name',
sequenceNumber: 1,
capability: { kind: 'skill', name: 'Writing' },
}),
);
expect(store.activeAgentCapabilityKeysByNodeId.get('node-1')).toEqual(
new Set(['skill:id:skill-1', 'skill:name:Writing']),
);
});
it('clears progress immediately for node and execution lifecycle cleanup', () => {
store.handleAgentNodeProgress(makeAgentProgress());
store.clearAgentNodeProgress('Message an Agent');
expect(store.activeAgentCapabilityKeysByNodeId.size).toBe(0);
store.handleAgentNodeProgress(makeAgentProgress({ sequenceNumber: 1 }));
store.resetExecutionState();
expect(store.activeAgentCapabilityKeysByNodeId.size).toBe(0);
store.setActiveExecutionId('exec-1');
store.handleAgentNodeProgress(makeAgentProgress());
store.markExecutionAsStopped();
expect(store.activeAgentCapabilityKeysByNodeId.size).toBe(0);
});
});
describe('event hook', () => {
it('fires onWorkflowExecutionStateChange with workflowId and field discriminator', () => {
const id = createWorkflowDocumentId('wf-1');

106
packages/frontend/editor-ui/src/app/stores/workflowExecutionState.store.ts
View File

@ -20,6 +20,7 @@ import type {
ITaskStartedData,
} from 'n8n-workflow';
import type { NodeExecuteBefore } from '@n8n/api-types/push/execution';
import type { AgentNodeCapability, AgentNodeProgress } from '@n8n/api-types';
import type {
IExecutionResponse,
IExecutionsStopData,
@ -51,6 +52,11 @@ import type {
NodesSetPayload,
} from './workflowDocument/useWorkflowDocumentNodes';
import type { ExecutionOutputMap } from '@/app/types/executionData';
import { AGENT_CAPABILITY_ACTIVE_MIN_DURATION } from '@/app/constants/durations';
import {
capabilityActivityKeys,
type AgentCapabilityActivityKey,
} from '@/features/agents/utils/agentCapabilityActivity';
const EMPTY_EXECUTION_ISSUES_BY_NODE_NAME = new Map<string, ComputedRef<string[]>>();
const EMPTY_EXECUTION_PIN_DATA_BY_NODE_NAME: IPinData = {};
@ -157,6 +163,100 @@ export function useWorkflowExecutionStateStore(id: WorkflowDocumentId) {
* intentionally not wired into the change-event mechanism below.
*/
const executingNode = useExecutingNode();
const activeAgentCapabilityCalls = shallowReactive(
new Map<
string,
{
nodeId: string;
nodeName: string;
capability: AgentNodeCapability;
startedAt: number;
}
>(),
);
const latestAgentProgressByCapabilityCall = new Map<
string,
{ nodeName: string; sequenceNumber: number }
>();
const agentCapabilityRemovalTimers = new Map<string, ReturnType<typeof setTimeout>>();
const activeAgentCapabilityKeysByNodeId = computed(() => {
const keysByNodeId = new Map<string, Set<AgentCapabilityActivityKey>>();
for (const call of activeAgentCapabilityCalls.values()) {
const keys = keysByNodeId.get(call.nodeId) ?? new Set<AgentCapabilityActivityKey>();
for (const key of capabilityActivityKeys(call.capability)) keys.add(key);
keysByNodeId.set(call.nodeId, keys);
}
return keysByNodeId;
});
function capabilityCallKey(data: AgentNodeProgress['data']): string {
return `${data.executionId}:${data.nodeId}:${data.runIndex}:${data.itemIndex}:${data.toolCallId}`;
}
function removeAgentCapabilityCall(key: string) {
const timer = agentCapabilityRemovalTimers.get(key);
if (timer) clearTimeout(timer);
agentCapabilityRemovalTimers.delete(key);
activeAgentCapabilityCalls.delete(key);
}
function clearAgentNodeProgress(nodeName: string) {
for (const [key, call] of activeAgentCapabilityCalls) {
if (call.nodeName === nodeName) removeAgentCapabilityCall(key);
}
for (const [key, call] of latestAgentProgressByCapabilityCall) {
if (call.nodeName === nodeName) latestAgentProgressByCapabilityCall.delete(key);
}
}
function clearAgentProgress() {
for (const timer of agentCapabilityRemovalTimers.values()) clearTimeout(timer);
agentCapabilityRemovalTimers.clear();
activeAgentCapabilityCalls.clear();
latestAgentProgressByCapabilityCall.clear();
}
function handleAgentNodeProgress({ data }: AgentNodeProgress) {
if (activeExecutionId.value !== data.executionId) return;
const callKey = capabilityCallKey(data);
const latest = latestAgentProgressByCapabilityCall.get(callKey);
if (latest && data.sequenceNumber <= latest.sequenceNumber) return;
latestAgentProgressByCapabilityCall.set(callKey, {
nodeName: data.nodeName,
sequenceNumber: data.sequenceNumber,
});
if (data.status === 'running') {
const existing = activeAgentCapabilityCalls.get(callKey);
const removalTimer = agentCapabilityRemovalTimers.get(callKey);
if (removalTimer) clearTimeout(removalTimer);
agentCapabilityRemovalTimers.delete(callKey);
activeAgentCapabilityCalls.set(callKey, {
nodeId: data.nodeId,
nodeName: data.nodeName,
capability: data.capability,
startedAt: existing?.startedAt ?? Date.now(),
});
return;
}
const call = activeAgentCapabilityCalls.get(callKey);
if (!call) return;
const remainingDuration = Math.max(
0,
AGENT_CAPABILITY_ACTIVE_MIN_DURATION - (Date.now() - call.startedAt),
);
if (remainingDuration === 0) {
removeAgentCapabilityCall(callKey);
return;
}
agentCapabilityRemovalTimers.set(
callKey,
setTimeout(() => removeAgentCapabilityCall(callKey), remainingDuration),
);
}
const onWorkflowExecutionStateChange = createEventHook<WorkflowExecutionStateChangeEvent>();
@ -477,6 +577,7 @@ export function useWorkflowExecutionStateStore(id: WorkflowDocumentId) {
runningScopes.clear();
executionRunningByNodeId.clear();
executionWaitingForNextByNodeId.clear();
clearAgentProgress();
});
/**
@ -502,6 +603,7 @@ export function useWorkflowExecutionStateStore(id: WorkflowDocumentId) {
// --- Write API ---
function setActiveExecutionId(value: string | null | undefined) {
if (activeExecutionId.value !== value) clearAgentProgress();
// When transitioning to a real execution id while a pending scaffold
// is staged (e.g. REST response arrives before executionStarted push),
// migrate the scaffold into the id-keyed executionData store so the
@ -832,6 +934,7 @@ export function useWorkflowExecutionStateStore(id: WorkflowDocumentId) {
lastSuccessfulExecutionId.value = null;
stoppedExecutionId.value = null;
executingNode.clearNodeExecutionQueue();
clearAgentProgress();
fireChange(CHANGE_ACTION.DELETE, 'state');
}
@ -921,6 +1024,7 @@ export function useWorkflowExecutionStateStore(id: WorkflowDocumentId) {
activeExecutionRunDataByNodeId,
activeExecutionRunDataOutputMapByNodeId,
activeExecutionWaitingByNodeId,
activeAgentCapabilityKeysByNodeId,
executionRunningByNodeId,
executionWaitingForNextByNodeId,
resolveExecutionTriggerNodeName,
@ -953,6 +1057,8 @@ export function useWorkflowExecutionStateStore(id: WorkflowDocumentId) {
setActiveExecutionRunData,
clearActiveExecutionStartedData,
addActiveNodeExecutionStartedData,
handleAgentNodeProgress,
clearAgentNodeProgress,
renameActiveExecutionNode,
resetExecutionState,
markExecutionAsStopped,

23
packages/frontend/editor-ui/src/features/agents/utils/agentCapabilityActivity.ts Normal file
View File

@ -0,0 +1,23 @@
import type { AgentNodeCapability } from '@n8n/api-types';
export type AgentCapabilityActivityKey = string;
export function toolActivityKey(name: string): AgentCapabilityActivityKey {
return `tool:${name}`;
}
export function skillIdActivityKey(id: string): AgentCapabilityActivityKey {
return `skill:id:${id}`;
}
export function skillNameActivityKey(name: string): AgentCapabilityActivityKey {
return `skill:name:${name}`;
}
export function capabilityActivityKeys(
capability: AgentNodeCapability,
): AgentCapabilityActivityKey[] {
if (capability.kind === 'tool') return [toolActivityKey(capability.name)];
if (capability.id !== undefined) return [skillIdActivityKey(capability.id)];
return [skillNameActivityKey(capability.name)];
}

13
packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/CanvasNodeAgent.vue
View File

@ -20,6 +20,7 @@ import CanvasNodeAgentChips from './parts/CanvasNodeAgentChips.vue';
import { buildAgentCardChips } from './parts/canvasNodeAgentChips.utils';
import { useAgentNavigation } from '@/features/agents/composables/useAgentNavigation';
import { AGENT_NODE_SIZE } from '@/app/utils/nodeViewUtils';
import { injectWorkflowExecutionStateStore } from '@/app/stores/workflowExecutionState.store';
// Width comes from the shared constant so canvas placement and tidy-up layout
// stay in sync with the rendered card.
@ -35,6 +36,7 @@ const $style = useCssModule();
const i18n = useI18n();
const nodeTypesStore = useNodeTypesStore();
const nav = useAgentNavigation();
const workflowExecutionStateStore = injectWorkflowExecutionStateStore();
const { catalog: modelCatalog, ensureLoaded: ensureModelsLoaded } = useModelCatalog();
const {
@ -131,6 +133,11 @@ function resolveNodeTypeLabel(nodeType: string, version?: number): string | unde
const chips = computed(() =>
summary.value ? buildAgentCardChips(summary.value, resolveNodeTypeLabel) : [],
);
const activeCapabilityKeys = computed(
() =>
workflowExecutionStateStore.value.activeAgentCapabilityKeysByNodeId.get(id.value) ??
new Set<string>(),
);
// The picker is NDV-parameter-input shaped; it only reads `parameter.name`, so a
// minimal synthesized definition matching the node's `agentId` property is enough.
@ -230,7 +237,11 @@ watch(
{{ modelName || i18n.baseText('agentNode.card.noModel') }}
</N8nText>
</div>
<CanvasNodeAgentChips v-if="chips.length" :chips="chips" />
<CanvasNodeAgentChips
v-if="chips.length"
:chips="chips"
:active-capability-keys="activeCapabilityKeys"
/>
</template>
</template>
<div v-else :class="[$style.picker, 'nodrag', 'nowheel']">

49
packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/CanvasNodeAgentChips.test.ts
View File

@ -19,7 +19,7 @@ const renderComponent = createComponentRenderer(CanvasNodeAgentChips, {
});
function chip(label: string, overrides: Partial<AgentCardChip> = {}): AgentCardChip {
return { key: `k:${label}`, icon: 'zap', label, ...overrides };
return { key: `k:${label}`, icon: 'zap', label, activityKeys: [], ...overrides };
}
beforeEach(() => {
@ -68,4 +68,51 @@ describe('CanvasNodeAgentChips', () => {
expect(queryByTestId('chip-node-icon')).toBeNull();
});
it('animates a matching inline capability with the canvas running state', () => {
const { getByTestId } = renderComponent({
props: {
chips: [chip('Lookup', { activityKeys: ['tool:lookup'] })],
activeCapabilityKeys: new Set(['tool:lookup']),
},
});
const activeChip = getByTestId('canvas-node-agent-chip');
expect(activeChip).toHaveAttribute('aria-busy', 'true');
expect(activeChip.className).not.toContain('running');
expect(activeChip.parentElement?.className).toContain('running');
});
it('does not animate an inactive inline capability', () => {
const { getByTestId } = renderComponent({
props: {
chips: [chip('Lookup', { activityKeys: ['tool:lookup'] })],
activeCapabilityKeys: new Set(),
},
});
const inactiveChip = getByTestId('canvas-node-agent-chip');
expect(inactiveChip).toHaveAttribute('aria-busy', 'false');
expect(inactiveChip.className).not.toContain('running');
expect(inactiveChip.parentElement?.className).not.toContain('running');
});
it('marks the overflow pill active when a hidden capability is active', () => {
const { getByTestId } = renderComponent({
props: {
chips: [
chip('Visible', { activityKeys: ['tool:visible'] }),
chip('Hidden', { activityKeys: ['tool:hidden'] }),
],
maxInline: 1,
isReadOnly: true,
activeCapabilityKeys: new Set(['tool:hidden']),
},
});
const overflowChip = getByTestId('canvas-node-agent-chips-overflow');
expect(overflowChip).toHaveAttribute('aria-busy', 'true');
expect(overflowChip.className).not.toContain('running');
expect(overflowChip.parentElement?.className).toContain('running');
});
});

94
packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/CanvasNodeAgentChips.vue
View File

@ -7,14 +7,20 @@ import NodeIcon from '@/app/components/NodeIcon.vue';
import { useNodeTypesStore } from '@/app/stores/nodeTypes.store';
import AgentChipButton from '@/features/agents/components/AgentChipButton.vue';
import { type AgentCardChip, MAX_INLINE_AGENT_CHIPS } from './canvasNodeAgentChips.utils';
import type { AgentCapabilityActivityKey } from '@/features/agents/utils/agentCapabilityActivity';
const props = withDefaults(
defineProps<{
chips: AgentCardChip[];
maxInline?: number;
isReadOnly?: boolean;
activeCapabilityKeys?: ReadonlySet<AgentCapabilityActivityKey>;
}>(),
{ maxInline: MAX_INLINE_AGENT_CHIPS, isReadOnly: false },
{
maxInline: MAX_INLINE_AGENT_CHIPS,
isReadOnly: false,
activeCapabilityKeys: () => new Set<AgentCapabilityActivityKey>(),
},
);
const i18n = useI18n();
@ -31,6 +37,9 @@ const inlineChips = computed(() =>
})),
);
const overflowChips = computed(() => props.chips.slice(props.maxInline));
const isChipActive = (chip: AgentCardChip) =>
(chip.activityKeys ?? []).some((key) => props.activeCapabilityKeys.has(key));
const isOverflowActive = computed(() => overflowChips.value.some(isChipActive));
const overflowItems = computed<Array<ActionDropdownItem<string>>>(() =>
overflowChips.value.map((chip) => ({ id: chip.key, label: chip.label, icon: chip.icon })),
);
@ -38,29 +47,39 @@ const overflowItems = computed<Array<ActionDropdownItem<string>>>(() =>
<template>
<div :class="$style.chips" data-test-id="canvas-node-agent-chips">
<AgentChipButton
<span
v-for="chip in inlineChips"
:key="chip.key"
:icon="chip.nodeTypeDescription ? undefined : chip.icon"
:clickable="false"
data-test-id="canvas-node-agent-chip"
:class="[$style.chipWrapper, { [$style.running]: isChipActive(chip) }]"
>
<template v-if="chip.nodeTypeDescription" #icon>
<NodeIcon :node-type="chip.nodeTypeDescription" :size="16" :class="$style.nodeIcon" />
</template>
{{ chip.label }}
</AgentChipButton>
<AgentChipButton
<AgentChipButton
:icon="chip.nodeTypeDescription ? undefined : chip.icon"
:clickable="false"
:aria-busy="isChipActive(chip)"
data-test-id="canvas-node-agent-chip"
>
<template v-if="chip.nodeTypeDescription" #icon>
<NodeIcon :node-type="chip.nodeTypeDescription" :size="16" :class="$style.nodeIcon" />
</template>
{{ chip.label }}
</AgentChipButton>
</span>
<span
v-if="overflowChips.length && isReadOnly"
:clickable="false"
data-test-id="canvas-node-agent-chips-overflow"
:class="[$style.chipWrapper, { [$style.running]: isOverflowActive }]"
>
{{
i18n.baseText('agentNode.card.moreChips', {
interpolate: { count: overflowChips.length },
})
}}
</AgentChipButton>
<AgentChipButton
:clickable="false"
:aria-busy="isOverflowActive"
data-test-id="canvas-node-agent-chips-overflow"
>
{{
i18n.baseText('agentNode.card.moreChips', {
interpolate: { count: overflowChips.length },
})
}}
</AgentChipButton>
</span>
<N8nActionDropdown
v-else-if="overflowChips.length"
:items="overflowItems"
@ -69,19 +88,23 @@ const overflowItems = computed<Array<ActionDropdownItem<string>>>(() =>
data-test-id="canvas-node-agent-chips-overflow"
>
<template #activator>
<AgentChipButton>
{{
i18n.baseText('agentNode.card.moreChips', {
interpolate: { count: overflowChips.length },
})
}}
</AgentChipButton>
<span :class="[$style.chipWrapper, { [$style.running]: isOverflowActive }]">
<AgentChipButton :aria-busy="isOverflowActive">
{{
i18n.baseText('agentNode.card.moreChips', {
interpolate: { count: overflowChips.length },
})
}}
</AgentChipButton>
</span>
</template>
</N8nActionDropdown>
</div>
</template>
<style lang="scss" module>
@use '../_canvasNodeStyles.scss' as styles;
.chips {
display: flex;
flex-wrap: wrap;
@ -91,4 +114,23 @@ const overflowItems = computed<Array<ActionDropdownItem<string>>>(() =>
.nodeIcon {
flex-shrink: 0;
}
.chipWrapper {
display: inline-flex;
position: relative;
isolation: isolate;
border-radius: var(--radius--full);
}
/* stylelint-disable */
.running::after {
@include styles.status-animated-after;
@include styles.status-running-animation;
border-radius: inherit;
pointer-events: none;
}
@include styles.status-animation-definitions;
/* stylelint-enable */
</style>

41
packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/canvasNodeAgentChips.utils.test.ts
View File

@ -33,12 +33,26 @@ describe('buildAgentCardChips', () => {
expect(chips.some((c) => c.key.startsWith('task:'))).toBe(false);
});
it('humanizes raw tool names like the edit page', () => {
it('humanizes node tool names while tracking their runtime-safe names', () => {
const summary = makeSummary({
tools: [{ type: 'node', name: 'send_telegram_message' }],
tools: [{ type: 'node', name: 'Send Telegram Message' }],
});
expect(buildAgentCardChips(summary)[0].label).toBe('Send telegram message');
expect(buildAgentCardChips(summary)[0]).toMatchObject({
label: 'Send telegram message',
activityKeys: ['tool:Send_Telegram_Message'],
});
});
it('tracks workflow tools by their runtime-safe name while preserving the display label', () => {
const summary = makeSummary({
tools: [{ type: 'workflow', name: 'Lookup Orders & Returns' }],
});
expect(buildAgentCardChips(summary)[0]).toMatchObject({
label: 'Lookup orders & returns',
activityKeys: ['tool:lookup-orders-returns'],
});
});
it('orders MCP servers between tools and skills, with the MCP icon', () => {
@ -116,7 +130,7 @@ describe('buildAgentCardChips', () => {
tools: [
{
type: 'node',
name: 'send_message',
name: 'Send Message',
nodeType: 'n8n-nodes-base.telegramTool',
nodeTypeVersion: 2,
},
@ -136,13 +150,13 @@ describe('buildAgentCardChips', () => {
tools: [
{
type: 'node',
name: 'send_message',
name: 'Send Message',
nodeType: 'n8n-nodes-base.telegramTool',
nodeTypeVersion: 1,
},
{
type: 'node',
name: 'get_chat',
name: 'Get Chat',
nodeType: 'n8n-nodes-base.telegramTool',
nodeTypeVersion: 1,
},
@ -155,6 +169,21 @@ describe('buildAgentCardChips', () => {
label: '2 Telegram',
nodeType: 'n8n-nodes-base.telegramTool',
nodeTypeVersion: 1,
activityKeys: ['tool:Send_Message', 'tool:Get_Chat'],
});
});
it('uses skill id and name activity keys while leaving MCP inactive', () => {
const summary = makeSummary({
mcpServers: [{ name: 'notion-mcp' }],
skills: [{ id: 'skill-1', name: 'Research' }],
});
const chips = buildAgentCardChips(summary);
expect(chips.find((chip) => chip.key === 'mcp:notion-mcp')?.activityKeys).toEqual([]);
expect(chips.find((chip) => chip.key === 'skill:skill-1')?.activityKeys).toEqual([
'skill:id:skill-1',
'skill:name:Research',
]);
});
});

32
packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/canvasNodeAgentChips.utils.ts
View File

@ -1,7 +1,18 @@
import type { AgentCapabilitySummary, AgentCapabilityTool } from '@n8n/api-types';
import {
sanitizeAgentToolName,
type AgentCapabilitySummary,
type AgentCapabilityTool,
} from '@n8n/api-types';
import type { IconName } from '@n8n/design-system';
import { nodeNameToToolName } from 'n8n-workflow';
import { formatToolNameForDisplay } from '@/features/agents/utils/toolDisplayName';
import { MIN_GROUPED_TOOLS_PER_TYPE } from '@/features/agents/components/AgentCapabilitiesSection.utils';
import {
skillIdActivityKey,
skillNameActivityKey,
toolActivityKey,
type AgentCapabilityActivityKey,
} from '@/features/agents/utils/agentCapabilityActivity';
export interface AgentCardChip {
/** Stable key for v-for and overflow dropdown item ids. */
@ -9,6 +20,8 @@ export interface AgentCardChip {
/** Fallback design-system icon, used when `nodeType` is absent or unresolved. */
icon: IconName;
label: string;
/** Runtime identities that can activate this chip during an agent invocation. */
activityKeys: AgentCapabilityActivityKey[];
/**
* Node type + version for node-tool chips, so the chip can render the node's
* own icon (via `NodeIcon`) instead of the generic fallback. Absent for
@ -34,11 +47,19 @@ const TOOL_ICONS = {
} as const satisfies Record<AgentCapabilityTool['type'], IconName>;
function individualToolChip(tool: AgentCapabilityTool, index: number): AgentCardChip {
const activityName =
tool.type === 'workflow'
? sanitizeAgentToolName(tool.name)
: tool.type === 'node'
? nodeNameToToolName(tool.name)
: tool.name;
return {
key: `tool:${tool.type}:${tool.name}:${index}`,
icon: TOOL_ICONS[tool.type],
// Mirror the agent edit page, which humanizes every tool name.
label: formatToolNameForDisplay(tool.name),
activityKeys: [toolActivityKey(activityName)],
nodeType: tool.nodeType,
nodeTypeVersion: tool.nodeTypeVersion,
};
@ -86,6 +107,7 @@ function buildToolChips(
key: `tool:node:${nodeType}`,
icon: TOOL_ICONS.node,
label: `${group.length} ${nodeTypeLabels.get(nodeType)}`,
activityKeys: group.map(({ tool }) => toolActivityKey(nodeNameToToolName(tool.name))),
// All members share this node type, so the group chip shows its icon.
nodeType,
nodeTypeVersion: group[0].tool.nodeTypeVersion,
@ -121,11 +143,17 @@ export function buildAgentCardChips(
key: `mcp:${server.name}`,
icon: 'mcp',
label: formatToolNameForDisplay(server.name),
activityKeys: [],
});
}
for (const skill of summary.skills) {
chips.push({ key: `skill:${skill.id}`, icon: 'sparkles', label: skill.name });
chips.push({
key: `skill:${skill.id}`,
icon: 'sparkles',
label: skill.name,
activityKeys: [skillIdActivityKey(skill.id), skillNameActivityKey(skill.name)],
});
}
return chips;

9
packages/nodes-base/nodes/MessageAnAgent/shared.ts
View File

@ -189,6 +189,13 @@ export const commonProperties: INodeProperties[] = [
placeholder: 'Add Option',
default: {},
options: [
{
displayName: 'Enable Streaming',
name: 'enableStreaming',
type: 'boolean',
default: true,
description: "Whether to stream the agent's response as it generates text",
},
{
displayName: 'Invoke Agent',
name: 'invokeMode',
@ -585,6 +592,7 @@ export async function execute(this: IExecuteFunctions): Promise<INodeExecutionDa
const advanced = this.getNodeParameter('advanced', i, {}) as {
sessionId?: string;
allowOtherNodesData?: boolean;
enableStreaming?: boolean;
};
const allowOtherNodesData =
agentSource === 'inline' ? false : (advanced.allowOtherNodesData ?? false);
@ -611,6 +619,7 @@ export async function execute(this: IExecuteFunctions): Promise<INodeExecutionDa
...source,
sessionId,
outputSchema,
...(advanced.enableStreaming === false ? { enableStreaming: false } : {}),
inputDataScope: runOnceForAll ? 'all' : 'item',
exposeWorkflowData: allowOtherNodesData,
},

14
packages/workflow/src/interfaces.ts
View File

@ -2288,6 +2288,11 @@ export type ExecuteAgentInfo = ExecuteAgentSource & {
* surfaced on {@link ExecuteAgentData.structuredOutput}.
*/
outputSchema?: JSONSchema7;
/**
* Whether to forward generated text to a streaming workflow response. Defaults
* to true when omitted. This does not affect execution progress events.
*/
enableStreaming?: boolean;
/**
* Which slice of the calling node's input the agent's `fetch_input_data`
* tool should expose: the single current item (`'item'`, default) or all
@ -2334,6 +2339,14 @@ export interface ExecuteAgentWorkflowContext {
runExecutionData: IRunExecutionData;
}
export interface ExecuteAgentInvocationContext {
nodeId: string;
nodeName: string;
runIndex: number;
itemIndex: number;
sendResponseChunk?: (type: ChunkType, content?: string) => Promise<void>;
}
export interface ExecuteAgentOptions {
node?: INode;
parentWorkflowId: string;
@ -3675,6 +3688,7 @@ export interface IWorkflowExecuteAdditionalData {
executionMode: WorkflowExecuteMode,
outputSchema?: JSONSchema7,
workflowContext?: ExecuteAgentWorkflowContext,
invocationContext?: ExecuteAgentInvocationContext,
) => Promise<ExecuteAgentData>;
listAgents?: (userId: string) => Promise<Array<{ id: string; name: string }>>;
getRunExecutionData: (executionId: string) => Promise<IRunExecutionData | undefined>;