From 93ca2d1fc09fa3d2032c33ce112dbec8493fb546 Mon Sep 17 00:00:00 2001 From: Anne Aguirre Date: Thu, 10 Sep 2026 15:43:28 +0000 Subject: [PATCH] fix(ai-builder): Guard against 429 errors on channel integrations (#38106) Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com> --- .../agent-chat-stream-consumer.test.ts | 114 +++++++ .../channel-rate-limit.guard.test.ts | 57 ++++ .../__tests__/channel-rate-limit.test.ts | 209 ++++++++++++ .../__tests__/helpers/replay-test-helpers.ts | 7 +- .../integration-action-executor.test.ts | 303 ++++++++++++++++-- ...integration-context-query-executor.test.ts | 103 ++++++ .../__tests__/n8n-chat-integration.test.ts | 19 +- .../agents/integrations/agent-chat-bridge.ts | 13 +- .../agent-chat-stream-consumer.ts | 37 ++- .../integrations/channel-rate-limit.guard.ts | 26 ++ .../agents/integrations/channel-rate-limit.ts | 89 +++++ .../integration-action-executor.ts | 28 +- .../integration-context-query-executor.ts | 29 +- .../integrations/integration-error-codes.ts | 1 + .../integrations/integration-helpers.ts | 4 + 15 files changed, 975 insertions(+), 64 deletions(-) create mode 100644 packages/cli/src/modules/agents/integrations/__tests__/agent-chat-stream-consumer.test.ts create mode 100644 packages/cli/src/modules/agents/integrations/__tests__/channel-rate-limit.guard.test.ts create mode 100644 packages/cli/src/modules/agents/integrations/__tests__/channel-rate-limit.test.ts create mode 100644 packages/cli/src/modules/agents/integrations/channel-rate-limit.guard.ts create mode 100644 packages/cli/src/modules/agents/integrations/channel-rate-limit.ts diff --git a/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-stream-consumer.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-stream-consumer.test.ts new file mode 100644 index 00000000000..a21c38925f0 --- /dev/null +++ b/packages/cli/src/modules/agents/integrations/__tests__/agent-chat-stream-consumer.test.ts @@ -0,0 +1,114 @@ +import type { StreamChunk } from '@n8n/agents'; +import type { Thread } from 'chat'; +import type { Logger } from 'n8n-workflow'; +import { mock } from 'vitest-mock-extended'; + +import { INTEGRATION_ERROR_CODES } from '../integration-error-codes'; +import { AgentChatStreamConsumer } from '../agent-chat-stream-consumer'; + +function makeStream(chunks: StreamChunk[]): AsyncGenerator { + return (async function* () { + for (const chunk of chunks) yield chunk; + })(); +} + +function makeConsumer( + postErrorToThread: (thread: Thread | null, error: unknown) => Promise, +) { + return new AgentChatStreamConsumer({ + disableStreaming: true, + logger: mock(), + postErrorToThread, + handleSuspension: vi.fn().mockResolvedValue('skipped'), + handleMessage: vi.fn().mockResolvedValue(false), + isIntegrationActionTool: () => true, + }); +} + +const thread = mock>(); + +describe('AgentChatStreamConsumer — rate-limit fallback', () => { + it('posts a fallback error after a RATE_LIMIT_EXCEEDED tool result', async () => { + const postErrorToThread = vi.fn().mockResolvedValue(undefined); + const consumer = makeConsumer(postErrorToThread); + + const rateLimitResult = { + ok: false, + error: { + code: INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED, + message: + 'The Slack integration has exceeded its rate limit. Please wait a few minutes before trying again.', + }, + }; + + await consumer.consume( + makeStream([ + { + type: 'tool-result', + toolCallId: 'tc-1', + toolName: 'slack_action', + output: rateLimitResult, + }, + { type: 'text-delta', id: 't-1', delta: 'Let me try again' }, + ]), + thread, + ); + + expect(postErrorToThread).toHaveBeenCalledTimes(1); + const postedError = postErrorToThread.mock.calls[0][1]; + expect(postedError).toEqual(rateLimitResult); + }); + + it('still posts the fallback when a later tool result succeeds', async () => { + const postErrorToThread = vi.fn().mockResolvedValue(undefined); + const consumer = makeConsumer(postErrorToThread); + + const rateLimitResult = { + ok: false, + error: { + code: INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED, + message: 'The Slack integration has exceeded its rate limit.', + }, + }; + + await consumer.consume( + makeStream([ + { + type: 'tool-result', + toolCallId: 'tc-1', + toolName: 'slack_action', + output: rateLimitResult, + }, + { + type: 'tool-result', + toolCallId: 'tc-2', + toolName: 'linear_action', + output: { ok: true }, + }, + ]), + thread, + ); + + // rate-limit is not cleared by a later success + expect(postErrorToThread).toHaveBeenCalledTimes(1); + }); + + it('does not post a fallback when there is no rate-limit or error', async () => { + const postErrorToThread = vi.fn().mockResolvedValue(undefined); + const consumer = makeConsumer(postErrorToThread); + + await consumer.consume( + makeStream([ + { + type: 'tool-result', + toolCallId: 'tc-1', + toolName: 'slack_action', + output: { ok: true }, + }, + ]), + thread, + ); + + expect(postErrorToThread).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/channel-rate-limit.guard.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/channel-rate-limit.guard.test.ts new file mode 100644 index 00000000000..8188d388e3f --- /dev/null +++ b/packages/cli/src/modules/agents/integrations/__tests__/channel-rate-limit.guard.test.ts @@ -0,0 +1,57 @@ +import { describe, it, expect } from 'vitest'; + +import { CHANNEL_RATE_LIMIT_COOLDOWN_MS } from '../channel-rate-limit'; +import { ChannelRateLimitGuard } from '../channel-rate-limit.guard'; + +describe('ChannelRateLimitGuard', () => { + it('is not blocked before record is called', () => { + const guard = new ChannelRateLimitGuard(); + expect(guard.isBlocked('slack:cred-a', 1_000)).toBe(false); + }); + + it('is blocked after record and unblocked after the cooldown expires', () => { + const guard = new ChannelRateLimitGuard(); + guard.record('slack:cred-a', 1_000); + expect(guard.isBlocked('slack:cred-a', 1_000)).toBe(true); + expect(guard.isBlocked('slack:cred-a', 1_000 + CHANNEL_RATE_LIMIT_COOLDOWN_MS)).toBe(false); + }); + + it('is unblocked exactly at the expiry boundary', () => { + const guard = new ChannelRateLimitGuard(); + guard.record('slack:cred-a', 1_000); + // nowMs === until → still blocked (nowMs > until is the clear condition) + expect(guard.isBlocked('slack:cred-a', 1_000 + CHANNEL_RATE_LIMIT_COOLDOWN_MS - 1)).toBe(true); + expect(guard.isBlocked('slack:cred-a', 1_000 + CHANNEL_RATE_LIMIT_COOLDOWN_MS)).toBe(false); + }); + + it('extends an existing block when a second record would end later', () => { + const guard = new ChannelRateLimitGuard(); + guard.record('slack:cred-a', 1_000); + guard.record('slack:cred-a', 10_000); + // First until = 1_000 + COOLDOWN; second until = 10_000 + COOLDOWN (later) + expect(guard.isBlocked('slack:cred-a', 5_000 + 10_000)).toBe(true); + expect(guard.isBlocked('slack:cred-a', 10_000 + CHANNEL_RATE_LIMIT_COOLDOWN_MS)).toBe(false); + }); + + it('keeps the later expiry when a second record would end earlier', () => { + const guard = new ChannelRateLimitGuard(); + guard.record('slack:cred-a', 10_000); + guard.record('slack:cred-a', 1_000); + // First until = 10_000 + COOLDOWN; second until = 1_000 + COOLDOWN (earlier) + expect(guard.isBlocked('slack:cred-a', 5_000 + 10_000)).toBe(true); + expect(guard.isBlocked('slack:cred-a', 10_000 + CHANNEL_RATE_LIMIT_COOLDOWN_MS)).toBe(false); + }); + + it('is a no-op for an empty connectionId', () => { + const guard = new ChannelRateLimitGuard(); + guard.record('', 1_000); + expect(guard.isBlocked('', 1_000)).toBe(false); + }); + + it('tracks connections independently', () => { + const guard = new ChannelRateLimitGuard(); + guard.record('slack:cred-a', 1_000); + expect(guard.isBlocked('slack:cred-b', 1_000)).toBe(false); + expect(guard.isBlocked('linear:cred-c', 1_000)).toBe(false); + }); +}); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/channel-rate-limit.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/channel-rate-limit.test.ts new file mode 100644 index 00000000000..346a4cf7c94 --- /dev/null +++ b/packages/cli/src/modules/agents/integrations/__tests__/channel-rate-limit.test.ts @@ -0,0 +1,209 @@ +import { describe, it, expect } from 'vitest'; + +import { INTEGRATION_ERROR_CODES } from '../integration-error-codes'; + +import { + CHANNEL_RATE_LIMIT_COOLDOWN_MS, + isHttp429, + isRateLimitedToolOutput, + channelRateLimitMessage, + caughtIntegrationError, + rateLimitMessageFromError, +} from '../channel-rate-limit'; +import { ChannelRateLimitGuard } from '../channel-rate-limit.guard'; + +describe('channel-rate-limit', () => { + describe('isHttp429', () => { + it('returns true for an error with response.status 429', () => { + const error = Object.assign(new Error('limited'), { + response: { status: 429 }, + }); + expect(isHttp429(error)).toBe(true); + }); + + it('returns false for a 500 error', () => { + const error = Object.assign(new Error('server error'), { + response: { status: 500 }, + }); + expect(isHttp429(error)).toBe(false); + }); + + it('returns false for a message-only 429 (no response.status field)', () => { + expect(isHttp429(new Error('Discord API failed: 429'))).toBe(false); + }); + + it('returns false for a plain error with no response', () => { + expect(isHttp429(new Error('boom'))).toBe(false); + }); + }); + + describe('isRateLimitedToolOutput', () => { + it('returns true for a single RATE_LIMIT_EXCEEDED result', () => { + expect( + isRateLimitedToolOutput({ + ok: false, + error: { code: INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED, message: 'limited' }, + }), + ).toBe(true); + }); + + it('returns true for a batch results array containing a RATE_LIMIT_EXCEEDED entry', () => { + expect( + isRateLimitedToolOutput({ + ok: true, + results: [ + { action: 'respond', result: { ok: true } }, + { + action: 'add_reaction', + result: { + ok: false, + error: { code: INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED, message: 'limited' }, + }, + }, + ], + }), + ).toBe(true); + }); + + it('returns false for an ACTION_FAILED result', () => { + expect( + isRateLimitedToolOutput({ + ok: false, + error: { code: INTEGRATION_ERROR_CODES.ACTION_FAILED, message: 'x' }, + }), + ).toBe(false); + }); + + it('returns false for a successful result', () => { + expect(isRateLimitedToolOutput({ ok: true })).toBe(false); + }); + + it('returns false for a non-record value', () => { + expect(isRateLimitedToolOutput(null)).toBe(false); + expect(isRateLimitedToolOutput('string')).toBe(false); + }); + }); + + describe('channelRateLimitMessage', () => { + it('capitalises the platform name', () => { + expect(channelRateLimitMessage('slack')).toContain('Slack'); + expect(channelRateLimitMessage('discord')).toContain('Discord'); + }); + }); + + describe('rateLimitMessageFromError', () => { + it('returns the error.message from a rate-limited tool output', () => { + const output = { + ok: false, + error: { code: INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED, message: 'Slack is limited' }, + }; + expect(rateLimitMessageFromError(output)).toBe('Slack is limited'); + }); + + it('extracts the message from a batch results array containing a RATE_LIMIT_EXCEEDED entry', () => { + const output = { + ok: true, + results: [ + { action: 'respond', result: { ok: true } }, + { + action: 'add_reaction', + result: { + ok: false, + error: { + code: INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED, + message: 'Slack is limited', + }, + }, + }, + ], + }; + expect(rateLimitMessageFromError(output)).toBe('Slack is limited'); + }); + + it('returns undefined for a batch with no rate-limited entry', () => { + const output = { + ok: true, + results: [ + { action: 'respond', result: { ok: true } }, + { + action: 'add_reaction', + result: { + ok: false, + error: { code: INTEGRATION_ERROR_CODES.ACTION_FAILED, message: 'x' }, + }, + }, + ], + }; + expect(rateLimitMessageFromError(output)).toBeUndefined(); + }); + + it('returns undefined for an unrelated error', () => { + expect(rateLimitMessageFromError(new Error('something else'))).toBeUndefined(); + }); + + it('returns undefined for an Error whose message merely contains the rate-limit phrase', () => { + // Classification is status/code-based only — a message-only match must + // not be treated as a rate limit. + const error = new Error('Slack is temporarily limiting requests. Try later.'); + expect(rateLimitMessageFromError(error)).toBeUndefined(); + }); + }); + + describe('caughtIntegrationError', () => { + it('records the guard and returns RATE_LIMIT_EXCEEDED on a 429', () => { + const guard = new ChannelRateLimitGuard(); + const error = Object.assign(new Error('rate limited'), { + response: { status: 429 }, + }); + const result = caughtIntegrationError(error, { + connectionId: 'slack:cred-a', + platform: 'slack', + guard, + failedCode: INTEGRATION_ERROR_CODES.ACTION_FAILED, + }); + expect(result.ok).toBe(false); + if (!result.ok) { + expect(result.error.code).toBe(INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED); + expect(result.error.message).toContain('Slack'); + } + expect(guard.isBlocked('slack:cred-a')).toBe(true); + }); + + it('returns the failedCode for a non-429 error', () => { + const guard = new ChannelRateLimitGuard(); + const result = caughtIntegrationError(new Error('boom'), { + connectionId: 'slack:cred-a', + platform: 'slack', + guard, + failedCode: INTEGRATION_ERROR_CODES.ACTION_FAILED, + }); + expect(result).toEqual({ + ok: false, + error: { code: INTEGRATION_ERROR_CODES.ACTION_FAILED, message: 'boom' }, + }); + expect(guard.isBlocked('slack:cred-a')).toBe(false); + }); + + it('works with a guard of undefined', () => { + const error = Object.assign(new Error('rate limited'), { + response: { status: 429 }, + }); + const result = caughtIntegrationError(error, { + connectionId: 'slack:cred-a', + platform: 'slack', + guard: undefined, + failedCode: INTEGRATION_ERROR_CODES.ACTION_FAILED, + }); + expect(result.ok).toBe(false); + if (!result.ok) { + expect(result.error.code).toBe(INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED); + } + }); + }); + + describe('CHANNEL_RATE_LIMIT_COOLDOWN_MS', () => { + it('is 15 minutes (900_000 ms)', () => { + expect(CHANNEL_RATE_LIMIT_COOLDOWN_MS).toBe(900_000); + }); + }); +}); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts b/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts index 2f59a0c0fb4..76ecc6169e6 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/helpers/replay-test-helpers.ts @@ -10,6 +10,7 @@ import { ChatIntegrationRegistry, type AgentChatIntegration } from '../../agent- import type { ChatIntegrationService, ChatInstance } from '../../chat-integration.service'; import type { ComponentMapper } from '../../component-mapper'; import { ChatIntegrationActionExecutor } from '../../integration-action-executor'; +import { ChannelRateLimitGuard } from '../../channel-rate-limit.guard'; import type { IntegrationMessageContextService } from '../../integration-message-context.service'; import type { IntegrationMessageContext, @@ -233,7 +234,11 @@ export function createReplayContextSetup(params: { const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(params.chat); - const actionExecutor = new ChatIntegrationActionExecutor(chatIntegrationService, registry); + const actionExecutor = new ChatIntegrationActionExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([params.integration], 'agent-1')[0]; return { diff --git a/packages/cli/src/modules/agents/integrations/__tests__/integration-action-executor.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/integration-action-executor.test.ts index db02ef3e35a..9663b9b0f5e 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/integration-action-executor.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/integration-action-executor.test.ts @@ -55,7 +55,9 @@ import { type AgentChatIntegrationContext, } from '../agent-chat-integration'; import { ChatIntegrationActionExecutor } from '../integration-action-executor'; +import { ChannelRateLimitGuard } from '../channel-rate-limit.guard'; import { getIntegrationToolConnectionDescriptors } from '../integration-tools'; +import type { IntegrationMessageContext } from '../integration-tool-types'; import { LinearIntegration } from '../platforms/linear-integration'; import { SlackIntegration } from '../platforms/slack/slack-integration'; import type { ChatIntegrationService, ChatInstance } from '../chat-integration.service'; @@ -134,7 +136,11 @@ describe('ChatIntegrationActionExecutor', () => { const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -183,7 +189,11 @@ describe('ChatIntegrationActionExecutor', () => { const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; const result = await executor.execute({ @@ -225,7 +235,11 @@ describe('ChatIntegrationActionExecutor', () => { const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(undefined); chatIntegrationService.getChatInstanceForTools.mockResolvedValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -271,7 +285,11 @@ describe('ChatIntegrationActionExecutor', () => { chat.thread.mockReturnValue(sentThread as never); const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -303,7 +321,11 @@ describe('ChatIntegrationActionExecutor', () => { chat.thread.mockReturnValue(sentThread as never); const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -339,7 +361,11 @@ describe('ChatIntegrationActionExecutor', () => { chatIntegrationService.getChatInstance.mockReturnValue(chat); const registry = buildRegistry(); Container.set(ChatIntegrationRegistry, registry); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, registry); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -409,7 +435,11 @@ describe('ChatIntegrationActionExecutor', () => { const registry = buildRegistry(); registry.register(new ShortCallbackTelegramIntegration()); Container.set(ChatIntegrationRegistry, registry); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, registry); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([telegram], 'agent-1')[0]; const result = await executor.execute({ @@ -472,7 +502,11 @@ describe('ChatIntegrationActionExecutor', () => { chatIntegrationService.getChatInstance.mockReturnValue(chat); const registry = buildRegistry(); registry.register(new ShortCallbackTelegramIntegration()); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, registry); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([telegram], 'agent-1', () => ({ actions: ['respond', 'send_dm', 'edit_message'], }))[0]; @@ -513,7 +547,11 @@ describe('ChatIntegrationActionExecutor', () => { chatIntegrationService.getChatInstance.mockReturnValue(chat); const registry = buildRegistry(); registry.register(new ShortCallbackTelegramIntegration()); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, registry); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([telegram], 'agent-1', () => ({ actions: ['respond', 'send_dm', 'edit_message'], }))[0]; @@ -549,7 +587,11 @@ describe('ChatIntegrationActionExecutor', () => { chatIntegrationService.getChatInstance.mockReturnValue(chat); const registry = buildRegistry(); registry.register(new ShortCallbackTelegramIntegration()); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, registry); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([telegram], 'agent-1', () => ({ actions: ['respond', 'send_dm', 'edit_message'], }))[0]; @@ -596,7 +638,11 @@ describe('ChatIntegrationActionExecutor', () => { const registry = buildRegistry(); registry.register(new ShortCallbackTelegramIntegration()); Container.set(ChatIntegrationRegistry, registry); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, registry); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([telegram], 'agent-1', () => ({ actions: ['respond', 'send_dm', 'edit_message'], }))[0]; @@ -660,7 +706,11 @@ describe('ChatIntegrationActionExecutor', () => { chat.getAdapter.mockReturnValue(slackAdapter); const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -711,7 +761,11 @@ describe('ChatIntegrationActionExecutor', () => { chat.getAdapter.mockReturnValue(slackAdapter); const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -757,7 +811,11 @@ describe('ChatIntegrationActionExecutor', () => { chat.getAdapter.mockReturnValue({ addReaction }); const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([discord], 'agent-1')[0]; const inboundThreadId = 'discord:800000000000000001:700000000000000001:600000000000000001'; const inboundTarget = { @@ -823,7 +881,11 @@ describe('ChatIntegrationActionExecutor', () => { chat.getAdapter.mockReturnValue({ addReaction }); const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([discord], 'agent-1')[0]; const explicitThreadId = 'discord:800000000000000001:700000000000000002:600000000000000001'; @@ -882,7 +944,11 @@ describe('ChatIntegrationActionExecutor', () => { chat.getAdapter.mockReturnValue(slackAdapter); const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -905,7 +971,11 @@ describe('ChatIntegrationActionExecutor', () => { it('returns a structured error when the selected connection is unavailable', async () => { const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(undefined); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -957,7 +1027,11 @@ describe('ChatIntegrationActionExecutor', () => { chat.getAdapter.mockReturnValue({ client: linearClient }); const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; const result = await executor.execute({ @@ -1057,7 +1131,11 @@ describe('ChatIntegrationActionExecutor', () => { chat.getAdapter.mockReturnValue({ client: linearClient }); const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; const result = await executor.execute({ @@ -1124,7 +1202,11 @@ describe('ChatIntegrationActionExecutor', () => { chat.getAdapter.mockReturnValue({ client: linearClient }); const chatIntegrationService = mock(); chatIntegrationService.getChatInstance.mockReturnValue(chat); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; const result = await executor.execute({ @@ -1205,7 +1287,11 @@ describe('ChatIntegrationActionExecutor', () => { chatIntegrationService.getChatInstance.mockReturnValue(chat); const registry = buildRegistry(); Container.set(ChatIntegrationRegistry, registry); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, registry); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; return { executor, descriptor, thread }; } @@ -1300,7 +1386,11 @@ describe('ChatIntegrationActionExecutor', () => { it('succeeds without touching the chat instance when the reply is optional', async () => { const chatIntegrationService = mock(); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -1321,7 +1411,11 @@ describe('ChatIntegrationActionExecutor', () => { it('rejects when the reply expectation is required', async () => { const chatIntegrationService = mock(); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; const result = await executor.execute({ @@ -1341,7 +1435,11 @@ describe('ChatIntegrationActionExecutor', () => { it('rejects when the reply expectation is unset or the message context is missing', async () => { const chatIntegrationService = mock(); - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, buildRegistry()); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + new ChannelRateLimitGuard(), + ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; for (const currentMessageContext of [slackMessageContext, undefined]) { @@ -1362,3 +1460,160 @@ describe('ChatIntegrationActionExecutor', () => { }); }); }); + +// --------------------------------------------------------------------------- +// Rate-limit handling +// --------------------------------------------------------------------------- + +describe('ChatIntegrationActionExecutor — rate-limit handling', () => { + it('returns RATE_LIMIT_EXCEEDED when the adapter throws a 429', async () => { + const slackAdapter = { + addReaction: vi + .fn() + .mockRejectedValue(Object.assign(new Error('rate limited'), { response: { status: 429 } })), + }; + const chat = mock(); + chat.getAdapter.mockReturnValue(slackAdapter); + const chatIntegrationService = mock(); + chatIntegrationService.getChatInstance.mockReturnValue(chat); + const guard = new ChannelRateLimitGuard(); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + guard, + ); + const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; + + const result = await executor.execute({ + descriptor, + action: 'add_reaction', + input: { emoji: 'eyes' }, + awaitResponse: false, + currentMessageContext: { + integrationConnectionId: 'slack:cred-a', + platform: 'slack', + target: { + type: 'thread', + threadId: 'slack:C123:123.456', + channelId: 'slack:C123', + }, + messageId: '123.456', + updatedAt: '2026-05-18T10:00:00.000Z', + }, + }); + + expect(result).toEqual({ + ok: false, + error: { + code: 'RATE_LIMIT_EXCEEDED', + message: expect.stringContaining('Slack'), + }, + }); + expect(guard.isBlocked('slack:cred-a')).toBe(true); + }); + + it('does not call the adapter again on a blocked connection', async () => { + const slackAdapter = { + addReaction: vi + .fn() + .mockRejectedValue(Object.assign(new Error('rate limited'), { response: { status: 429 } })), + }; + const chat = mock(); + chat.getAdapter.mockReturnValue(slackAdapter); + const chatIntegrationService = mock(); + chatIntegrationService.getChatInstance.mockReturnValue(chat); + const guard = new ChannelRateLimitGuard(); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + guard, + ); + const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; + + const ctx: IntegrationMessageContext = { + integrationConnectionId: 'slack:cred-a', + platform: 'slack', + target: { + type: 'thread', + threadId: 'slack:C123:123.456', + channelId: 'slack:C123', + }, + messageId: '123.456', + updatedAt: '2026-05-18T10:00:00.000Z', + }; + + // First call: 429 → records the block + await executor.execute({ + descriptor, + action: 'add_reaction', + input: { emoji: 'eyes' }, + awaitResponse: false, + currentMessageContext: ctx, + }); + + expect(slackAdapter.addReaction).toHaveBeenCalledTimes(1); + + // Second call: blocked → adapter never called + const result2 = await executor.execute({ + descriptor, + action: 'add_reaction', + input: { emoji: 'thumbsup' }, + awaitResponse: false, + currentMessageContext: ctx, + }); + + expect(slackAdapter.addReaction).toHaveBeenCalledTimes(1); + expect(chatIntegrationService.getChatInstance).toHaveBeenCalledTimes(1); + expect(result2).toEqual({ + ok: false, + error: { + code: 'RATE_LIMIT_EXCEEDED', + message: expect.stringContaining('Slack'), + }, + }); + }); + + it('still returns ACTION_FAILED for a non-429 error', async () => { + const slackAdapter = { + addReaction: vi.fn().mockRejectedValue(new Error('network failure')), + }; + const chat = mock(); + chat.getAdapter.mockReturnValue(slackAdapter); + const chatIntegrationService = mock(); + chatIntegrationService.getChatInstance.mockReturnValue(chat); + const guard = new ChannelRateLimitGuard(); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + buildRegistry(), + guard, + ); + const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; + + const result = await executor.execute({ + descriptor, + action: 'add_reaction', + input: { emoji: 'eyes' }, + awaitResponse: false, + currentMessageContext: { + integrationConnectionId: 'slack:cred-a', + platform: 'slack', + target: { + type: 'thread', + threadId: 'slack:C123:123.456', + channelId: 'slack:C123', + }, + messageId: '123.456', + updatedAt: '2026-05-18T10:00:00.000Z', + }, + }); + + expect(result).toEqual({ + ok: false, + error: { + code: 'ACTION_FAILED', + message: 'network failure', + }, + }); + expect(guard.isBlocked('slack:cred-a')).toBe(false); + }); +}); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/integration-context-query-executor.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/integration-context-query-executor.test.ts index dae3679e1d2..a84f4f08901 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/integration-context-query-executor.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/integration-context-query-executor.test.ts @@ -6,6 +6,7 @@ import type { AgentRepository } from '../../repositories/agent.repository'; import { ChatIntegrationRegistry } from '../agent-chat-integration'; import type { ChatIntegrationService, ChatInstance } from '../chat-integration.service'; import { ChatIntegrationContextQueryExecutor } from '../integration-context-query-executor'; +import { ChannelRateLimitGuard } from '../channel-rate-limit.guard'; import { getIntegrationToolConnectionDescriptors } from '../integration-tools'; import { LinearIntegration } from '../platforms/linear-integration'; import { SlackIntegration } from '../platforms/slack/slack-integration'; @@ -67,6 +68,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; @@ -136,6 +138,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; @@ -191,6 +194,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; @@ -250,6 +254,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; @@ -316,6 +321,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; @@ -412,6 +418,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; @@ -559,6 +566,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; @@ -634,6 +642,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; @@ -692,6 +701,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; @@ -736,6 +746,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([linear], 'agent-1')[0]; @@ -771,6 +782,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; @@ -808,6 +820,7 @@ describe('ChatIntegrationContextQueryExecutor', () => { const executor = new ChatIntegrationContextQueryExecutor( chatIntegrationService, buildRegistry(), + new ChannelRateLimitGuard(), ); const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; @@ -826,3 +839,93 @@ describe('ChatIntegrationContextQueryExecutor', () => { expect(result).toMatchObject({ ok: true, nextCursor: 'channel-cursor-page-3' }); }); }); + +// --------------------------------------------------------------------------- +// Rate-limit handling +// --------------------------------------------------------------------------- + +describe('ChatIntegrationContextQueryExecutor — rate-limit handling', () => { + it('returns RATE_LIMIT_EXCEEDED when the adapter throws a 429', async () => { + const usersList = vi + .fn() + .mockRejectedValue(Object.assign(new Error('rate limited'), { response: { status: 429 } })); + const slackAdapter = { + client: { users: { list: usersList } }, + withToken: vi.fn(async (options: Record) => options), + }; + const chat = mock(); + chat.getAdapter.mockReturnValue(slackAdapter); + const chatIntegrationService = mock(); + chatIntegrationService.getChatInstance.mockReturnValue(undefined); + chatIntegrationService.getChatInstanceForTools.mockResolvedValue(chat); + const guard = new ChannelRateLimitGuard(); + const executor = new ChatIntegrationContextQueryExecutor( + chatIntegrationService, + buildRegistry(), + guard, + ); + const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; + + const result = await executor.execute({ + descriptor, + query: 'search_users', + input: { query: 'Michael' }, + }); + + expect(result).toEqual({ + ok: false, + error: { + code: 'RATE_LIMIT_EXCEEDED', + message: expect.stringContaining('Slack'), + }, + }); + expect(guard.isBlocked('slack:cred-a')).toBe(true); + }); + + it('does not call getChatInstanceForTools again on a blocked connection', async () => { + const usersList = vi + .fn() + .mockRejectedValue(Object.assign(new Error('rate limited'), { response: { status: 429 } })); + const slackAdapter = { + client: { users: { list: usersList } }, + withToken: vi.fn(async (options: Record) => options), + }; + const chat = mock(); + chat.getAdapter.mockReturnValue(slackAdapter); + const chatIntegrationService = mock(); + chatIntegrationService.getChatInstance.mockReturnValue(undefined); + chatIntegrationService.getChatInstanceForTools.mockResolvedValue(chat); + const guard = new ChannelRateLimitGuard(); + const executor = new ChatIntegrationContextQueryExecutor( + chatIntegrationService, + buildRegistry(), + guard, + ); + const descriptor = getIntegrationToolConnectionDescriptors([slack], 'agent-1')[0]; + + // First call: 429 → records the block + await executor.execute({ + descriptor, + query: 'search_users', + input: { query: 'Michael' }, + }); + + expect(chatIntegrationService.getChatInstanceForTools).toHaveBeenCalledTimes(1); + + // Second call: blocked → adapter never called + const result2 = await executor.execute({ + descriptor, + query: 'search_users', + input: { query: 'John' }, + }); + + expect(chatIntegrationService.getChatInstanceForTools).toHaveBeenCalledTimes(1); + expect(result2).toEqual({ + ok: false, + error: { + code: 'RATE_LIMIT_EXCEEDED', + message: expect.stringContaining('Slack'), + }, + }); + }); +}); diff --git a/packages/cli/src/modules/agents/integrations/__tests__/n8n-chat-integration.test.ts b/packages/cli/src/modules/agents/integrations/__tests__/n8n-chat-integration.test.ts index 89109a425de..0bad264a702 100644 --- a/packages/cli/src/modules/agents/integrations/__tests__/n8n-chat-integration.test.ts +++ b/packages/cli/src/modules/agents/integrations/__tests__/n8n-chat-integration.test.ts @@ -4,6 +4,7 @@ import { mock } from 'vitest-mock-extended'; import { ChatIntegrationRegistry } from '../agent-chat-integration'; import { ChatIntegrationActionExecutor } from '../integration-action-executor'; import { ChatIntegrationContextQueryExecutor } from '../integration-context-query-executor'; +import { ChannelRateLimitGuard } from '../channel-rate-limit.guard'; import type { ChatIntegrationService } from '../chat-integration.service'; import { createIntegrationActionTool } from '../integration-tools'; import type { @@ -210,7 +211,11 @@ describe('internal integration dispatch', () => { }); it('action executor skips getChatInstance and routes respond to the integration', async () => { - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, registry); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const result = await executor.execute({ descriptor, action: 'respond', @@ -223,7 +228,11 @@ describe('internal integration dispatch', () => { }); it('action executor returns UNSUPPORTED_ACTION for send_dm on n8n_chat', async () => { - const executor = new ChatIntegrationActionExecutor(chatIntegrationService, registry); + const executor = new ChatIntegrationActionExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const result = await executor.execute({ descriptor, action: 'send_dm', @@ -241,7 +250,11 @@ describe('internal integration dispatch', () => { lastName: 'Lovelace', email: 'ada@example.com', } as User); - const executor = new ChatIntegrationContextQueryExecutor(chatIntegrationService, registry); + const executor = new ChatIntegrationContextQueryExecutor( + chatIntegrationService, + registry, + new ChannelRateLimitGuard(), + ); const result = await executor.execute({ descriptor, query: 'get_user', diff --git a/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts b/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts index d20318db338..380a1692818 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-bridge.ts @@ -50,6 +50,8 @@ import { downloadDiscordAttachment } from './platforms/discord-operations'; import { type InternalThread, toInternalThreadId } from './types'; +import { rateLimitMessageFromError } from './channel-rate-limit'; + const RESET_SESSION_COMMAND = '/new'; /** Cache key prefix for the per-conversation session-generation pointer, shared across mains. */ @@ -998,7 +1000,8 @@ export class AgentChatBridge { throwOnDeliveryError = false, ): Promise { const message = error instanceof Error ? error.message : 'An unexpected error occurred'; - + // Resolve a rate-limit message if the error is a rate-limit error, otherwise undefined. + const rateLimitMessage = rateLimitMessageFromError(error); this.logger.error('[AgentChatBridge] Error in handler', { agentId: this.agentId, threadId: thread?.id, @@ -1019,9 +1022,11 @@ export class AgentChatBridge { // A `UserError` is written for people and names the misconfiguration, // which lets an agent owner fix it without reading server logs. const text = - error instanceof UserError - ? `⚠️ This agent is misconfigured: ${error.message} An agent owner has to fix this in n8n.` - : '⚠️ Something went wrong while processing your request. Please try again.'; + rateLimitMessage !== undefined + ? `⚠️ ${rateLimitMessage}` + : error instanceof UserError + ? `⚠️ This agent is misconfigured: ${error.message} An agent owner has to fix this in n8n.` + : '⚠️ Something went wrong while processing your request. Please try again.'; await thread.post(text); } catch (postError) { this.logger.error('[AgentChatBridge] Failed to post error message', { diff --git a/packages/cli/src/modules/agents/integrations/agent-chat-stream-consumer.ts b/packages/cli/src/modules/agents/integrations/agent-chat-stream-consumer.ts index 04214f10921..cb83938323b 100644 --- a/packages/cli/src/modules/agents/integrations/agent-chat-stream-consumer.ts +++ b/packages/cli/src/modules/agents/integrations/agent-chat-stream-consumer.ts @@ -6,6 +6,7 @@ import { OperationalError, type Logger } from 'n8n-workflow'; import type { BridgeStatusHandle } from './agent-chat-integration'; import { isIntegrationActionSuspendPayload } from './agent-chat-suspension-cards'; import { type TextEndFn, type TextYieldFn } from './types'; +import { isRateLimitedToolOutput } from './channel-rate-limit'; type SuspendedChunk = Extract; type MessageChunk = Extract; @@ -59,8 +60,10 @@ interface ResponseState { * retry recovered) and suppressed by visible output (the agent's own text * explains the failure). A 'suspension' fallback is neither — the user * still has an approval request they never received. + * A 'rate-limit' fallback is set when the integration action tool fails with a + * HTTP 429. The user must be informed that the integration is rate-limited. */ - fallbackSource: 'tool-error' | 'suspension' | null; + fallbackSource: 'tool-error' | 'suspension' | 'rate-limit' | null; fallbackError: unknown; } @@ -209,12 +212,7 @@ export class AgentChatStreamConsumer { responseState.hasVisibleResponse = true; break; case 'tool-result': - if (chunk.isError) { - responseState.fallbackSource = 'tool-error'; - responseState.fallbackError = chunk.output; - } else if (responseState.fallbackSource === 'tool-error') { - responseState.fallbackSource = null; - } + this.noteToolResult(chunk, responseState); if (this.isSilentOutcome(chunk)) responseState.suppressText = true; break; default: @@ -229,6 +227,22 @@ export class AgentChatStreamConsumer { } } + private noteToolResult(chunk: ToolResultChunk, state: ResponseState): void { + if (isRateLimitedToolOutput(chunk.output)) { + state.fallbackSource = 'rate-limit'; + state.fallbackError = chunk.output; + return; + } + if (chunk.isError) { + state.fallbackSource = 'tool-error'; + state.fallbackError = chunk.output; + return; + } + if (state.fallbackSource === 'tool-error') { + state.fallbackSource = null; + } + } + /** * True when this bridge's integration action tool reported that no reply * will be sent (`do_not_respond`). Checked per tool so an arbitrary tool @@ -294,7 +308,7 @@ export class AgentChatStreamConsumer { // it). A dropped approval card is never explained by prior text — the run // stays suspended, so the user must be told to retry. if (state.fallbackSource === 'tool-error' && state.hasVisibleResponse) return; - + // 'rate-limit' and 'suspension' always post. await lifecycle.startDiscreteResponse(); await this.options.postErrorToThread(thread, state.fallbackError, throwOnDeliveryError); state.hasVisibleResponse = true; @@ -376,12 +390,7 @@ export class AgentChatStreamConsumer { responseState.hasVisibleResponse = true; break; case 'tool-result': - if (chunk.isError) { - responseState.fallbackSource = 'tool-error'; - responseState.fallbackError = chunk.output; - } else if (responseState.fallbackSource === 'tool-error') { - responseState.fallbackSource = null; - } + this.noteToolResult(chunk, responseState); if (this.isSilentOutcome(chunk)) { responseState.suppressText = true; // Nothing has been posted yet in buffered mode, so the diff --git a/packages/cli/src/modules/agents/integrations/channel-rate-limit.guard.ts b/packages/cli/src/modules/agents/integrations/channel-rate-limit.guard.ts new file mode 100644 index 00000000000..fe751c597d6 --- /dev/null +++ b/packages/cli/src/modules/agents/integrations/channel-rate-limit.guard.ts @@ -0,0 +1,26 @@ +import { Service } from '@n8n/di'; + +import { CHANNEL_RATE_LIMIT_COOLDOWN_MS } from './channel-rate-limit'; + +@Service() +export class ChannelRateLimitGuard { + private readonly blockedUntilMs = new Map(); + + isBlocked(connectionId: string, nowMs = Date.now()): boolean { + if (!connectionId) return false; + const until = this.blockedUntilMs.get(connectionId); + if (until === undefined) return false; + if (nowMs >= until) { + this.blockedUntilMs.delete(connectionId); + return false; + } + return true; + } + + record(connectionId: string, nowMs = Date.now()): void { + if (!connectionId) return; + const until = nowMs + CHANNEL_RATE_LIMIT_COOLDOWN_MS; + const existing = this.blockedUntilMs.get(connectionId) ?? 0; + this.blockedUntilMs.set(connectionId, Math.max(until, existing)); + } +} diff --git a/packages/cli/src/modules/agents/integrations/channel-rate-limit.ts b/packages/cli/src/modules/agents/integrations/channel-rate-limit.ts new file mode 100644 index 00000000000..e0d3017d1a9 --- /dev/null +++ b/packages/cli/src/modules/agents/integrations/channel-rate-limit.ts @@ -0,0 +1,89 @@ +import { Time } from '@n8n/constants'; +import { httpStatusFromError } from '@n8n/backend-network'; +import { isRecord } from '@n8n/utils/is-record'; + +import { INTEGRATION_ERROR_CODES } from './integration-error-codes'; +import { + integrationError, + rateLimitExceeded, + type IntegrationErrorResponse, +} from './integration-helpers'; +import type { ChannelRateLimitGuard } from './channel-rate-limit.guard'; + +export const CHANNEL_RATE_LIMIT_COOLDOWN_MS = 15 * Time.minutes.toMilliseconds; + +function isRateLimitExceededResult(value: unknown): boolean { + return ( + isRecord(value) && + value.ok === false && + isRecord(value.error) && + value.error.code === INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED + ); +} + +function rateLimitMessageFromResult(value: unknown): string | undefined { + if ( + isRecord(value) && + value.ok === false && + isRecord(value.error) && + value.error.code === INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED && + typeof value.error.message === 'string' + ) { + return value.error.message; + } + return undefined; +} + +export function rateLimitMessageFromError(error: unknown): string | undefined { + if (!isRateLimitedToolOutput(error)) return undefined; + // Single result: { ok: false, error: { code, message } } + const single = rateLimitMessageFromResult(error); + if (single !== undefined) return single; + // Batched action calls nest per-operation results under `results`; the + // stream consumer stores the whole batch as the fallback, so extract the + // message from the first rate-limited entry before falling back. + if (isRecord(error) && Array.isArray(error.results)) { + for (const entry of error.results) { + if (!isRecord(entry)) continue; + const message = rateLimitMessageFromResult(entry.result); + if (message !== undefined) return message; + } + } + return undefined; +} + +export function isRateLimitedToolOutput(output: unknown): boolean { + if (isRateLimitExceededResult(output)) return true; + if (!isRecord(output) || !Array.isArray(output.results)) return false; + return output.results.some((entry) => isRecord(entry) && isRateLimitExceededResult(entry.result)); +} + +export function isHttp429(error: unknown): boolean { + return httpStatusFromError(error) === 429; +} + +export function channelRateLimitMessage(platform: string): string { + const label = `${platform.charAt(0).toUpperCase()}${platform.slice(1)}`; + return `The ${label} integration has exceeded its rate limit. Please wait a few minutes before trying again.`; +} + +export function caughtIntegrationError( + error: unknown, + params: { + connectionId: string; + platform: string; + guard: ChannelRateLimitGuard | undefined; + failedCode: + | typeof INTEGRATION_ERROR_CODES.ACTION_FAILED + | typeof INTEGRATION_ERROR_CODES.CONTEXT_QUERY_FAILED; + }, +): IntegrationErrorResponse { + if (isHttp429(error)) { + params.guard?.record(params.connectionId); + return rateLimitExceeded(channelRateLimitMessage(params.platform)); + } + return integrationError( + params.failedCode, + error instanceof Error ? error.message : String(error), + ); +} diff --git a/packages/cli/src/modules/agents/integrations/integration-action-executor.ts b/packages/cli/src/modules/agents/integrations/integration-action-executor.ts index 52f62891d88..cfff01329f0 100644 --- a/packages/cli/src/modules/agents/integrations/integration-action-executor.ts +++ b/packages/cli/src/modules/agents/integrations/integration-action-executor.ts @@ -18,6 +18,7 @@ import { integrationError, normalizePlatformId, unsupportedAction, + rateLimitExceeded, } from './integration-helpers'; import type { IntegrationAction, @@ -26,6 +27,8 @@ import type { IntegrationMessageContext, IntegrationToolConnectionDescriptor, } from './integration-tools'; +import { ChannelRateLimitGuard } from './channel-rate-limit.guard'; +import { caughtIntegrationError, channelRateLimitMessage } from './channel-rate-limit'; // The shared wire schema from @n8n/api-types — the same definition the tool // boundary validates against and the editor-ui renderer parses with. @@ -67,6 +70,7 @@ export class ChatIntegrationActionExecutor implements IntegrationActionExecutor constructor( private readonly chatIntegrationService: ChatIntegrationService, private readonly integrationRegistry: ChatIntegrationRegistry, + private readonly channelRateLimitGuard: ChannelRateLimitGuard, ) {} async execute(params: { @@ -83,7 +87,9 @@ export class ChatIntegrationActionExecutor implements IntegrationActionExecutor if (params.action === 'do_not_respond') { return this.doNotRespond(params); } - + if (this.channelRateLimitGuard.isBlocked(params.descriptor.integrationConnectionId)) { + return rateLimitExceeded(channelRateLimitMessage(params.descriptor.integration.type)); + } const unsupportedAction = () => integrationError( INTEGRATION_ERROR_CODES.UNSUPPORTED_ACTION, @@ -105,10 +111,12 @@ export class ChatIntegrationActionExecutor implements IntegrationActionExecutor }); return result ?? unsupportedAction(); } catch (error) { - return integrationError( - INTEGRATION_ERROR_CODES.ACTION_FAILED, - error instanceof Error ? error.message : String(error), - ); + return caughtIntegrationError(error, { + connectionId: params.descriptor.integrationConnectionId, + platform: params.descriptor.integration.type, + guard: this.channelRateLimitGuard, + failedCode: INTEGRATION_ERROR_CODES.ACTION_FAILED, + }); } } @@ -157,10 +165,12 @@ export class ChatIntegrationActionExecutor implements IntegrationActionExecutor return unsupportedAction(); } catch (error) { - return integrationError( - INTEGRATION_ERROR_CODES.ACTION_FAILED, - error instanceof Error ? error.message : String(error), - ); + return caughtIntegrationError(error, { + connectionId: params.descriptor.integrationConnectionId, + platform: params.descriptor.integration.type, + guard: this.channelRateLimitGuard, + failedCode: INTEGRATION_ERROR_CODES.ACTION_FAILED, + }); } } diff --git a/packages/cli/src/modules/agents/integrations/integration-context-query-executor.ts b/packages/cli/src/modules/agents/integrations/integration-context-query-executor.ts index e983b2333d5..56d525f03d3 100644 --- a/packages/cli/src/modules/agents/integrations/integration-context-query-executor.ts +++ b/packages/cli/src/modules/agents/integrations/integration-context-query-executor.ts @@ -3,12 +3,14 @@ import { Service } from '@n8n/di'; import { ChatIntegrationRegistry } from './agent-chat-integration'; import { ChatIntegrationService } from './chat-integration.service'; import { INTEGRATION_ERROR_CODES } from './integration-error-codes'; -import { connectionUnavailable, integrationError } from './integration-helpers'; +import { connectionUnavailable, integrationError, rateLimitExceeded } from './integration-helpers'; import type { IntegrationContextQuery, IntegrationContextQueryExecutor, IntegrationToolConnectionDescriptor, } from './integration-tools'; +import { ChannelRateLimitGuard } from './channel-rate-limit.guard'; +import { caughtIntegrationError, channelRateLimitMessage } from './channel-rate-limit'; /** * Thin dispatcher that resolves the platform integration for a descriptor and @@ -20,6 +22,7 @@ export class ChatIntegrationContextQueryExecutor implements IntegrationContextQu constructor( private readonly chatIntegrationService: ChatIntegrationService, private readonly integrationRegistry: ChatIntegrationRegistry, + private readonly channelRateLimitGuard: ChannelRateLimitGuard, ) {} async execute(params: { @@ -30,6 +33,10 @@ export class ChatIntegrationContextQueryExecutor implements IntegrationContextQu }): Promise { if (!params.descriptor.agentId) return connectionUnavailable(); + if (this.channelRateLimitGuard.isBlocked(params.descriptor.integrationConnectionId)) { + return rateLimitExceeded(channelRateLimitMessage(params.descriptor.integration.type)); + } + const integrationDef = this.integrationRegistry.get(params.descriptor.integration.type); if (integrationDef && !integrationDef.requiresChatInstance) { if (!integrationDef.executeContextQuery) { @@ -46,10 +53,12 @@ export class ChatIntegrationContextQueryExecutor implements IntegrationContextQu input: params.input, }); } catch (error) { - return integrationError( - INTEGRATION_ERROR_CODES.CONTEXT_QUERY_FAILED, - error instanceof Error ? error.message : String(error), - ); + return caughtIntegrationError(error, { + connectionId: params.descriptor.integrationConnectionId, + platform: params.descriptor.integration.type, + guard: this.channelRateLimitGuard, + failedCode: INTEGRATION_ERROR_CODES.CONTEXT_QUERY_FAILED, + }); } } @@ -81,10 +90,12 @@ export class ChatIntegrationContextQueryExecutor implements IntegrationContextQu input: params.input, }); } catch (error) { - return integrationError( - INTEGRATION_ERROR_CODES.CONTEXT_QUERY_FAILED, - error instanceof Error ? error.message : String(error), - ); + return caughtIntegrationError(error, { + connectionId: params.descriptor.integrationConnectionId, + platform: params.descriptor.integration.type, + guard: this.channelRateLimitGuard, + failedCode: INTEGRATION_ERROR_CODES.CONTEXT_QUERY_FAILED, + }); } } } diff --git a/packages/cli/src/modules/agents/integrations/integration-error-codes.ts b/packages/cli/src/modules/agents/integrations/integration-error-codes.ts index dbe3fa91f0c..ca9794ccedd 100644 --- a/packages/cli/src/modules/agents/integrations/integration-error-codes.ts +++ b/packages/cli/src/modules/agents/integrations/integration-error-codes.ts @@ -10,6 +10,7 @@ export const INTEGRATION_ERROR_CODES = { UNSUPPORTED_ACTION: 'UNSUPPORTED_ACTION', CONTEXT_QUERY_FAILED: 'CONTEXT_QUERY_FAILED', ACTION_FAILED: 'ACTION_FAILED', + RATE_LIMIT_EXCEEDED: 'RATE_LIMIT_EXCEEDED', } as const; export type IntegrationErrorCode = diff --git a/packages/cli/src/modules/agents/integrations/integration-helpers.ts b/packages/cli/src/modules/agents/integrations/integration-helpers.ts index 10de5ceaa27..bf1c2d682da 100644 --- a/packages/cli/src/modules/agents/integrations/integration-helpers.ts +++ b/packages/cli/src/modules/agents/integrations/integration-helpers.ts @@ -17,6 +17,10 @@ export function integrationError( return { ok: false, error: { code, message } }; } +export function rateLimitExceeded(message: string): IntegrationErrorResponse { + return integrationError(INTEGRATION_ERROR_CODES.RATE_LIMIT_EXCEEDED, message); +} + export function connectionUnavailable(): IntegrationErrorResponse { return integrationError( INTEGRATION_ERROR_CODES.CONNECTION_NOT_AVAILABLE,