mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 13:39:44 +02:00
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>
This commit is contained in:
Anne Aguirre
GitHub
parent
bea96e54c3
commit
93ca2d1fc0
|
|
@ -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<StreamChunk> {
|
||||
return (async function* () {
|
||||
for (const chunk of chunks) yield chunk;
|
||||
})();
|
||||
}
|
||||
|
||||
function makeConsumer(
|
||||
postErrorToThread: (thread: Thread<unknown, unknown> | null, error: unknown) => Promise<void>,
|
||||
) {
|
||||
return new AgentChatStreamConsumer({
|
||||
disableStreaming: true,
|
||||
logger: mock<Logger>(),
|
||||
postErrorToThread,
|
||||
handleSuspension: vi.fn().mockResolvedValue('skipped'),
|
||||
handleMessage: vi.fn().mockResolvedValue(false),
|
||||
isIntegrationActionTool: () => true,
|
||||
});
|
||||
}
|
||||
|
||||
const thread = mock<Thread<unknown, unknown>>();
|
||||
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -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<TChat extends ChatInstance>(params: {
|
|||
|
||||
const chatIntegrationService = mock<ChatIntegrationService>();
|
||||
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 {
|
||||
|
|
|
|||
|
|
@ -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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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>();
|
||||
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<ChatIntegrationService>();
|
||||
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<ChatIntegrationService>();
|
||||
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<ChatIntegrationService>();
|
||||
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<ChatInstance>();
|
||||
chat.getAdapter.mockReturnValue(slackAdapter);
|
||||
const chatIntegrationService = mock<ChatIntegrationService>();
|
||||
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<ChatInstance>();
|
||||
chat.getAdapter.mockReturnValue(slackAdapter);
|
||||
const chatIntegrationService = mock<ChatIntegrationService>();
|
||||
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<ChatInstance>();
|
||||
chat.getAdapter.mockReturnValue(slackAdapter);
|
||||
const chatIntegrationService = mock<ChatIntegrationService>();
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>) => options),
|
||||
};
|
||||
const chat = mock<ChatInstance>();
|
||||
chat.getAdapter.mockReturnValue(slackAdapter);
|
||||
const chatIntegrationService = mock<ChatIntegrationService>();
|
||||
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<string, unknown>) => options),
|
||||
};
|
||||
const chat = mock<ChatInstance>();
|
||||
chat.getAdapter.mockReturnValue(slackAdapter);
|
||||
const chatIntegrationService = mock<ChatIntegrationService>();
|
||||
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'),
|
||||
},
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -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<void> {
|
||||
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', {
|
||||
|
|
|
|||
|
|
@ -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<StreamChunk, { type: 'tool-call-suspended' }>;
|
||||
type MessageChunk = Extract<StreamChunk, { type: 'message' }>;
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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<string, number>();
|
||||
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
|
@ -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),
|
||||
);
|
||||
}
|
||||
|
|
@ -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,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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<unknown> {
|
||||
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,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 =
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user