From 4f7fbc14d38d55d88897876ccb2d8bdb7777fad1 Mon Sep 17 00:00:00 2001 From: Michael Drury Date: Fri, 21 Aug 2026 09:11:07 +0000 Subject: [PATCH] feat: Add streaming to Message an Agent workflows (#36734) --- .../sanitize-agent-tool-name.test.ts | 15 + packages/@n8n/api-types/src/agents/index.ts | 1 + .../src/agents/sanitize-agent-tool-name.ts | 14 + packages/@n8n/api-types/src/push/execution.ts | 21 ++ packages/@n8n/api-types/src/push/index.ts | 2 + .../workflow-execute-additional-data.test.ts | 55 ++++ .../agent-workflow-execution.service.test.ts | 72 ++++- .../__tests__/workflow-agent-stream.test.ts | 296 ++++++++++++++++++ .../agent-workflow-execution.service.ts | 43 ++- .../json-config/agent-config-composition.ts | 17 +- .../modules/agents/workflow-agent-stream.ts | 164 ++++++++++ .../src/workflow-execute-additional-data.ts | 19 +- .../__tests__/execute-context.test.ts | 111 ++++++- .../base-execute-context.ts | 51 +++ .../node-execution-context/execute-context.ts | 42 --- .../handlers/agentNodeProgress.ts | 10 + .../usePushConnection/handlers/index.ts | 1 + .../handlers/nodeExecuteAfter.test.ts | 2 + .../handlers/nodeExecuteAfter.ts | 1 + .../usePushConnection.test.ts | 28 +- .../usePushConnection/usePushConnection.ts | 3 + .../editor-ui/src/app/constants/durations.ts | 3 + .../workflowExecutionState.store.test.ts | 130 +++++++- .../stores/workflowExecutionState.store.ts | 106 +++++++ .../agents/utils/agentCapabilityActivity.ts | 23 ++ .../nodes/render-types/CanvasNodeAgent.vue | 13 +- .../parts/CanvasNodeAgentChips.test.ts | 49 ++- .../parts/CanvasNodeAgentChips.vue | 94 ++++-- .../parts/canvasNodeAgentChips.utils.test.ts | 41 ++- .../parts/canvasNodeAgentChips.utils.ts | 32 +- .../nodes-base/nodes/MessageAnAgent/shared.ts | 9 + packages/workflow/src/interfaces.ts | 14 + 32 files changed, 1368 insertions(+), 114 deletions(-) create mode 100644 packages/@n8n/api-types/src/agents/__tests__/sanitize-agent-tool-name.test.ts create mode 100644 packages/@n8n/api-types/src/agents/sanitize-agent-tool-name.ts create mode 100644 packages/cli/src/modules/agents/__tests__/workflow-agent-stream.test.ts create mode 100644 packages/cli/src/modules/agents/workflow-agent-stream.ts create mode 100644 packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/agentNodeProgress.ts create mode 100644 packages/frontend/editor-ui/src/features/agents/utils/agentCapabilityActivity.ts diff --git a/packages/@n8n/api-types/src/agents/__tests__/sanitize-agent-tool-name.test.ts b/packages/@n8n/api-types/src/agents/__tests__/sanitize-agent-tool-name.test.ts new file mode 100644 index 00000000000..93b817eff23 --- /dev/null +++ b/packages/@n8n/api-types/src/agents/__tests__/sanitize-agent-tool-name.test.ts @@ -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'); + }); +}); diff --git a/packages/@n8n/api-types/src/agents/index.ts b/packages/@n8n/api-types/src/agents/index.ts index ccbdad686f2..b5e78e5c8bc 100644 --- a/packages/@n8n/api-types/src/agents/index.ts +++ b/packages/@n8n/api-types/src/agents/index.ts @@ -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'; diff --git a/packages/@n8n/api-types/src/agents/sanitize-agent-tool-name.ts b/packages/@n8n/api-types/src/agents/sanitize-agent-tool-name.ts new file mode 100644 index 00000000000..e051f1378b1 --- /dev/null +++ b/packages/@n8n/api-types/src/agents/sanitize-agent-tool-name.ts @@ -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'; +} diff --git a/packages/@n8n/api-types/src/push/execution.ts b/packages/@n8n/api-types/src/push/execution.ts index a518a91fa64..785e72b7f2c 100644 --- a/packages/@n8n/api-types/src/push/execution.ts +++ b/packages/@n8n/api-types/src/push/execution.ts @@ -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; diff --git a/packages/@n8n/api-types/src/push/index.ts b/packages/@n8n/api-types/src/push/index.ts index 1993a46cc8e..b476e60618c 100644 --- a/packages/@n8n/api-types/src/push/index.ts +++ b/packages/@n8n/api-types/src/push/index.ts @@ -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 diff --git a/packages/cli/src/__tests__/workflow-execute-additional-data.test.ts b/packages/cli/src/__tests__/workflow-execute-additional-data.test.ts index b6c8e04e6ca..9a1781087fb 100644 --- a/packages/cli/src/__tests__/workflow-execute-additional-data.test.ts +++ b/packages/cli/src/__tests__/workflow-execute-additional-data.test.ts @@ -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({ + 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({ userId: 'user-1', diff --git a/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts b/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts index f82e47a07a3..494130f5f97 100644 --- a/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts +++ b/packages/cli/src/modules/agents/__tests__/agent-workflow-execution.service.test.ts @@ -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().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().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().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')); diff --git a/packages/cli/src/modules/agents/__tests__/workflow-agent-stream.test.ts b/packages/cli/src/modules/agents/__tests__/workflow-agent-stream.test.ts new file mode 100644 index 00000000000..2a98e1bb1c3 --- /dev/null +++ b/packages/cli/src/modules/agents/__tests__/workflow-agent-stream.test.ts @@ -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().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(), + 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({ 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(), + 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(); + }); +}); diff --git a/packages/cli/src/modules/agents/agent-workflow-execution.service.ts b/packages/cli/src/modules/agents/agent-workflow-execution.service.ts index 8da6c187416..513267e23dd 100644 --- a/packages/cli/src/modules/agents/agent-workflow-execution.service.ts +++ b/packages/cli/src/modules/agents/agent-workflow-execution.service.ts @@ -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 { 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 { 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 { 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 { 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 diff --git a/packages/cli/src/modules/agents/json-config/agent-config-composition.ts b/packages/cli/src/modules/agents/json-config/agent-config-composition.ts index dd84344321b..f7a683d7550 100644 --- a/packages/cli/src/modules/agents/json-config/agent-config-composition.ts +++ b/packages/cli/src/modules/agents/json-config/agent-config-composition.ts @@ -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); -} diff --git a/packages/cli/src/modules/agents/workflow-agent-stream.ts b/packages/cli/src/modules/agents/workflow-agent-stream.ts new file mode 100644 index 00000000000..755ef1501fc --- /dev/null +++ b/packages/cli/src/modules/agents/workflow-agent-stream.ts @@ -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; + +type PendingTool = { + toolName: string; + input: unknown; +}; + +function isRecord(value: unknown): value is Record { + 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(); + + private readonly runningCapabilities = new Map(); + + constructor(private readonly observer?: WorkflowAgentStreamObserver) {} + + async observe(chunk: StreamChunk): Promise { + 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 { + if (this.openTextId !== undefined) await this.endText(); + this.openTextId = id; + await this.emit({ type: 'response-begin' }); + } + + private async endText(): Promise { + if (this.openTextId === undefined) return; + this.openTextId = undefined; + await this.emit({ type: 'response-end' }); + } + + private async startCapability(toolCallId: string, toolName: string): Promise { + 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 { + 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 { + await this.observer?.(event); + } +} diff --git a/packages/cli/src/workflow-execute-additional-data.ts b/packages/cli/src/workflow-execute-additional-data.ts index 177267e15d1..0d56df4dc61 100644 --- a/packages/cli/src/workflow-execute-additional-data.ts +++ b/packages/cli/src/workflow-execute-additional-data.ts @@ -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 { 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 diff --git a/packages/core/src/execution-engine/node-execution-context/__tests__/execute-context.test.ts b/packages/core/src/execution-engine/node-execution-context/__tests__/execute-context.test.ts index 7e0ad316ea6..cbc673dbdf3 100644 --- a/packages/core/src/execution-engine/node-execution-context/__tests__/execute-context.test.ts +++ b/packages/core/src/execution-engine/node-execution-context/__tests__/execute-context.test.ts @@ -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()); + const sendChunkHandler = vi.fn(); + hooks.addHandler('sendChunk', sendChunkHandler); + const executeAgentMock = vi.fn().mockResolvedValue({ response: 'ok' }); + const additionalData = mock({ + 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 }), ); }); }); diff --git a/packages/core/src/execution-engine/node-execution-context/base-execute-context.ts b/packages/core/src/execution-engine/node-execution-context/base-execute-context.ts index d83d8f84f02..68defdda69c 100644 --- a/packages/core/src/execution-engine/node-execution-context/base-execute-context.ts +++ b/packages/core/src/execution-engine/node-execution-context/base-execute-context.ts @@ -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 { + 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 { return await this.additionalData.getRunExecutionData(executionId); } diff --git a/packages/core/src/execution-engine/node-execution-context/execute-context.ts b/packages/core/src/execution-engine/node-execution-context/execute-context.ts index ce0fe476f0e..0fd62a5734b 100644 --- a/packages/core/src/execution-engine/node-execution-context/execute-context.ts +++ b/packages/core/src/execution-engine/node-execution-context/execute-context.ts @@ -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 { - 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, diff --git a/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/agentNodeProgress.ts b/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/agentNodeProgress.ts new file mode 100644 index 00000000000..07fa0ee6d57 --- /dev/null +++ b/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/agentNodeProgress.ts @@ -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); +} diff --git a/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/index.ts b/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/index.ts index 088c0bbb654..6b1a5d46781 100644 --- a/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/index.ts +++ b/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/index.ts @@ -1,3 +1,4 @@ +export * from './agentNodeProgress'; export * from './builderCreditsUpdated'; export * from './executionFinished'; export * from './executionRecovered'; diff --git a/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/nodeExecuteAfter.test.ts b/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/nodeExecuteAfter.test.ts index 056e4c8941c..d012248c032 100644 --- a/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/nodeExecuteAfter.test.ts +++ b/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/nodeExecuteAfter.test.ts @@ -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); diff --git a/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/nodeExecuteAfter.ts b/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/nodeExecuteAfter.ts index 6b4752597ab..cb29155a964 100644 --- a/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/nodeExecuteAfter.ts +++ b/packages/frontend/editor-ui/src/app/composables/usePushConnection/handlers/nodeExecuteAfter.ts @@ -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) { diff --git a/packages/frontend/editor-ui/src/app/composables/usePushConnection/usePushConnection.test.ts b/packages/frontend/editor-ui/src/app/composables/usePushConnection/usePushConnection.test.ts index 199b069f96f..54c67e561e9 100644 --- a/packages/frontend/editor-ui/src/app/composables/usePushConnection/usePushConnection.test.ts +++ b/packages/frontend/editor-ui/src/app/composables/usePushConnection/usePushConnection.test.ts @@ -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(); diff --git a/packages/frontend/editor-ui/src/app/composables/usePushConnection/usePushConnection.ts b/packages/frontend/editor-ui/src/app/composables/usePushConnection/usePushConnection.ts index cd759ca5099..e2dc1c8c52d 100644 --- a/packages/frontend/editor-ui/src/app/composables/usePushConnection/usePushConnection.ts +++ b/packages/frontend/editor-ui/src/app/composables/usePushConnection/usePushConnection.ts @@ -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 = {}): Execut } as ExecutionSummary; } +function makeAgentProgress(overrides: Partial = {}): 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; + + 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'); diff --git a/packages/frontend/editor-ui/src/app/stores/workflowExecutionState.store.ts b/packages/frontend/editor-ui/src/app/stores/workflowExecutionState.store.ts index 02b983b5ca9..46375ede2d8 100644 --- a/packages/frontend/editor-ui/src/app/stores/workflowExecutionState.store.ts +++ b/packages/frontend/editor-ui/src/app/stores/workflowExecutionState.store.ts @@ -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>(); 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>(); + + const activeAgentCapabilityKeysByNodeId = computed(() => { + const keysByNodeId = new Map>(); + for (const call of activeAgentCapabilityCalls.values()) { + const keys = keysByNodeId.get(call.nodeId) ?? new Set(); + 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(); @@ -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, diff --git a/packages/frontend/editor-ui/src/features/agents/utils/agentCapabilityActivity.ts b/packages/frontend/editor-ui/src/features/agents/utils/agentCapabilityActivity.ts new file mode 100644 index 00000000000..f7e7bd4d1a8 --- /dev/null +++ b/packages/frontend/editor-ui/src/features/agents/utils/agentCapabilityActivity.ts @@ -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)]; +} diff --git a/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/CanvasNodeAgent.vue b/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/CanvasNodeAgent.vue index 96df7360f42..1d87ed941b2 100644 --- a/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/CanvasNodeAgent.vue +++ b/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/CanvasNodeAgent.vue @@ -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(), +); // 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') }} - +
diff --git a/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/CanvasNodeAgentChips.test.ts b/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/CanvasNodeAgentChips.test.ts index 15ecb8d4e34..70ac1e57f4e 100644 --- a/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/CanvasNodeAgentChips.test.ts +++ b/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/CanvasNodeAgentChips.test.ts @@ -19,7 +19,7 @@ const renderComponent = createComponentRenderer(CanvasNodeAgentChips, { }); function chip(label: string, overrides: Partial = {}): 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'); + }); }); diff --git a/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/CanvasNodeAgentChips.vue b/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/CanvasNodeAgentChips.vue index 38e2019a183..303c54933f5 100644 --- a/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/CanvasNodeAgentChips.vue +++ b/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/CanvasNodeAgentChips.vue @@ -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; }>(), - { maxInline: MAX_INLINE_AGENT_CHIPS, isReadOnly: false }, + { + maxInline: MAX_INLINE_AGENT_CHIPS, + isReadOnly: false, + activeCapabilityKeys: () => new Set(), + }, ); 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>>(() => overflowChips.value.map((chip) => ({ id: chip.key, label: chip.label, icon: chip.icon })), ); @@ -38,29 +47,39 @@ const overflowItems = computed>>(() => diff --git a/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/canvasNodeAgentChips.utils.test.ts b/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/canvasNodeAgentChips.utils.test.ts index 6d90d0ce813..9aa3cf32f35 100644 --- a/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/canvasNodeAgentChips.utils.test.ts +++ b/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/canvasNodeAgentChips.utils.test.ts @@ -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', + ]); + }); }); diff --git a/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/canvasNodeAgentChips.utils.ts b/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/canvasNodeAgentChips.utils.ts index 635695b5038..43d01002dcb 100644 --- a/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/canvasNodeAgentChips.utils.ts +++ b/packages/frontend/editor-ui/src/features/workflows/canvas/components/elements/nodes/render-types/parts/canvasNodeAgentChips.utils.ts @@ -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; 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; diff --git a/packages/nodes-base/nodes/MessageAnAgent/shared.ts b/packages/nodes-base/nodes/MessageAnAgent/shared.ts index 9310d260189..6ebb11cb84d 100644 --- a/packages/nodes-base/nodes/MessageAnAgent/shared.ts +++ b/packages/nodes-base/nodes/MessageAnAgent/shared.ts @@ -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 Promise; +} + export interface ExecuteAgentOptions { node?: INode; parentWorkflowId: string; @@ -3675,6 +3688,7 @@ export interface IWorkflowExecuteAdditionalData { executionMode: WorkflowExecuteMode, outputSchema?: JSONSchema7, workflowContext?: ExecuteAgentWorkflowContext, + invocationContext?: ExecuteAgentInvocationContext, ) => Promise; listAgents?: (userId: string) => Promise>; getRunExecutionData: (executionId: string) => Promise;