test(core): Add channel integration tests (#33080)

Co-authored-by: Michael Drury <michael.drury@n8n.io>
This commit is contained in:
yehorkardash
2026-06-30 16:59:05 +02:00
committed by GitHub
parent 387bb73665
commit 2cbd4b19f1
30 changed files with 4112 additions and 7 deletions
+1
View File
@@ -36,6 +36,7 @@ build-storybook.log
build.log
tcr-dry.log
.agent-setup/
.agent-recordings/
sbom-source.cdx.json
*.junit.xml
junit.xml
+2
View File
@@ -19,6 +19,8 @@
"format:check": "biome ci .",
"lint": "eslint . --quiet",
"lint:fix": "eslint . --fix",
"recording:list": "node scripts/agent-integration-recordings.mjs --list",
"recording:export": "node scripts/agent-integration-recordings.mjs",
"start": "node ../../scripts/os-normalize.mjs --dir bin n8n",
"test": "N8N_LOG_LEVEL=silent DB_SQLITE_POOL_SIZE=4 DB_TYPE=sqlite vitest run",
"test:unit": "N8N_LOG_LEVEL=silent DB_SQLITE_POOL_SIZE=4 DB_TYPE=sqlite vitest run",
@@ -0,0 +1,119 @@
import { mkdir, readFile, readdir, writeFile } from 'fs/promises';
import { join, resolve } from 'path';
function parseArgs(argv) {
const parsed = {
args: [],
recordingDir: undefined,
outputDir: undefined,
};
for (let i = 0; i < argv.length; i++) {
const arg = argv[i];
if (arg === '--recording-dir' || arg === '-r') {
parsed.recordingDir = argv[++i];
continue;
}
if (arg === '--output-dir' || arg === '-o') {
parsed.outputDir = argv[++i];
continue;
}
parsed.args.push(arg);
}
return parsed;
}
function recordingDir(cliRecordingDir) {
return (
cliRecordingDir ??
process.env.N8N_AGENT_INTEGRATION_RECORDING_DIR ??
resolve(process.cwd(), '.agent-recordings', 'channel-integrations')
);
}
function sanitizeSessionId(value) {
return value.replace(/[^a-zA-Z0-9._-]/g, '-');
}
async function listSessions(cliRecordingDir) {
const dir = recordingDir(cliRecordingDir);
console.log(`Scanning ${dir} for sessions...`);
let files = [];
try {
files = await readdir(dir);
} catch {
return [];
}
const sessions = await Promise.all(
files
.filter((file) => file.endsWith('.jsonl'))
.map(async (file) => {
const contents = await readFile(join(dir, file), 'utf8');
return {
sessionId: file.slice(0, -'.jsonl'.length),
entries: contents.split('\n').filter(Boolean).length,
};
}),
);
return sessions.sort((a, b) => a.sessionId.localeCompare(b.sessionId));
}
async function exportSession(sessionId, cliRecordingDir) {
const file = join(recordingDir(cliRecordingDir), `${sanitizeSessionId(sessionId)}.jsonl`);
const contents = await readFile(file, 'utf8');
return contents
.split('\n')
.filter(Boolean)
.map((line) => JSON.parse(line));
}
async function writeExport(sessionId, records, outputDir) {
await mkdir(outputDir, { recursive: true });
const outputPath = join(outputDir, `${sanitizeSessionId(sessionId)}.json`);
await writeFile(outputPath, `${JSON.stringify(records, null, 2)}\n`, 'utf8');
return outputPath;
}
function printHelp() {
console.log(`
Usage:
node scripts/agent-integration-recordings.mjs --list [--recording-dir <dir>]
node scripts/agent-integration-recordings.mjs <session-id> [--recording-dir <dir>] [--output-dir <dir>]
Options:
--recording-dir, -r Directory containing recording JSONL files
--output-dir, -o Directory to write exported JSON files. If omitted, export prints to stdout
Environment:
N8N_AGENT_INTEGRATION_RECORDING_DIR Directory containing recording JSONL files
`);
}
const { args, recordingDir: cliRecordingDir, outputDir } = parseArgs(process.argv.slice(2));
const arg = args[0];
try {
if (arg === '--list' || arg === '-l') {
const sessions = await listSessions(cliRecordingDir);
if (sessions.length === 0) {
console.log('No channel integration recordings found.');
} else {
for (const session of sessions) {
console.log(`${session.sessionId} (${session.entries} entries)`);
}
}
} else if (!arg || arg === '--help' || arg === '-h') {
printHelp();
} else {
const records = await exportSession(arg, cliRecordingDir);
if (outputDir) {
const outputPath = await writeExport(arg, records, outputDir);
console.log(`Exported ${records.length} entries to ${outputPath}`);
} else {
console.log(JSON.stringify(records, null, 2));
}
}
} catch (error) {
console.error(error instanceof Error ? error.message : String(error));
process.exit(1);
}
@@ -19,6 +19,7 @@ import { AgentPublishService } from './agent-publish.service';
import { AgentRunnableStateService } from './agent-runnable-state.service';
import { ChatIntegrationRegistry } from './integrations/agent-chat-integration';
import { ChatIntegrationService } from './integrations/chat-integration.service';
import { channelIntegrationRecorder } from './integrations/recording/channel-integration-recorder';
import { SlackAppSetupService } from './integrations/slack-app-setup.service';
import { AgentRepository } from './repositories/agent.repository';
@@ -288,6 +289,7 @@ export class AgentIntegrationsController {
headers: sanitizedHeaders,
body: requestBody,
});
await channelIntegrationRecorder.recordWebhook(platform, webRequest.clone());
// In Express, background tasks just need to not be garbage collected.
// We hold references to keep them alive for the lifetime of the process.
@@ -0,0 +1,171 @@
# Channel Integration Tests
This guide explains how to add chat platform integration tests for agent channel integrations.
The current suite covers three layers:
- Shared contract tests for behavior every channel integration must support.
- Synthetic platform tests for hand-built edge cases.
- Recorded replay tests for real webhook payloads captured from a local run.
These tests validate n8n's integration logic: routing inbound messages to the agent, preserving
message context, executing integration actions, resuming suspended tool calls, and avoiding
self-trigger loops.
## Design: real adapters, no fakes
These tests run the **real** `@chat-adapter/*` adapters and the real `chat` SDK. There are no fake
or in-memory adapters. The packages are ESM-only; in production they are loaded via `esm-loader`'s
`new Function()` indirection to survive the CJS transform, but **that indirection cannot run under
Vitest** — so the test helpers import them directly (`await import('@chat-adapter/telegram')`), which
Vitest loads natively. Any test that drives a code path which itself calls `esm-loader` (e.g. the
real `ComponentMapper` for rich cards) must `vi.mock('../esm-loader', …)` to redirect the loaders to
native dynamic imports — see the telegram tests for the pattern.
The only thing faked is the **external platform HTTP at the network boundary** — you cannot call
live Telegram/Slack/Linear from CI. Everything else (adapter, `AgentChatBridge`, integration
implementation, action executor, message-context service) is real.
### Answering the network boundary
Each platform adapter uses a different HTTP client, so the interception mechanism differs:
| Platform | Adapter HTTP client | Interception | Helper |
|----------|---------------------|--------------|--------|
| Telegram | native `fetch` | replace `globalThis.fetch` | `installFetchStub` (replay-test-helpers) |
| Slack | `@slack/web-api` (axios) | `nock` at the HTTP layer | inline in slack `replay-test-context` |
| Linear | `@linear/sdk` (GraphQL over fetch) | replace `globalThis.fetch` | `installFetchStub` |
Responses are answered from two sources, in order of preference:
1. **Recorded data** where it matters — the captured webhook payload is replayed into the real
adapter as the inbound event, and recorded outbound bodies inform assertions.
2. **Minimal stubs** for incidental calls the recordings don't need to pin down — identity bootstrap
(`getMe`, `auth.test`, Linear `viewer`), streaming lifecycle, and entity look-ups.
The assertions check what the adapter **sends** (the outbound request body), not what it receives,
so response stubs only need to be valid enough for the real adapter to proceed.
#### Platform notes
- **Telegram** — `getMe` returns the bot fixture so the adapter learns its identity; `sendMessage`
returns a minimal message. `reply_markup` is sent as a JSON object (not a stringified blob), so
read inline-keyboard callback data via `getTelegramInlineCallbackData`.
- **Slack** — agent replies go through Slack's assistant **streaming** API
(`chat.startStream``appendStream``stopStream`), not `chat.postMessage`. The nock handler
reconstructs the streamed text and records it as a synthetic `chat.postMessage` so assertions can
treat the reply as one outbound post. `webhookVerifier: () => true` bypasses signature checks (the
fixtures carry sanitized signatures); passing `botUserId` skips the `auth.test` lookup.
- **Linear** — webhooks are HMAC-signed (`linear-signature`) and timestamp-checked, so the helper
refreshes `webhookTimestamp` and signs the body. `@linear/sdk` strictly deserializes typed
entities and lazily fetches relationships, so the GraphQL stub returns fully-shaped entities
(e.g. a `Comment` needs `reactions: []`; an `AgentActivity` references `agentSession`/`sourceComment`
by id). Linear's "mention" is an **agent-session** event, not a comment — see the contract note
below.
## Test Layout
```text
src/modules/agents/integrations/
__tests__/channel-integration-contract.test.ts
__tests__/fixtures/<platform>/
__tests__/helpers/<platform>/replay-test-context.ts
__tests__/helpers/<platform>/synthetic-fixtures.ts
__tests__/helpers/replay-test-helpers.ts # shared: createReplayContextSetup, installFetchStub, …
platforms/__tests__/<platform>/recorded-integration.test.ts
platforms/__tests__/<platform>/synthetic-integration.test.ts
```
Each `<platform>/replay-test-context.ts` builds the real adapter + real `Chat`, installs the
network interceptor, wires `AgentChatBridge` via `createReplayContextSetup`, and exposes
`sendWebhook`, `latestContext`, `lastPost`, `apiCalls`, and `shutdown` (which restores the
interceptor).
## Shared Contract Tests
Use `runSharedChannelIntegrationContract()` when a scenario should behave the same across platforms.
It verifies that an integration:
- Routes a mention or DM to `executeForChatPublished()`.
- Subscribes the thread and routes follow-up messages.
- Persists latest message context for the integration context tool.
- Responds through the integration action executor into the latest thread.
- Ignores messages authored by the connected bot.
> **Linear is intentionally not in the shared contract.** The real `@chat-adapter/linear` only treats
> agent-session events as mentions (a bare comment has `isMention = false`), and agent-session vs
> comment threads don't share an id — so the comment-as-mention + subscribe/follow-up contract
> doesn't model real Linear behavior. Linear's real flow is covered by its recorded agent-session
> test (`platforms/__tests__/linear/recorded-integration.test.ts`).
Assert the real adapter's actual output, not a simplified shape. For example, the message-context
`channelId` is platform-prefixed (`telegram:123456`, `slack:C_SUPPORT`), and Slack agent replies are
recorded with a `markdown_text` body.
## Synthetic Integration Tests
Use synthetic tests for cases that are hard to capture reliably or need narrow edge-case control:
platform-specific mention routing (Telegram group mentions), ignored messages (bot-authored), thread
identity rules (Telegram forum topics), tool-resume from cards/inline keyboards, action-executor
behavior (`send_dm`, `respond`).
Name each test as a scenario, not an implementation detail
(`'routes a Telegram group mention to a new agent conversation'`, not `'handles a Bot API update'`).
The assertion should prove integration behavior: the agent executor received the expected cleaned
message and integration type; the latest context has the expected platform target; the outbound
adapter call targets the right channel/thread; blocked messages don't call the agent executor.
## Recorded Replay Tests
Use recorded tests to lock down real webhook shapes against the real adapter. A recorded replay test:
1. Loads `recorded-session.json`.
2. Builds fixtures from the recorded webhook body and recorded bot metadata.
3. Sends the payload through the replay context webhook (driving the real adapter).
4. Asserts n8n behavior: agent execution, message context, and outbound request body.
Do not assert recorded fixture contents by themselves (e.g. `record.response.id`) — assert the
behavior n8n produced. Avoid overfitting to timestamps unless the timestamp is part of the thread or
message identity being tested.
> **Note on recording completeness.** The recordings primarily capture webhook **inputs**. The
> outbound side is answered by the interceptor stubs described above, because the recorder only
> captures `fetch` traffic (Slack uses axios) and adapter versions can drift from older recordings.
> If you want a recorded test to replay outbound responses verbatim, ensure the recording is current
> and complete for that adapter version.
## Recording Requests
Recording is controlled by `ChannelIntegrationRecorder` and the `recordAdapterCalls()` proxy. It can
capture `webhook` (incoming webhooks), `api-call` (adapter method calls), and `fetch` (outbound HTTP
to known platform APIs — note: `fetch` only, so axios-based clients like `@slack/web-api` are not
captured). Sensitive headers and token-shaped URLs are sanitized, but review every recording before
committing it.
To record during a local run:
```bash
export N8N_AGENT_INTEGRATION_RECORDING_ENABLED=true
export N8N_AGENT_INTEGRATION_RECORDING_SESSION_ID=telegram-basic
export N8N_AGENT_INTEGRATION_RECORDING_DIR="$PWD/.agent-recordings/channel-integrations"
pnpm dev # a real n8n instance with a real bot connected (real credentials)
# then: message the bot, trigger callbacks, let the agent reply
```
Export and install the session:
```bash
pnpm recording:list --recording-dir ~/Documents/my-recordings
pnpm recording:export telegram-basic --output-dir src/modules/agents/integrations/__tests__/fixtures/telegram
# review, sanitize, minimize → rename to recorded-session.json
```
## Checklist
Before committing a new channel integration test:
- The test name describes a user-visible scenario or routing rule.
- The test asserts n8n integration behavior against the **real** adapter's actual output.
- Shared behavior is covered in the contract test where the platform fits its model.
- Platform-specific behavior is covered in a synthetic or recorded platform test.
- Recorded fixtures are sanitized, minimal, and reviewed manually.
- The network interceptor is restored on teardown (`shutdown()`).
@@ -0,0 +1,94 @@
import { readFileSync } from 'fs';
import { jsonParse } from 'n8n-workflow';
import { join } from 'path';
import { runSharedChannelIntegrationContract } from './helpers/channel-integration-contract';
import {
createSlackReplayContext,
type SlackReplayFixtures,
} from './helpers/slack/replay-test-context';
import {
createTelegramReplayContext,
type TelegramReplayFixtures,
} from './helpers/telegram/replay-test-context';
const slackFixtures = jsonParse<SlackReplayFixtures>(
readFileSync(join(__dirname, 'fixtures/slack/basic.json'), 'utf8'),
);
const telegramFixtures = jsonParse<TelegramReplayFixtures>(
readFileSync(join(__dirname, 'fixtures/telegram/basic.json'), 'utf8'),
);
runSharedChannelIntegrationContract({
name: 'Slack',
fixtures: slackFixtures,
expected: {
message: 'hello agent',
followUpMessage: 'follow up',
integrationType: 'slack',
context: {
integrationConnectionId: 'slack:cred-slack',
platform: 'slack',
messageId: '1719000000.000100',
interactingUserId: 'U_ALICE',
agentUserId: 'U_BOT',
target: {
type: 'thread',
threadId: 'slack:C_SUPPORT:1719000000.000100',
channelId: 'slack:C_SUPPORT',
},
},
resourceId: 'U_ALICE',
firstPost: {
channel: 'C_SUPPORT',
thread_ts: '1719000000.000100',
markdown_text: 'Got it',
},
respondPost: {
channel: 'C_SUPPORT',
thread_ts: '1719000000.000100',
text: 'Action response',
},
respondTarget: { threadId: 'slack:C_SUPPORT:1719000000.000100' },
},
createContext: async () => await createSlackReplayContext(slackFixtures),
});
// NOTE: Linear is intentionally not in the shared contract. The real
// @chat-adapter/linear only treats agent-session events as mentions (a bare
// Comment has isMention=false), and agent-session vs comment threads don't share
// an id — so the comment-as-mention + subscribe/follow-up contract doesn't model
// real Linear behavior. Linear's real flow is covered by its recorded
// agent-session test (platforms/__tests__/linear/recorded-integration.test.ts).
runSharedChannelIntegrationContract({
name: 'Telegram',
fixtures: telegramFixtures,
expected: {
message: 'hello agent',
followUpMessage: 'follow up',
integrationType: 'telegram',
context: {
integrationConnectionId: 'telegram:cred-telegram',
platform: 'telegram',
messageId: '123456:11',
interactingUserId: '123456',
target: {
type: 'thread',
threadId: 'telegram:123456',
channelId: 'telegram:123456',
},
},
resourceId: '123456',
firstPost: {
chat_id: '123456',
text: 'Got it',
},
respondPost: {
chat_id: '123456',
text: 'Action response',
},
respondTarget: { threadId: 'telegram:123456' },
},
createContext: async () => await createTelegramReplayContext(telegramFixtures),
});
@@ -0,0 +1,153 @@
import { mkdtemp, rm, writeFile } from 'fs/promises';
import { tmpdir } from 'os';
import { join } from 'path';
import { ChannelIntegrationRecorder } from '../recording/channel-integration-recorder';
describe('ChannelIntegrationRecorder', () => {
let recordingDir: string;
beforeEach(async () => {
recordingDir = await mkdtemp(join(tmpdir(), 'n8n-channel-recordings-'));
});
afterEach(async () => {
await rm(recordingDir, { recursive: true, force: true });
});
it('sanitizes webhook URL and host headers when appending records', async () => {
const recorder = new ChannelIntegrationRecorder({
enabled: true,
sessionId: 'test-session',
recordingDir,
});
const headers = new Headers();
headers.set('host', 'localhost:5678');
headers.set('x-forwarded-for', '127.0.0.1');
headers.set('x-forwarded-host', 'dev-tunnel.example.com');
headers.set('x-telegram-bot-api-secret-token', 'secret-token');
headers.set('x-custom-header', 'kept');
await recorder.recordWebhook(
'telegram',
new Request(
'http://localhost:5678/rest/projects/project-1/agents/v2/agent-1/webhooks/telegram?foo=bar',
{
method: 'POST',
headers,
body: '{}',
},
),
);
const [record] = await recorder.getRecords('test-session');
expect(record).toMatchObject({
type: 'webhook',
url: 'https://n8n.host.com/rest/projects/project-1/agents/v2/agent-1/webhooks/telegram?foo=bar',
});
if (record.type !== 'webhook') throw new Error('Expected a webhook record');
expect(record.headers.host).toBe('https://n8n.host.com');
expect(record.headers['x-forwarded-for']).toBe('111.111.111.111');
expect(record.headers['x-forwarded-host']).toBe('https://n8n.host.com');
expect(record.headers['x-telegram-bot-api-secret-token']).toBe('[REDACTED]');
expect(record.headers['x-custom-header']).toBe('kept');
});
it('does not replace fetch URL or host headers with the sanitized n8n host', async () => {
const recorder = new ChannelIntegrationRecorder({
enabled: true,
sessionId: 'fetch-session',
recordingDir,
});
const originalFetch = globalThis.fetch;
globalThis.fetch = vi.fn(async () => {
await Promise.resolve();
return new Response(JSON.stringify({ ok: true, result: true }), {
status: 200,
headers: { host: 'api.telegram.org' },
});
});
try {
recorder.startFetchRecording([/api\.telegram\.org/i]);
const requestHeaders = new Headers();
requestHeaders.set('host', 'api.telegram.org');
requestHeaders.set('x-telegram-bot-api-secret-token', 'secret-token');
await fetch('https://api.telegram.org/bot123456:secret/sendMessage', {
method: 'POST',
headers: requestHeaders,
body: '{}',
});
recorder.stopFetchRecording();
const [record] = await recorder.getRecords('fetch-session');
expect(record).toMatchObject({
type: 'fetch',
url: 'https://api.telegram.org/bot123456789:abcdefghijkl/sendMessage',
});
if (record.type !== 'fetch') throw new Error('Expected a fetch record');
expect(record.requestHeaders?.host).toBe('api.telegram.org');
expect(record.requestHeaders?.['x-telegram-bot-api-secret-token']).toBe('[REDACTED]');
} finally {
recorder.stopFetchRecording();
globalThis.fetch = originalFetch;
}
});
it('records request metadata when fetch receives a Request input', async () => {
const recorder = new ChannelIntegrationRecorder({
enabled: true,
sessionId: 'request-fetch-session',
recordingDir,
});
const originalFetch = globalThis.fetch;
globalThis.fetch = vi.fn(async () => {
await Promise.resolve();
return new Response(JSON.stringify({ ok: true, result: true }), { status: 200 });
});
try {
recorder.startFetchRecording([/api\.telegram\.org/i]);
const requestHeaders = new Headers();
requestHeaders.set('x-telegram-bot-api-secret-token', 'secret-token');
await fetch(
new Request('https://api.telegram.org/bot123456:secret/sendMessage', {
method: 'PATCH',
headers: requestHeaders,
body: '{"text":"hello"}',
}),
);
recorder.stopFetchRecording();
const [record] = await recorder.getRecords('request-fetch-session');
if (record.type !== 'fetch') throw new Error('Expected a fetch record');
expect(record.method).toBe('PATCH');
expect(record.requestBody).toBe('{"text":"hello"}');
expect(record.requestHeaders?.['x-telegram-bot-api-secret-token']).toBe('[REDACTED]');
} finally {
recorder.stopFetchRecording();
globalThis.fetch = originalFetch;
}
});
it('does not fail the caller when writing a webhook record fails', async () => {
const filePath = join(recordingDir, 'not-a-directory');
await writeFile(filePath, 'x', 'utf8');
const recorder = new ChannelIntegrationRecorder({
enabled: true,
sessionId: 'best-effort-session',
recordingDir: filePath,
});
await expect(
recorder.recordWebhook(
'telegram',
new Request('https://n8n.example.com/webhook', {
method: 'POST',
body: '{}',
}),
),
).resolves.toBeUndefined();
});
});
@@ -0,0 +1,102 @@
{
"botUser": {
"id": "USER_BOT",
"name": "agentname",
"displayName": "AgentName",
"app": true
},
"user": {
"id": "USER_ALICE",
"name": "alice",
"displayName": "Alice Developer",
"email": "alice@example.com"
},
"issue": {
"id": "ISSUE_123",
"identifier": "ENG-123",
"title": "Investigate customer workflow",
"description": "Customer reported unexpected workflow behavior.",
"url": "https://linear.app/n8n/issue/ENG-123/investigate-customer-workflow"
},
"mention": {
"type": "Comment",
"action": "create",
"organizationId": "ORG_1",
"webhookId": "WEBHOOK_1",
"webhookTimestamp": 1719000000000,
"data": {
"id": "COMMENT_1",
"issueId": "ISSUE_123",
"body": "@AgentName hello agent",
"user": {
"id": "USER_ALICE",
"name": "alice",
"displayName": "Alice Developer",
"email": "alice@example.com"
},
"issue": {
"id": "ISSUE_123",
"identifier": "ENG-123",
"title": "Investigate customer workflow",
"description": "Customer reported unexpected workflow behavior.",
"url": "https://linear.app/n8n/issue/ENG-123/investigate-customer-workflow"
},
"createdAt": "2024-06-21T12:00:00.000Z",
"url": "https://linear.app/n8n/issue/ENG-123#comment-COMMENT_1"
}
},
"followUp": {
"type": "Comment",
"action": "create",
"organizationId": "ORG_1",
"webhookId": "WEBHOOK_1",
"webhookTimestamp": 1719000001000,
"data": {
"id": "COMMENT_2",
"issueId": "ISSUE_123",
"body": "follow up",
"user": {
"id": "USER_ALICE",
"name": "alice",
"displayName": "Alice Developer",
"email": "alice@example.com"
},
"issue": {
"id": "ISSUE_123",
"identifier": "ENG-123",
"title": "Investigate customer workflow",
"description": "Customer reported unexpected workflow behavior.",
"url": "https://linear.app/n8n/issue/ENG-123/investigate-customer-workflow"
},
"createdAt": "2024-06-21T12:00:01.000Z",
"url": "https://linear.app/n8n/issue/ENG-123#comment-COMMENT_2"
}
},
"selfMessage": {
"type": "Comment",
"action": "create",
"organizationId": "ORG_1",
"webhookId": "WEBHOOK_1",
"webhookTimestamp": 1719000002000,
"data": {
"id": "COMMENT_3",
"issueId": "ISSUE_123",
"body": "bot echo",
"user": {
"id": "USER_BOT",
"name": "agentname",
"displayName": "AgentName",
"app": true
},
"issue": {
"id": "ISSUE_123",
"identifier": "ENG-123",
"title": "Investigate customer workflow",
"description": "Customer reported unexpected workflow behavior.",
"url": "https://linear.app/n8n/issue/ENG-123/investigate-customer-workflow"
},
"createdAt": "2024-06-21T12:00:02.000Z",
"url": "https://linear.app/n8n/issue/ENG-123#comment-COMMENT_3"
}
}
}
@@ -0,0 +1,37 @@
[
{
"type": "webhook",
"timestamp": 1782476984606,
"platform": "linear",
"method": "POST",
"url": "https://n8n.host.com/rest/projects/project-1/agents/v2/agent-1/webhooks/linear",
"headers": {
"content-type": "application/json; charset=utf-8",
"host": "https://n8n.host.com",
"linear-event": "AgentSessionEvent",
"user-agent": "Linear-Webhook",
"x-forwarded-for": "111.111.111.111",
"x-forwarded-host": "https://n8n.host.com",
"x-forwarded-proto": "https"
},
"body": "{\"type\":\"AgentSessionEvent\",\"action\":\"created\",\"createdAt\":\"2026-06-26T12:29:43.714Z\",\"organizationId\":\"ORG_LINEAR\",\"appUserId\":\"USER_APP\",\"agentSession\":{\"id\":\"AGENT_SESSION_1\",\"createdAt\":\"2026-06-26T12:29:42.928Z\",\"updatedAt\":\"2026-06-26T12:29:42.928Z\",\"creatorId\":\"USER_ALICE\",\"appUserId\":\"USER_APP\",\"commentId\":\"COMMENT_SOURCE\",\"issueId\":\"ISSUE_1\",\"url\":\"https://linear.app/workspace/issue/YEH-1/get-familiar-with-linear#agent-session-AGENT_SESSION_1\",\"creator\":{\"id\":\"USER_ALICE\",\"name\":\"alice\",\"displayName\":\"Alice Developer\",\"email\":\"alice@example.com\",\"url\":\"https://linear.app/workspace/profiles/alice\"},\"comment\":{\"id\":\"COMMENT_SOURCE\",\"body\":\"@testapp hey\",\"userId\":\"USER_ALICE\",\"issueId\":\"ISSUE_1\"},\"issue\":{\"id\":\"ISSUE_1\",\"title\":\"Get familiar with Linear\",\"teamId\":\"TEAM_1\",\"team\":{\"id\":\"TEAM_1\",\"key\":\"YEH\",\"name\":\"Recorded Team\"},\"identifier\":\"YEH-1\",\"url\":\"https://linear.app/workspace/issue/YEH-1/get-familiar-with-linear\",\"description\":\"Welcome to Linear!\"}},\"previousComments\":[{\"id\":\"COMMENT_SOURCE\",\"body\":\"@testapp hey\",\"userId\":\"USER_ALICE\",\"issueId\":\"ISSUE_1\"}],\"guidance\":null,\"promptContext\":\"<issue identifier=\\\"YEH-1\\\"><title>Get familiar with Linear</title></issue>\"}"
},
{
"type": "fetch",
"timestamp": 1782476990071,
"method": "POST",
"url": "https://api.linear.app/graphql",
"durationMs": 268,
"requestHeaders": {
"Content-Type": "application/json",
"Authorization": "[REDACTED]",
"User-Agent": "n8n-monorepo@2.28.0"
},
"requestBody": "{\"query\":\"mutation createAgentActivity($input: AgentActivityCreateInput!) { agentActivityCreate(input: $input) { success } }\",\"variables\":{\"input\":{\"agentSessionId\":\"AGENT_SESSION_1\",\"content\":{\"type\":\"response\",\"body\":\"Hey! I'm here and ready to help.\"}}}}",
"status": 200,
"responseHeaders": {
"content-type": "application/json; charset=utf-8"
},
"responseBody": "{\"data\":{\"agentActivityCreate\":{\"agentActivity\":{\"id\":\"AGENT_ACTIVITY_1\"},\"lastSyncId\":2170157622,\"success\":true}}}"
}
]
@@ -0,0 +1,52 @@
{
"botUserId": "U_BOT",
"user": {
"id": "U_ALICE",
"name": "alice",
"real_name": "Alice Developer"
},
"channel": {
"id": "C_SUPPORT",
"name": "support"
},
"mention": {
"type": "event_callback",
"team_id": "T_TEAM",
"event": {
"type": "app_mention",
"user": "U_ALICE",
"text": "<@U_BOT> hello agent",
"ts": "1719000000.000100",
"thread_ts": "1719000000.000100",
"channel": "C_SUPPORT",
"channel_type": "channel"
}
},
"followUp": {
"type": "event_callback",
"team_id": "T_TEAM",
"event": {
"type": "message",
"user": "U_ALICE",
"text": "follow up",
"ts": "1719000001.000200",
"thread_ts": "1719000000.000100",
"channel": "C_SUPPORT",
"channel_type": "channel"
}
},
"selfMessage": {
"type": "event_callback",
"team_id": "T_TEAM",
"event": {
"type": "message",
"user": "U_BOT",
"bot_id": "B_BOT",
"text": "bot echo",
"ts": "1719000002.000300",
"thread_ts": "1719000000.000100",
"channel": "C_SUPPORT",
"channel_type": "channel"
}
}
}
@@ -0,0 +1,215 @@
[
{
"type": "webhook",
"timestamp": 1782378393751,
"platform": "slack",
"method": "POST",
"url": "https://n8n.host.com/rest/projects/project-1/agents/v2/agent-1/webhooks/slack",
"headers": {
"accept": "*/*",
"accept-encoding": "gzip,deflate",
"content-length": "758",
"content-type": "application/json",
"host": "https://n8n.host.com",
"user-agent": "Slackbot 1.0 (+https://api.slack.com/robots)",
"x-forwarded-for": "https://n8n.host.com",
"x-forwarded-host": "https://n8n.host.com",
"x-forwarded-proto": "https",
"x-slack-request-timestamp": "1782378393",
"x-slack-signature": "[REDACTED]"
},
"body": "{\"token\":\"SLACK_VERIFICATION_TOKEN\",\"team_id\":\"T_TEAM\",\"context_team_id\":\"T_TEAM\",\"context_enterprise_id\":null,\"api_app_id\":\"A_APP\",\"event\":{\"type\":\"message\",\"subtype\":\"channel_join\",\"user\":\"U_BOT\",\"text\":\"<@U_BOT> has joined the channel\",\"inviter\":\"U_USER\",\"ts\":\"1782378393.393439\",\"channel\":\"C_CHANNEL\",\"event_ts\":\"1782378393.393439\",\"channel_type\":\"channel\"},\"type\":\"event_callback\",\"event_id\":\"Ev_CHANNEL_JOIN\",\"event_time\":1782378393,\"authorizations\":[{\"enterprise_id\":null,\"team_id\":\"T_TEAM\",\"user_id\":\"U_BOT\",\"is_bot\":true,\"is_enterprise_install\":false}],\"is_ext_shared_channel\":false,\"event_context\":\"mock-channel-join-context\"}"
},
{
"type": "webhook",
"timestamp": 1782378398889,
"platform": "slack",
"method": "POST",
"url": "https://n8n.host.com/rest/projects/project-1/agents/v2/agent-1/webhooks/slack",
"headers": {
"accept": "*/*",
"accept-encoding": "gzip,deflate",
"content-length": "758",
"content-type": "application/json",
"host": "https://n8n.host.com",
"user-agent": "Slackbot 1.0 (+https://api.slack.com/robots)",
"x-forwarded-for": "https://n8n.host.com",
"x-forwarded-host": "https://n8n.host.com",
"x-forwarded-proto": "https",
"x-slack-request-timestamp": "1782378396",
"x-slack-retry-num": "1",
"x-slack-retry-reason": "http_timeout",
"x-slack-signature": "[REDACTED]"
},
"body": "{\"token\":\"SLACK_VERIFICATION_TOKEN\",\"team_id\":\"T_TEAM\",\"context_team_id\":\"T_TEAM\",\"context_enterprise_id\":null,\"api_app_id\":\"A_APP\",\"event\":{\"type\":\"message\",\"subtype\":\"channel_join\",\"user\":\"U_BOT\",\"text\":\"<@U_BOT> has joined the channel\",\"inviter\":\"U_USER\",\"ts\":\"1782378393.393439\",\"channel\":\"C_CHANNEL\",\"event_ts\":\"1782378393.393439\",\"channel_type\":\"channel\"},\"type\":\"event_callback\",\"event_id\":\"Ev_CHANNEL_JOIN\",\"event_time\":1782378393,\"authorizations\":[{\"enterprise_id\":null,\"team_id\":\"T_TEAM\",\"user_id\":\"U_BOT\",\"is_bot\":true,\"is_enterprise_install\":false}],\"is_ext_shared_channel\":false,\"event_context\":\"mock-channel-join-context\"}"
},
{
"type": "webhook",
"timestamp": 1782378398886,
"platform": "slack",
"method": "POST",
"url": "https://n8n.host.com/rest/projects/project-1/agents/v2/agent-1/webhooks/slack",
"headers": {
"accept": "*/*",
"accept-encoding": "gzip,deflate",
"content-length": "944",
"content-type": "application/json",
"host": "https://n8n.host.com",
"user-agent": "Slackbot 1.0 (+https://api.slack.com/robots)",
"x-forwarded-for": "https://n8n.host.com",
"x-forwarded-host": "https://n8n.host.com",
"x-forwarded-proto": "https",
"x-slack-request-timestamp": "1782378393",
"x-slack-signature": "[REDACTED]"
},
"body": "{\"token\":\"SLACK_VERIFICATION_TOKEN\",\"team_id\":\"T_TEAM\",\"api_app_id\":\"A_APP\",\"event\":{\"type\":\"app_mention\",\"user\":\"U_USER\",\"ts\":\"1782378390.841549\",\"client_msg_id\":\"mock-client-message-id\",\"text\":\"<@U_BOT> hey\",\"team\":\"T_TEAM\",\"blocks\":[{\"type\":\"rich_text\",\"block_id\":\"mention-block\",\"elements\":[{\"type\":\"rich_text_section\",\"elements\":[{\"type\":\"user\",\"user_id\":\"U_BOT\"},{\"type\":\"text\",\"text\":\" hey\"}]}]}],\"channel\":\"C_CHANNEL\",\"action_token\":\"mock-action-token\",\"event_ts\":\"1782378390.841549\"},\"type\":\"event_callback\",\"event_id\":\"Ev_APP_MENTION\",\"event_time\":1782378390,\"authorizations\":[{\"enterprise_id\":null,\"team_id\":\"T_TEAM\",\"user_id\":\"U_BOT\",\"is_bot\":true,\"is_enterprise_install\":false}],\"is_ext_shared_channel\":false,\"event_context\":\"mock-app-mention-context\"}"
},
{
"type": "webhook",
"timestamp": 1782378398892,
"platform": "slack",
"method": "POST",
"url": "https://n8n.host.com/rest/projects/project-1/agents/v2/agent-1/webhooks/slack",
"headers": {
"accept": "*/*",
"accept-encoding": "gzip,deflate",
"content-length": "944",
"content-type": "application/json",
"host": "https://n8n.host.com",
"user-agent": "Slackbot 1.0 (+https://api.slack.com/robots)",
"x-forwarded-for": "https://n8n.host.com",
"x-forwarded-host": "https://n8n.host.com",
"x-forwarded-proto": "https",
"x-slack-request-timestamp": "1782378397",
"x-slack-retry-num": "1",
"x-slack-retry-reason": "http_timeout",
"x-slack-signature": "[REDACTED]"
},
"body": "{\"token\":\"SLACK_VERIFICATION_TOKEN\",\"team_id\":\"T_TEAM\",\"api_app_id\":\"A_APP\",\"event\":{\"type\":\"app_mention\",\"user\":\"U_USER\",\"ts\":\"1782378390.841549\",\"client_msg_id\":\"mock-client-message-id\",\"text\":\"<@U_BOT> hey\",\"team\":\"T_TEAM\",\"blocks\":[{\"type\":\"rich_text\",\"block_id\":\"mention-block\",\"elements\":[{\"type\":\"rich_text_section\",\"elements\":[{\"type\":\"user\",\"user_id\":\"U_BOT\"},{\"type\":\"text\",\"text\":\" hey\"}]}]}],\"channel\":\"C_CHANNEL\",\"action_token\":\"mock-action-token\",\"event_ts\":\"1782378390.841549\"},\"type\":\"event_callback\",\"event_id\":\"Ev_APP_MENTION\",\"event_time\":1782378390,\"authorizations\":[{\"enterprise_id\":null,\"team_id\":\"T_TEAM\",\"user_id\":\"U_BOT\",\"is_bot\":true,\"is_enterprise_install\":false}],\"is_ext_shared_channel\":false,\"event_context\":\"mock-app-mention-context\"}"
},
{
"type": "api-call",
"timestamp": 1782378403298,
"platform": "slack",
"method": "postMessage",
"args": [
"slack:C_CHANNEL:1782378390.841549",
{
"markdown": "Hey! 👋 How can I help you today?"
}
],
"response": {
"id": "1782378403.317389",
"threadId": "slack:C_CHANNEL:1782378390.841549",
"raw": {
"ok": true,
"channel": "C_CHANNEL",
"ts": "1782378403.317389",
"message": {
"user": "U_BOT",
"type": "message",
"ts": "1782378403.317389",
"bot_id": "B_BOT",
"app_id": "A_APP",
"text": "Hey! :wave: How can I help you today?",
"team": "T_TEAM",
"thread_ts": "1782378390.841549",
"parent_user_id": "U_USER"
}
}
}
},
{
"type": "webhook",
"timestamp": 1782378403654,
"platform": "slack",
"method": "POST",
"url": "https://n8n.host.com/rest/projects/project-1/agents/v2/agent-1/webhooks/slack",
"headers": {
"accept": "*/*",
"accept-encoding": "gzip,deflate",
"content-length": "1489",
"content-type": "application/json",
"host": "https://n8n.host.com",
"user-agent": "Slackbot 1.0 (+https://api.slack.com/robots)",
"x-forwarded-for": "https://n8n.host.com",
"x-forwarded-host": "https://n8n.host.com",
"x-forwarded-proto": "https",
"x-slack-request-timestamp": "1782378403",
"x-slack-signature": "[REDACTED]"
},
"body": "{\"token\":\"SLACK_VERIFICATION_TOKEN\",\"team_id\":\"T_TEAM\",\"context_team_id\":\"T_TEAM\",\"context_enterprise_id\":null,\"api_app_id\":\"A_APP\",\"event\":{\"type\":\"message\",\"user\":\"U_BOT\",\"ts\":\"1782378403.317389\",\"bot_id\":\"B_BOT\",\"app_id\":\"A_APP\",\"text\":\"Hey! :wave: How can I help you today?\",\"team\":\"T_TEAM\",\"bot_profile\":{\"id\":\"B_BOT\",\"deleted\":false,\"name\":\"Test agent\",\"updated\":1782378094,\"app_id\":\"A_APP\",\"user_id\":\"U_BOT\",\"icons\":{},\"team_id\":\"T_TEAM\"},\"thread_ts\":\"1782378390.841549\",\"parent_user_id\":\"U_USER\",\"blocks\":[],\"channel\":\"C_CHANNEL\",\"event_ts\":\"1782378403.317389\",\"channel_type\":\"channel\"},\"type\":\"event_callback\",\"event_id\":\"Ev_BOT_CHANNEL_RESPONSE\",\"event_time\":1782378403,\"authorizations\":[{\"enterprise_id\":null,\"team_id\":\"T_TEAM\",\"user_id\":\"U_BOT\",\"is_bot\":true,\"is_enterprise_install\":false}],\"is_ext_shared_channel\":false,\"event_context\":\"mock-bot-channel-response-context\"}"
},
{
"type": "webhook",
"timestamp": 1782379186355,
"platform": "slack",
"method": "POST",
"url": "https://n8n.host.com/rest/projects/project-1/agents/v2/agent-1/webhooks/slack",
"headers": {
"accept": "*/*",
"accept-encoding": "gzip,deflate",
"content-length": "1012",
"content-type": "application/json",
"host": "https://n8n.host.com",
"user-agent": "Slackbot 1.0 (+https://api.slack.com/robots)",
"x-forwarded-for": "https://n8n.host.com",
"x-forwarded-host": "https://n8n.host.com",
"x-forwarded-proto": "https",
"x-slack-request-timestamp": "1782379186",
"x-slack-signature": "[REDACTED]"
},
"body": "{\"token\":\"SLACK_VERIFICATION_TOKEN\",\"team_id\":\"T_TEAM\",\"context_team_id\":\"T_TEAM\",\"context_enterprise_id\":null,\"api_app_id\":\"A_APP\",\"event\":{\"type\":\"message\",\"user\":\"U_USER\",\"ts\":\"1782379185.654229\",\"client_msg_id\":\"mock-dm-client-message-id\",\"text\":\"DM message. What's your name?\",\"team\":\"T_TEAM\",\"blocks\":[],\"channel\":\"D_DM\",\"action_token\":\"mock-dm-action-token\",\"event_ts\":\"1782379185.654229\",\"channel_type\":\"im\"},\"type\":\"event_callback\",\"event_id\":\"Ev_DM_MESSAGE\",\"event_time\":1782379185,\"authorizations\":[{\"enterprise_id\":null,\"team_id\":\"T_TEAM\",\"user_id\":\"U_BOT\",\"is_bot\":true,\"is_enterprise_install\":false}],\"is_ext_shared_channel\":false,\"event_context\":\"mock-dm-message-context\"}"
},
{
"type": "api-call",
"timestamp": 1782379192134,
"platform": "slack",
"method": "postMessage",
"args": [
"slack:D_DM:",
{
"markdown": "I'm Assistant."
}
],
"response": {
"id": "1782379192.163089",
"threadId": "slack:D_DM:",
"raw": {
"ok": true,
"channel": "D_DM",
"ts": "1782379192.163089",
"message": {
"user": "U_BOT",
"type": "message",
"ts": "1782379192.163089",
"bot_id": "B_BOT",
"app_id": "A_APP",
"text": "I'm Assistant.",
"team": "T_TEAM"
}
}
}
},
{
"type": "webhook",
"timestamp": 1782379192852,
"platform": "slack",
"method": "POST",
"url": "https://n8n.host.com/rest/projects/project-1/agents/v2/agent-1/webhooks/slack",
"headers": {
"accept": "*/*",
"accept-encoding": "gzip,deflate",
"content-length": "1991",
"content-type": "application/json",
"host": "https://n8n.host.com",
"user-agent": "Slackbot 1.0 (+https://api.slack.com/robots)",
"x-forwarded-for": "https://n8n.host.com",
"x-forwarded-host": "https://n8n.host.com",
"x-forwarded-proto": "https",
"x-slack-request-timestamp": "1782379192",
"x-slack-signature": "[REDACTED]"
},
"body": "{\"token\":\"SLACK_VERIFICATION_TOKEN\",\"team_id\":\"T_TEAM\",\"context_team_id\":\"T_TEAM\",\"context_enterprise_id\":null,\"api_app_id\":\"A_APP\",\"event\":{\"type\":\"message\",\"user\":\"U_BOT\",\"ts\":\"1782379192.163089\",\"bot_id\":\"B_BOT\",\"app_id\":\"A_APP\",\"text\":\"I'm Assistant.\",\"team\":\"T_TEAM\",\"bot_profile\":{\"id\":\"B_BOT\",\"deleted\":false,\"name\":\"Test agent\",\"updated\":1782378094,\"app_id\":\"A_APP\",\"user_id\":\"U_BOT\",\"icons\":{},\"team_id\":\"T_TEAM\"},\"blocks\":[],\"channel\":\"D_DM\",\"event_ts\":\"1782379192.163089\",\"channel_type\":\"im\"},\"type\":\"event_callback\",\"event_id\":\"Ev_BOT_DM_RESPONSE\",\"event_time\":1782379192,\"authorizations\":[{\"enterprise_id\":null,\"team_id\":\"T_TEAM\",\"user_id\":\"U_BOT\",\"is_bot\":true,\"is_enterprise_install\":false}],\"is_ext_shared_channel\":false,\"event_context\":\"mock-bot-dm-response-context\"}"
}
]
@@ -0,0 +1,111 @@
{
"bot": {
"id": 777000,
"is_bot": true,
"first_name": "n8n Agent",
"username": "n8n_agent_bot"
},
"user": {
"id": 123456,
"is_bot": false,
"first_name": "Alice",
"username": "alice_dev"
},
"chat": {
"id": 123456,
"type": "private",
"first_name": "Alice",
"username": "alice_dev"
},
"mention": {
"update_id": 10001,
"message": {
"message_id": 11,
"from": {
"id": 123456,
"is_bot": false,
"first_name": "Alice",
"username": "alice_dev"
},
"chat": {
"id": 123456,
"type": "private",
"first_name": "Alice",
"username": "alice_dev"
},
"date": 1719000000,
"text": "hello agent"
}
},
"followUp": {
"update_id": 10002,
"message": {
"message_id": 12,
"from": {
"id": 123456,
"is_bot": false,
"first_name": "Alice",
"username": "alice_dev"
},
"chat": {
"id": 123456,
"type": "private",
"first_name": "Alice",
"username": "alice_dev"
},
"date": 1719000001,
"text": "follow up"
}
},
"selfMessage": {
"update_id": 10003,
"message": {
"message_id": 13,
"from": {
"id": 777000,
"is_bot": true,
"first_name": "n8n Agent",
"username": "n8n_agent_bot"
},
"chat": {
"id": 123456,
"type": "private",
"first_name": "Alice",
"username": "alice_dev"
},
"date": 1719000002,
"text": "bot echo"
}
},
"callbackBase": {
"update_id": 10004,
"callback_query": {
"id": "callback-1",
"from": {
"id": 123456,
"is_bot": false,
"first_name": "Alice",
"username": "alice_dev"
},
"message": {
"message_id": 1000,
"from": {
"id": 777000,
"is_bot": true,
"first_name": "n8n Agent",
"username": "n8n_agent_bot"
},
"chat": {
"id": 123456,
"type": "private",
"first_name": "Alice",
"username": "alice_dev"
},
"date": 1719000003,
"text": "Approval required"
},
"chat_instance": "chat-instance-1",
"data": "placeholder"
}
}
}
@@ -0,0 +1,102 @@
[
{
"type": "fetch",
"timestamp": 1782371203481,
"method": "POST",
"url": "https://api.telegram.org/bot123456789:abcdefghijkl/getMe",
"durationMs": 31,
"requestHeaders": {
"Content-Type": "application/json"
},
"requestBody": "{}",
"status": 200,
"responseHeaders": {
"access-control-allow-methods": "GET, POST, OPTIONS",
"access-control-allow-origin": "*",
"access-control-expose-headers": "Content-Length,Content-Type,Date,Server,Connection",
"connection": "keep-alive",
"content-length": "419",
"content-type": "application/json",
"date": "Thu, 25 Jun 2026 07:06:43 GMT",
"server": "nginx/1.30.1",
"strict-transport-security": "max-age=31536000; includeSubDomains; preload"
},
"responseBody": "{\"ok\":true,\"result\":{\"id\":987654321,\"is_bot\":true,\"first_name\":\"TestBot\",\"username\":\"test_bot\",\"can_join_groups\":true,\"can_read_all_group_messages\":false,\"supports_inline_queries\":false,\"supports_guest_queries\":false,\"can_connect_to_business\":false,\"has_main_web_app\":false,\"has_topics_enabled\":false,\"allows_users_to_create_topics\":false,\"can_manage_bots\":false,\"supports_join_request_queries\":false}}"
},
{
"type": "webhook",
"timestamp": 1782371293351,
"platform": "telegram",
"method": "POST",
"url": "https://n8n.host.com/rest/projects/n1hfKHngCcLMKhwm/agents/v2/cUXaKtBmyOGVvl9j/webhooks/telegram",
"headers": {
"accept-encoding": "gzip, deflate",
"content-length": "337",
"content-type": "application/json",
"host": "n8n.host.com",
"x-forwarded-for": "11.111.1.111",
"x-forwarded-host": "n8n.host.com",
"x-forwarded-proto": "https",
"x-telegram-bot-api-secret-token": "[REDACTED]"
},
"body": "{\"update_id\":202827072,\n\"message\":{\"message_id\":178,\"from\":{\"id\":123456789,\"is_bot\":false,\"first_name\":\"Sofia\",\"last_name\":\"García \\ud83c\\udf3f\",\"username\":\"sofiadev\",\"language_code\":\"en\"},\"chat\":{\"id\":123456789,\"first_name\":\"Sofia\",\"last_name\":\"García \\ud83c\\udf3f\",\"username\":\"sofiadev\",\"type\":\"private\"},\"date\":1782371215,\"text\":\"hey\"}}"
},
{
"type": "fetch",
"timestamp": 1782371303784,
"method": "POST",
"url": "https://api.telegram.org/bot123456789:abcdefghijkl/sendMessage",
"durationMs": 4304,
"requestHeaders": {
"Content-Type": "application/json"
},
"requestBody": "{\"chat_id\":\"123456789\",\"text\":\"Test response\",\"parse_mode\":\"MarkdownV2\"}",
"status": 200,
"responseHeaders": {
"access-control-allow-methods": "GET, POST, OPTIONS",
"access-control-allow-origin": "*",
"access-control-expose-headers": "Content-Length,Content-Type,Date,Server,Connection",
"connection": "keep-alive",
"content-length": "489",
"content-type": "application/json",
"date": "Thu, 25 Jun 2026 07:08:24 GMT",
"server": "nginx/1.30.1",
"strict-transport-security": "max-age=31536000; includeSubDomains; preload"
},
"responseBody": "{\"ok\":true,\"result\":{\"message_id\":179,\"from\":{\"id\":987654321,\"is_bot\":true,\"first_name\":\"TestBot\",\"username\":\"test_bot\"},\"chat\":{\"id\":123456789,\"first_name\":\"Sofia\",\"last_name\":\"García \\ud83c\\udf3f\",\"username\":\"sofiadev\",\"type\":\"private\"},\"date\":1782371304,\"text\":\"Test response\"}}"
},
{
"type": "api-call",
"timestamp": 1782371305143,
"platform": "telegram",
"method": "postMessage",
"args": [
"telegram:123456789",
{
"markdown": "Test response"
}
],
"response": {
"id": "123456789:179",
"threadId": "telegram:123456789",
"raw": {
"message_id": 179,
"from": {
"id": 987654321,
"is_bot": true,
"first_name": "TestBot",
"username": "test_bot"
},
"chat": {
"id": 123456789,
"first_name": "Sofia",
"last_name": "García 🌿",
"username": "sofiadev",
"type": "private"
},
"date": 1782371304,
"text": "Test response"
}
}
}
]
@@ -0,0 +1,135 @@
import type { Mock } from 'vitest';
import type { ChatIntegrationActionExecutor } from '../../integration-action-executor';
import { createIntegrationContextTool } from '../../integration-tools';
import type {
IntegrationMessageContext,
IntegrationMessageContextStore,
IntegrationToolConnectionDescriptor,
} from '../../integration-tools';
export interface ChannelIntegrationReplayScenario {
name: string;
fixtures: {
mention: unknown;
followUp: unknown;
selfMessage: unknown;
};
expected: {
message: string;
followUpMessage: string;
integrationType: string;
context: Partial<IntegrationMessageContext>;
resourceId: string;
firstPost: Record<string, unknown>;
respondPost: Record<string, unknown>;
respondTarget: Record<string, unknown>;
};
createContext: () => Promise<ChannelIntegrationReplayContext>;
}
export interface ChannelIntegrationReplayContext {
agentExecutor: {
executeForChatPublished: Mock;
};
actionExecutor: ChatIntegrationActionExecutor;
descriptor: IntegrationToolConnectionDescriptor;
messageContextStore: IntegrationMessageContextStore;
sendWebhook: (payload: unknown) => Promise<Response>;
latestContext: () => IntegrationMessageContext | undefined;
latestThreadId: () => string | undefined;
lastPost: () => { body: Record<string, unknown> } | undefined;
shutdown: () => Promise<void>;
}
export function runSharedChannelIntegrationContract(scenario: ChannelIntegrationReplayScenario) {
describe(`${scenario.name} shared channel integration contract`, () => {
let ctx: ChannelIntegrationReplayContext;
afterEach(async () => {
await ctx?.shutdown();
});
it('handles mention, subscribes the thread, and routes follow-up messages', async () => {
ctx = await scenario.createContext();
await expect(ctx.sendWebhook(scenario.fixtures.mention)).resolves.toMatchObject({
status: 200,
});
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenCalledTimes(1);
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenCalledWith(
expect.objectContaining({
agentId: 'agent-1',
projectId: 'project-1',
message: scenario.expected.message,
integrationType: scenario.expected.integrationType,
}),
);
expect(ctx.lastPost()?.body).toMatchObject(scenario.expected.firstPost);
await ctx.sendWebhook(scenario.fixtures.followUp);
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenCalledTimes(2);
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenLastCalledWith(
expect.objectContaining({ message: scenario.expected.followUpMessage }),
);
});
it('persists current message context for the integration context tool', async () => {
ctx = await scenario.createContext();
await ctx.sendWebhook(scenario.fixtures.mention);
const context = ctx.latestContext();
expect(context).toMatchObject(scenario.expected.context);
const threadId = ctx.latestThreadId();
if (!threadId) throw new Error('Expected a latest thread ID');
const contextTool = createIntegrationContextTool({
descriptor: ctx.descriptor,
queryExecutor: {
execute: vi.fn(),
},
messageContextStore: ctx.messageContextStore,
}).build();
const result = await contextTool.handler!(
{ query: 'get_current_message_context', input: {} },
{ persistence: { threadId, resourceId: scenario.expected.resourceId } },
);
expect(result).toEqual({ ok: true, context });
});
it('responds in the latest thread through the integration action executor', async () => {
ctx = await scenario.createContext();
await ctx.sendWebhook(scenario.fixtures.mention);
const context = ctx.latestContext();
expect(context).toMatchObject(scenario.expected.context);
const result = await ctx.actionExecutor.execute({
descriptor: ctx.descriptor,
action: 'respond',
input: { message: { text: 'Action response' } },
awaitResponse: false,
currentMessageContext: context,
});
expect(result).toMatchObject({
ok: true,
messageContext: {
platform: scenario.expected.integrationType,
target: scenario.expected.respondTarget,
},
});
expect(ctx.lastPost()?.body).toMatchObject(scenario.expected.respondPost);
});
it('ignores messages authored by the connected bot', async () => {
ctx = await scenario.createContext();
await ctx.sendWebhook(scenario.fixtures.selfMessage);
expect(ctx.agentExecutor.executeForChatPublished).not.toHaveBeenCalled();
});
});
}
@@ -0,0 +1,314 @@
import type { StreamChunk } from '@n8n/agents';
import type { AgentIntegrationConfig } from '@n8n/api-types';
import type { Logger as BackendLogger } from '@n8n/backend-common';
import type { OutboundHttp } from '@n8n/backend-network';
import { createHmac } from 'crypto';
import type { Mock } from 'vitest';
import { mock } from 'vitest-mock-extended';
import type { ChatInstance } from '../../../chat-integration.service';
import { ComponentMapper } from '../../../component-mapper';
import type {
getIntegrationToolConnectionDescriptors,
IntegrationMessageContext,
} from '../../../integration-tools';
import { LinearIntegration } from '../../../platforms/linear-integration';
import {
createReplayContextSetup,
installFetchStub,
type MemoryMessageContextStore,
type ReplayApiCall,
type ReplayContextSetup,
type ReplayWebhookHandler,
sendJsonWebhook,
} from '../replay-test-helpers';
export interface LinearUserFixture {
id: string;
name: string;
displayName: string;
email?: string;
url?: string;
app?: boolean;
}
export interface LinearIssueFixture {
id: string;
identifier: string;
title: string;
description?: string;
url?: string;
team?: { id: string; key: string; name: string };
}
export interface LinearAgentSessionFixture {
id: string;
creatorId: string;
appUserId: string;
commentId: string;
issueId: string;
url?: string;
creator: LinearUserFixture;
comment: {
id: string;
body: string;
userId: string;
issueId: string;
};
issue: LinearIssueFixture;
}
export interface LinearCommentFixture {
id: string;
body: string;
user: LinearUserFixture;
issue: LinearIssueFixture;
createdAt: string;
updatedAt?: string;
url?: string;
}
export interface LinearAgentSessionEventFixture {
type: 'AgentSessionEvent';
action: 'created';
createdAt: string;
organizationId: string;
appUserId: string;
agentSession: LinearAgentSessionFixture;
}
export interface LinearCommentEventFixture {
type: 'Comment';
action: 'create';
organizationId: string;
webhookId: string;
webhookTimestamp: number;
data: LinearCommentFixture;
}
export interface LinearReplayFixtures {
botUser: LinearUserFixture;
mention: LinearAgentSessionEventFixture | LinearCommentEventFixture;
followUp?: LinearCommentEventFixture;
selfMessage?: LinearCommentEventFixture;
}
export type LinearApiCall = ReplayApiCall;
export interface LinearReplayContext extends Omit<ReplayContextSetup, 'nextStream' | 'chat'> {
chat: ChatInstance;
agentExecutor: {
executeForChatPublished: Mock;
resumeForChat: Mock;
};
apiCalls: LinearApiCall[];
descriptor: ReturnType<typeof getIntegrationToolConnectionDescriptors>[number];
integration: AgentIntegrationConfig;
messageContextStore: MemoryMessageContextStore;
sendWebhook: (payload: unknown) => Promise<Response>;
latestContext: () => IntegrationMessageContext | undefined;
latestThreadId: () => string | undefined;
lastPost: () => LinearApiCall | undefined;
}
const LINEAR_ACCESS_TOKEN = 'lin_test-access-token';
const LINEAR_WEBHOOK_SECRET = 'test-webhook-secret';
const POST_METHODS = new Set(['agentActivityCreate', 'createComment']);
/**
* Builds the canned `@linear/sdk` entities the adapter re-reads after posting.
* After `agentActivityCreate` the adapter resolves `agentActivity.sourceComment`
* and that comment's `user`, throwing if any are missing — so responses must be
* fully shaped (and dates must be ISO strings, which the SDK wraps as `Date`).
*/
function buildLinearStubData(fixtures: LinearReplayFixtures, botUser: LinearUserFixture) {
const session =
fixtures.mention.type === 'AgentSessionEvent' ? fixtures.mention.agentSession : undefined;
const createdAt = '2026-01-01T00:00:00.000Z';
const user = session
? {
id: session.creator.id,
displayName: session.creator.displayName,
name: session.creator.name,
email: session.creator.email ?? null,
avatarUrl: null,
}
: { id: botUser.id, displayName: botUser.displayName, name: botUser.name, email: null };
const sourceComment = session
? {
id: session.comment.id,
body: session.comment.body,
userId: session.comment.userId,
user,
createdAt,
updatedAt: createdAt,
parentId: null,
url: null,
// `@linear/sdk`'s Comment constructor maps over `reactions` unguarded.
reactions: [],
}
: undefined;
// `agentSessionId`/`sourceComment` are getters over nested id refs, and the
// SDK fetches the comment by id — so reference them by id, not inline.
const agentActivity = {
id: 'AGENT_ACTIVITY_1',
agentSession: { id: session?.id ?? 'AGENT_SESSION_1' },
sourceComment: { id: session?.comment.id ?? 'COMMENT_SOURCE' },
};
return { user, sourceComment, agentActivity };
}
/**
* Map a Linear GraphQL operation to a method name + canned response. Query
* responses must be non-empty entities — `@linear/sdk` wraps them in typed
* objects and reads fields like `archivedAt`/`sourceComment`, which throw on
* `undefined`.
*/
function resolveLinearOperation(
query: string,
variables: { id?: unknown },
stub: ReturnType<typeof buildLinearStubData>,
botUser: LinearUserFixture,
organizationId: string,
): { method: string; data: Record<string, unknown> } {
if (query.includes('agentActivityCreate')) {
return {
method: 'agentActivityCreate',
data: {
agentActivityCreate: { agentActivity: stub.agentActivity, lastSyncId: 1, success: true },
},
};
}
if (/commentCreate/i.test(query)) {
return {
method: 'createComment',
data: {
commentCreate: {
comment: stub.sourceComment ?? { id: 'COMMENT_NEW' },
lastSyncId: 1,
success: true,
},
},
};
}
if (/\bagentActivity\(/.test(query)) {
return { method: 'agentActivity', data: { agentActivity: stub.agentActivity } };
}
if (/\bcomment\(/.test(query)) {
return { method: 'comment', data: { comment: stub.sourceComment } };
}
if (/\buser\(/.test(query)) {
return { method: 'user', data: { user: stub.user } };
}
if (/\bissue\(/.test(query)) {
return { method: 'issue', data: { issue: { id: String(variables.id ?? 'ISSUE') } } };
}
if (/viewer/i.test(query)) {
return {
method: 'viewer',
data: {
viewer: {
id: botUser.id,
displayName: botUser.displayName,
organization: { id: organizationId },
},
},
};
}
return { method: 'graphql', data: {} };
}
export async function createLinearReplayContext(
fixtures: LinearReplayFixtures,
options: { stream?: StreamChunk[] } = {},
): Promise<LinearReplayContext> {
const organizationId = fixtures.mention.organizationId;
const mode = fixtures.mention.type === 'AgentSessionEvent' ? 'agent-sessions' : 'comments';
const stubData = buildLinearStubData(fixtures, fixtures.botUser);
const stub = installFetchStub({
match: /api\.linear\.app/,
onRequest: ({ body }) => {
const query = typeof body.query === 'string' ? body.query : '';
const variables = (body.variables ?? {}) as { id?: unknown; input?: Record<string, unknown> };
const { method, data } = resolveLinearOperation(
query,
variables,
stubData,
fixtures.botUser,
organizationId,
);
return {
apiCall: { method, body: variables.input ?? (variables as Record<string, unknown>) },
responseBody: { data },
};
},
});
const { createLinearAdapter } = await import('@chat-adapter/linear');
const { Chat } = await import('chat');
const { createMemoryState } = await import('@chat-adapter/state-memory');
const adapter = createLinearAdapter({
accessToken: LINEAR_ACCESS_TOKEN,
webhookSecret: LINEAR_WEBHOOK_SECRET,
userName: fixtures.botUser.displayName,
mode,
} as Parameters<typeof createLinearAdapter>[0]);
const chat = new Chat({
userName: 'n8n-agent-agent-1',
adapters: { linear: adapter } as unknown as Record<string, never>,
state: createMemoryState(),
});
const integration: AgentIntegrationConfig = { type: 'linear', credentialId: 'cred-linear' };
const setup = createReplayContextSetup({
chat: chat as never,
integrationImpl: new LinearIntegration(mock<BackendLogger>(), mock<OutboundHttp>()),
integration,
componentMapper: new ComponentMapper(),
stream: options.stream,
});
await chat.initialize();
const webhooks = chat.webhooks as Record<string, ReplayWebhookHandler>;
const sendWebhook = async (payload: unknown) => {
// Linear's webhook client verifies an HMAC `linear-signature` and rejects
// stale `webhookTimestamp`s, so refresh the timestamp and sign the body.
const signed = { ...(payload as Record<string, unknown>), webhookTimestamp: Date.now() };
const rawBody = JSON.stringify(signed);
const headers = new Headers();
headers.set(
'linear-signature',
createHmac('sha256', LINEAR_WEBHOOK_SECRET).update(rawBody).digest('hex'),
);
return await sendJsonWebhook(
async (request, requestOptions) => await webhooks.linear(request, requestOptions),
'https://n8n.example.com/rest/projects/project-1/agents/v2/agent-1/webhooks/linear',
signed,
headers,
);
};
return {
...setup,
chat: chat as unknown as ChatInstance,
apiCalls: stub.apiCalls,
sendWebhook,
latestContext: () => setup.messageContextStore.latest(),
latestThreadId: () => setup.messageContextStore.latestThreadId(),
lastPost: () => stub.apiCalls.filter((call) => POST_METHODS.has(call.method)).at(-1),
shutdown: async () => {
try {
await setup.shutdown();
} finally {
stub.restore();
}
},
};
}
@@ -0,0 +1,228 @@
import type { StreamChunk } from '@n8n/agents';
import type { AgentIntegrationConfig } from '@n8n/api-types';
import { Container } from '@n8n/di';
import type { Logger } from 'n8n-workflow';
import type { Mock } from 'vitest';
import { mock } from 'vitest-mock-extended';
import { AgentChatBridge } from '../../agent-chat-bridge';
import { ChatIntegrationRegistry, type AgentChatIntegration } from '../../agent-chat-integration';
import type { ChatIntegrationService, ChatInstance } from '../../chat-integration.service';
import type { ComponentMapper } from '../../component-mapper';
import { ChatIntegrationActionExecutor } from '../../integration-action-executor';
import type { IntegrationMessageContextService } from '../../integration-message-context.service';
import type {
IntegrationMessageContext,
IntegrationMessageContextStore,
} from '../../integration-tools';
import { getIntegrationToolConnectionDescriptors } from '../../integration-tools';
type AgentExecutorLike = ConstructorParameters<typeof AgentChatBridge>[2];
export type ReplayWebhookOptions = { waitUntil?: (task: Promise<unknown>) => void };
export type ReplayWebhookHandler = (
request: Request,
options?: ReplayWebhookOptions,
) => Promise<Response>;
export interface ReplayApiCall {
method: string;
body: Record<string, unknown>;
}
export class MemoryMessageContextStore implements IntegrationMessageContextStore {
private readonly contexts = new Map<string, IntegrationMessageContext>();
async getLatest(threadId: string): Promise<IntegrationMessageContext | null> {
return await Promise.resolve(this.contexts.get(threadId) ?? null);
}
async setLatest(
threadId: string,
_resourceId: string,
context: IntegrationMessageContext,
): Promise<void> {
this.contexts.set(threadId, context);
await Promise.resolve();
return;
}
latest(): IntegrationMessageContext | undefined {
return [...this.contexts.values()].at(-1);
}
latestThreadId(): string | undefined {
return [...this.contexts.keys()].at(-1);
}
}
export function toStream(chunks: StreamChunk[]): AsyncGenerator<StreamChunk> {
return (async function* stream() {
await Promise.resolve();
for (const chunk of chunks) yield chunk;
})();
}
export async function sendJsonWebhook(
handler: (
request: Request,
options?: { waitUntil?: (task: Promise<unknown>) => void },
) => Promise<Response>,
url: string,
payload: unknown,
headers: Headers = new Headers(),
): Promise<Response> {
const tasks: Array<Promise<unknown>> = [];
headers.set('content-type', 'application/json');
const response = await handler(
new Request(url, {
method: 'POST',
headers,
body: JSON.stringify(payload),
}),
{ waitUntil: (task) => tasks.push(task) },
);
await Promise.all(tasks);
return response;
}
export interface StubbedRequest {
httpMethod: string;
url: string;
body: Record<string, unknown>;
rawBody: string | undefined;
}
export interface StubResponse {
/** Recorded call exposed to assertions (platform method name + request body). */
apiCall: ReplayApiCall;
/** JSON body returned to the real adapter so it proceeds without live network I/O. */
responseBody: unknown;
status?: number;
}
/**
* Network-boundary interceptor for the real platform adapters. Replaces
* `globalThis.fetch` for URLs matching `match`, records each matched request as a
* {@link ReplayApiCall}, and returns a canned JSON response. The adapter under test
* runs for real; only the external platform HTTP is answered here (from recorded
* fixtures or minimal stubs). Non-matching requests fall through. Call `restore()`
* on teardown. Adapters that use a non-fetch HTTP client (e.g. Slack's axios-based
* `@slack/web-api`) need `nock` instead.
*/
export function installFetchStub(options: {
match: RegExp;
onRequest: (request: StubbedRequest) => StubResponse;
}): { apiCalls: ReplayApiCall[]; restore: () => void } {
const apiCalls: ReplayApiCall[] = [];
const originalFetch = globalThis.fetch;
globalThis.fetch = async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
const url = typeof input === 'string' ? input : input instanceof URL ? input.href : input.url;
if (!options.match.test(url)) return await originalFetch(input, init);
const httpMethod = init?.method ?? (input instanceof Request ? input.method : 'GET');
const rawBody =
typeof init?.body === 'string'
? init.body
: input instanceof Request
? await input
.clone()
.text()
.catch(() => undefined)
: undefined;
let body: Record<string, unknown> = {};
if (rawBody) {
try {
const parsed: unknown = JSON.parse(rawBody);
if (parsed && typeof parsed === 'object') body = parsed as Record<string, unknown>;
} catch {
body = {};
}
}
const { apiCall, responseBody, status } = options.onRequest({ httpMethod, url, body, rawBody });
apiCalls.push(apiCall);
return new Response(JSON.stringify(responseBody), {
status: status ?? 200,
headers: { 'content-type': 'application/json' },
});
};
return {
apiCalls,
restore: () => {
globalThis.fetch = originalFetch;
},
};
}
export interface ReplayContextSetup<TChat extends ChatInstance = ChatInstance> {
chat: TChat;
agentExecutor: {
executeForChatPublished: Mock;
resumeForChat: Mock;
};
actionExecutor: ChatIntegrationActionExecutor;
descriptor: ReturnType<typeof getIntegrationToolConnectionDescriptors>[number];
integration: AgentIntegrationConfig;
messageContextStore: MemoryMessageContextStore;
nextStream: (chunks: StreamChunk[]) => void;
shutdown: () => Promise<void>;
}
export function createReplayContextSetup<TChat extends ChatInstance>(params: {
chat: TChat;
integrationImpl: AgentChatIntegration;
integration: AgentIntegrationConfig;
componentMapper?: ComponentMapper;
stream?: StreamChunk[];
}): ReplayContextSetup<TChat> {
const registry = new ChatIntegrationRegistry();
registry.register(params.integrationImpl);
Container.set(ChatIntegrationRegistry, registry);
let stream = params.stream ?? [
{ type: 'text-delta', id: 'text-1', delta: 'Got it' },
{ type: 'finish', finishReason: 'stop' },
];
const agentExecutor = {
executeForChatPublished: vi.fn(() => toStream(stream)),
resumeForChat: vi.fn(() => toStream(stream)),
};
const messageContextStore = new MemoryMessageContextStore();
new AgentChatBridge(
params.chat as never,
'agent-1',
agentExecutor as AgentExecutorLike,
params.componentMapper ?? mock<ComponentMapper>(),
mock<Logger>(),
'project-1',
params.integration,
messageContextStore as unknown as IntegrationMessageContextService,
);
const chatIntegrationService = mock<ChatIntegrationService>();
chatIntegrationService.getChatInstance.mockReturnValue(params.chat);
const actionExecutor = new ChatIntegrationActionExecutor(chatIntegrationService, registry);
const descriptor = getIntegrationToolConnectionDescriptors([params.integration], 'agent-1')[0];
return {
chat: params.chat,
agentExecutor,
actionExecutor,
descriptor,
integration: params.integration,
messageContextStore,
nextStream: (chunks: StreamChunk[]) => {
stream = chunks;
},
shutdown: async () => {
await params.chat.shutdown();
Container.reset();
},
};
}
@@ -0,0 +1,251 @@
import type { StreamChunk } from '@n8n/agents';
import type { AgentIntegrationConfig } from '@n8n/api-types';
import nock from 'nock';
import type { Mock } from 'vitest';
import type { ChatInstance } from '../../../chat-integration.service';
import { ComponentMapper } from '../../../component-mapper';
import type {
getIntegrationToolConnectionDescriptors,
IntegrationMessageContext,
} from '../../../integration-tools';
import { SlackIntegration } from '../../../platforms/slack-integration';
import {
createReplayContextSetup,
type MemoryMessageContextStore,
type ReplayApiCall,
type ReplayContextSetup,
type ReplayWebhookHandler,
sendJsonWebhook,
} from '../replay-test-helpers';
export interface SlackUserFixture {
id: string;
name: string;
real_name: string;
}
export interface SlackChannelFixture {
id: string;
name: string;
}
export interface SlackEventFixture {
token?: string;
type: 'event_callback';
team_id: string;
authorizations?: Array<{
team_id: string;
user_id: string;
is_bot: boolean;
}>;
event: {
type: string;
subtype?: string;
user: string;
text: string;
ts: string;
thread_ts?: string;
channel: string;
channel_type: string;
team?: string;
bot_id?: string;
};
}
export interface SlackReplayFixtures {
botUserId: string;
user: SlackUserFixture;
channel: SlackChannelFixture;
mention: SlackEventFixture;
followUp: SlackEventFixture;
selfMessage: SlackEventFixture;
}
export type SlackApiCall = ReplayApiCall;
export interface SlackReplayContext
extends Omit<ReplayContextSetup, 'agentExecutor' | 'nextStream' | 'chat'> {
chat: ChatInstance;
agentExecutor: {
executeForChatPublished: Mock;
};
apiCalls: SlackApiCall[];
descriptor: ReturnType<typeof getIntegrationToolConnectionDescriptors>[number];
integration: AgentIntegrationConfig;
messageContextStore: MemoryMessageContextStore;
sendWebhook: (payload: unknown) => Promise<Response>;
latestContext: () => IntegrationMessageContext | undefined;
latestThreadId: () => string | undefined;
lastPost: () => SlackApiCall | undefined;
}
const SLACK_BOT_TOKEN = 'xoxb-test-token';
const SLACK_API_URL = 'https://slack.com/api/';
/** Extract the Web API method (`chat.postMessage`, `auth.test`, …) from a URL. */
function slackMethodFromUri(uri: string): string {
return uri.split('?')[0].replace(/^.*\/api\//, '');
}
/** `@slack/web-api` posts form-encoded bodies; normalize them to a record. */
function parseSlackBody(requestBody: unknown): Record<string, unknown> {
if (requestBody && typeof requestBody === 'object') return requestBody as Record<string, unknown>;
if (typeof requestBody !== 'string') return {};
const params = new URLSearchParams(requestBody);
const body: Record<string, unknown> = {};
for (const [key, value] of params) body[key] = value;
return body;
}
/** Concatenate `markdown_text` from a Slack streaming `chunks` form field. */
function collectStreamChunkText(body: Record<string, unknown>): string {
const raw = body.chunks;
if (typeof raw !== 'string') return '';
try {
const chunks = JSON.parse(raw) as Array<{ type?: string; text?: string }>;
return chunks
.filter((chunk) => chunk.type === 'markdown_text' && typeof chunk.text === 'string')
.map((chunk) => chunk.text)
.join('');
} catch {
return '';
}
}
/**
* The Slack adapter calls the Web API through `@slack/web-api` (axios), so a
* `fetch` stub can't see it — intercept at the HTTP layer with nock instead.
*
* Agent replies are sent through Slack's assistant streaming API
* (`chat.startStream` → `appendStream` → `stopStream`), not `chat.postMessage`.
* We reconstruct the streamed text and record it as a synthetic `chat.postMessage`
* so assertions can treat the agent reply as a single outbound post.
*/
function installSlackApiNock(botUserId: string) {
const apiCalls: SlackApiCall[] = [];
let streamSeq = 0;
let activeStream: { channel: string; threadTs: string; ts: string; text: string } | undefined;
nock('https://slack.com')
.persist()
.post(/\/api\/.+/)
.reply(200, (uri, requestBody) => {
const method = slackMethodFromUri(uri);
const body = parseSlackBody(requestBody);
if (method === 'chat.startStream') {
const ts = `1719000999.0000${++streamSeq}`;
activeStream = {
channel: String(body.channel ?? ''),
threadTs: String(body.thread_ts ?? ''),
ts,
text: collectStreamChunkText(body),
};
return { ok: true, ts };
}
if (method === 'chat.appendStream') {
if (activeStream) activeStream.text += collectStreamChunkText(body);
return { ok: true, ts: activeStream?.ts };
}
if (method === 'chat.stopStream') {
if (activeStream) {
activeStream.text += collectStreamChunkText(body);
// Surface the streamed reply as a single post for assertions.
apiCalls.push({
method: 'chat.postMessage',
body: {
channel: activeStream.channel,
thread_ts: activeStream.threadTs,
// Streamed agent replies carry text as markdown, matching the
// real adapter's non-streaming `chat.postMessage` for DMs.
markdown_text: activeStream.text,
},
});
}
const ts = activeStream?.ts ?? '1719000999.000999';
activeStream = undefined;
return { ok: true, ts, message: { ts } };
}
apiCalls.push({ method, body });
if (method === 'auth.test') {
return { ok: true, user_id: botUserId, user: 'n8n_agent', team_id: 'T_TEAM' };
}
if (method === 'chat.postMessage') {
return {
ok: true,
channel: body.channel,
ts: '1719000999.000999',
message: { ts: '1719000999.000999' },
};
}
if (method === 'conversations.open') {
return { ok: true, channel: { id: 'D_OPENED' } };
}
return { ok: true };
});
return { apiCalls, restore: () => nock.cleanAll() };
}
export async function createSlackReplayContext(
fixtures: SlackReplayFixtures,
options: { stream?: StreamChunk[] } = {},
): Promise<SlackReplayContext> {
const stub = installSlackApiNock(fixtures.botUserId);
const { createSlackAdapter } = await import('@chat-adapter/slack');
const { Chat } = await import('chat');
const { createMemoryState } = await import('@chat-adapter/state-memory');
const adapter = createSlackAdapter({
botToken: SLACK_BOT_TOKEN,
// Provided so the adapter skips the `auth.test` identity lookup on connect.
botUserId: fixtures.botUserId,
// Replay fixtures carry sanitized signatures, so bypass signature checks.
webhookVerifier: () => true,
mode: 'webhook',
apiUrl: SLACK_API_URL,
});
const chat = new Chat({
userName: 'n8n-agent-agent-1',
adapters: { slack: adapter } as unknown as Record<string, never>,
state: createMemoryState(),
});
const integration: AgentIntegrationConfig = { type: 'slack', credentialId: 'cred-slack' };
const setup = createReplayContextSetup({
chat: chat as never,
integrationImpl: new SlackIntegration(),
integration,
componentMapper: new ComponentMapper(),
stream: options.stream,
});
await chat.initialize();
const webhooks = chat.webhooks as Record<string, ReplayWebhookHandler>;
const sendWebhook = async (payload: unknown) =>
await sendJsonWebhook(
async (request, requestOptions) => await webhooks.slack(request, requestOptions),
'https://n8n.example.com/rest/projects/project-1/agents/v2/agent-1/webhooks/slack',
payload,
);
return {
...setup,
chat: chat as unknown as ChatInstance,
apiCalls: stub.apiCalls,
sendWebhook,
latestContext: () => setup.messageContextStore.latest(),
latestThreadId: () => setup.messageContextStore.latestThreadId(),
lastPost: () => stub.apiCalls.filter((call) => call.method === 'chat.postMessage').at(-1),
shutdown: async () => {
try {
await setup.shutdown();
} finally {
stub.restore();
}
},
};
}
@@ -0,0 +1,98 @@
import type {
SlackChannelFixture,
SlackEventFixture,
SlackReplayFixtures,
SlackUserFixture,
} from './replay-test-context';
export const slackUser = (overrides: Partial<SlackUserFixture> = {}): SlackUserFixture => ({
id: 'U_ALICE',
name: 'alice',
real_name: 'Alice Developer',
...overrides,
});
export const slackChannel = (
overrides: Partial<SlackChannelFixture> = {},
): SlackChannelFixture => ({
id: 'C_SUPPORT',
name: 'support',
...overrides,
});
export const slackEvent = (
overrides: Partial<SlackEventFixture['event']> = {},
): SlackEventFixture['event'] => ({
type: 'app_mention',
user: 'U_ALICE',
text: '<@U_BOT> hello agent',
ts: '1719000000.000100',
thread_ts: '1719000000.000100',
channel: 'C_SUPPORT',
channel_type: 'channel',
team: 'T_TEAM',
...overrides,
});
export const slackEventCallback = (
overrides: Omit<Partial<SlackEventFixture>, 'event'> & {
event?: Partial<SlackEventFixture['event']>;
} = {},
): SlackEventFixture => ({
token: 'SLACK_VERIFICATION_TOKEN',
type: 'event_callback',
team_id: 'T_TEAM',
authorizations: [{ team_id: 'T_TEAM', user_id: 'U_BOT', is_bot: true }],
event: slackEvent(overrides.event),
});
export const slackReplayFixtures = (
overrides: Partial<SlackReplayFixtures> = {},
): SlackReplayFixtures => {
const botUserId = overrides.botUserId ?? 'U_BOT';
const user = overrides.user ?? slackUser();
const channel = overrides.channel ?? slackChannel();
const mention =
overrides.mention ??
slackEventCallback({
event: {
type: 'app_mention',
user: user.id,
text: `<@${botUserId}> hello agent`,
channel: channel.id,
},
});
return {
botUserId,
user,
channel,
mention,
followUp:
overrides.followUp ??
slackEventCallback({
event: {
type: 'message',
user: user.id,
text: 'follow up',
ts: '1719000001.000200',
thread_ts: mention.event.thread_ts ?? mention.event.ts,
channel: channel.id,
},
}),
selfMessage:
overrides.selfMessage ??
slackEventCallback({
event: {
type: 'message',
user: botUserId,
bot_id: 'B_BOT',
text: 'bot echo',
ts: '1719000002.000300',
thread_ts: mention.event.thread_ts ?? mention.event.ts,
channel: channel.id,
},
}),
...overrides,
};
};
@@ -0,0 +1,270 @@
import type { StreamChunk } from '@n8n/agents';
import type { AgentIntegrationConfig } from '@n8n/api-types';
import type { Logger as BackendLogger } from '@n8n/backend-common';
import type { OutboundHttp, SsrfProtectionService } from '@n8n/backend-network';
import type { SsrfProtectionConfig } from '@n8n/config';
import type { InstanceSettings } from 'n8n-core';
import type { Mock } from 'vitest';
import { mock } from 'vitest-mock-extended';
import type { UrlService } from '@/services/url.service';
import type { AgentRepository } from '../../../../repositories/agent.repository';
import type { ChatInstance } from '../../../chat-integration.service';
import { ComponentMapper } from '../../../component-mapper';
import type { ChatIntegrationActionExecutor } from '../../../integration-action-executor';
import type {
getIntegrationToolConnectionDescriptors,
IntegrationMessageContext,
} from '../../../integration-tools';
import { TelegramIntegration } from '../../../platforms/telegram-integration';
import {
createReplayContextSetup,
installFetchStub,
type MemoryMessageContextStore,
type ReplayApiCall,
type ReplayContextSetup,
type ReplayWebhookHandler,
sendJsonWebhook,
} from '../replay-test-helpers';
export interface TelegramUserFixture {
id: number;
first_name: string;
is_bot: boolean;
username?: string;
last_name?: string;
language_code?: string;
}
export interface TelegramChatFixture {
id: number;
type: 'private' | 'group' | 'supergroup' | 'channel';
first_name?: string;
last_name?: string;
username?: string;
title?: string;
}
export interface TelegramMessageFixture {
message_id: number;
from?: TelegramUserFixture;
chat: TelegramChatFixture;
date: number;
text?: string;
message_thread_id?: number;
}
export interface TelegramCallbackQueryFixture {
id: string;
from: TelegramUserFixture;
message?: TelegramMessageFixture;
data?: string;
chat_instance?: string;
}
export interface TelegramUpdateFixture {
update_id: number;
message?: TelegramMessageFixture;
callback_query?: TelegramCallbackQueryFixture;
}
export type TelegramApiCall = ReplayApiCall;
export interface TelegramReplayFixtures {
mention: TelegramUpdateFixture;
followUp: TelegramUpdateFixture;
selfMessage: TelegramUpdateFixture;
callbackBase: TelegramUpdateFixture;
user: TelegramUserFixture;
bot: TelegramUserFixture;
chat: TelegramChatFixture;
}
export interface TelegramReplayContext extends Omit<ReplayContextSetup, 'nextStream' | 'chat'> {
chat: ChatInstance;
agentExecutor: {
executeForChatPublished: Mock;
resumeForChat: Mock;
};
actionExecutor: ChatIntegrationActionExecutor;
apiCalls: TelegramApiCall[];
descriptor: ReturnType<typeof getIntegrationToolConnectionDescriptors>[number];
integration: AgentIntegrationConfig;
messageContextStore: MemoryMessageContextStore;
sendTelegramWebhook: (payload: unknown) => Promise<Response>;
sendWebhook: (payload: unknown) => Promise<Response>;
latestContext: () => IntegrationMessageContext | undefined;
latestThreadId: () => string | undefined;
lastApiCall: (method: string) => TelegramApiCall | undefined;
lastPost: () => TelegramApiCall | undefined;
nextStream: (chunks: StreamChunk[]) => void;
}
// Sanitized bot token from the recorded sessions — using it keeps the real
// adapter's outbound URLs identical to the captured fixtures.
const TELEGRAM_BOT_TOKEN = '123456789:abcdefghijkl';
const TELEGRAM_SECRET_TOKEN = 'test-secret-token';
/** Extract the Bot API method (`sendMessage`, `getMe`, …) from a request URL. */
function telegramMethodFromUrl(url: string): string {
const path = url.split('?')[0];
return path.slice(path.lastIndexOf('/') + 1);
}
/**
* Answer the Telegram Bot API for the real `@chat-adapter/telegram` adapter:
* `getMe` returns the bot fixture (so the adapter learns its identity), and
* `sendMessage` returns a minimal message; every call is recorded for assertions.
*/
function installTelegramApiStub(bot: TelegramUserFixture) {
let nextMessageId = 1000;
return installFetchStub({
match: /api\.telegram\.org/,
onRequest: ({ url, body }) => {
const method = telegramMethodFromUrl(url);
let result: unknown = true;
if (method === 'getMe') {
result = bot;
} else if (method === 'sendMessage') {
result = {
message_id: nextMessageId++,
chat: { id: Number(body.chat_id) },
date: 1719000000,
text: body.text ?? '',
};
}
return { apiCall: { method, body }, responseBody: { ok: true, result } };
},
});
}
function createIntegration() {
const urlService = mock<UrlService>();
urlService.getWebhookBaseUrl.mockReturnValue('https://n8n.example.com/');
return new TelegramIntegration(
mock<BackendLogger>(),
urlService,
mock<AgentRepository>(),
mock<InstanceSettings>({ encryptionKey: 'test-encryption-key' }),
mock<OutboundHttp>(),
{ enabled: false } as SsrfProtectionConfig,
mock<SsrfProtectionService>(),
);
}
export function callbackPayloadWithData(
payload: TelegramUpdateFixture,
data: string,
messageId: number,
): TelegramUpdateFixture {
return {
...payload,
callback_query: payload.callback_query
? {
...payload.callback_query,
data,
message: payload.callback_query.message
? { ...payload.callback_query.message, message_id: messageId }
: undefined,
}
: undefined,
};
}
export async function createTelegramReplayContext(
fixtures: TelegramReplayFixtures,
options: {
stream?: StreamChunk[];
integration?: AgentIntegrationConfig;
} = {},
): Promise<TelegramReplayContext> {
const stub = installTelegramApiStub(fixtures.bot);
// Dynamic imports — the chat packages are ESM-only. Unlike production (which
// must route through esm-loader to dodge the CJS transform), vitest loads ESM
// natively, so the tests use the real adapters directly.
const { createTelegramAdapter } = await import('@chat-adapter/telegram');
const { Chat } = await import('chat');
const { createMemoryState } = await import('@chat-adapter/state-memory');
const adapter = createTelegramAdapter({
botToken: TELEGRAM_BOT_TOKEN,
mode: 'webhook',
secretToken: TELEGRAM_SECRET_TOKEN,
apiBaseUrl: 'https://api.telegram.org',
});
const chat = new Chat({
userName: 'n8n-agent-agent-1',
adapters: { telegram: adapter } as unknown as Record<string, never>,
state: createMemoryState(),
});
const integration = options.integration ?? {
type: 'telegram',
credentialId: 'cred-telegram',
settings: { accessMode: 'public', allowedUsers: [] },
};
const setup = createReplayContextSetup({
chat: chat as never,
integrationImpl: createIntegration(),
integration,
componentMapper: new ComponentMapper(),
stream: options.stream,
});
// Connects the adapter (calls `getMe`, served by the stub) and wires webhooks.
await chat.initialize();
const webhooks = chat.webhooks as Record<string, ReplayWebhookHandler>;
const sendTelegramWebhook = async (payload: unknown) => {
const headers = new Headers();
headers.set('x-telegram-bot-api-secret-token', TELEGRAM_SECRET_TOKEN);
return await sendJsonWebhook(
async (request, requestOptions) => await webhooks.telegram(request, requestOptions),
'https://n8n.example.com/rest/projects/project-1/agents/v2/agent-1/webhooks/telegram',
payload,
headers,
);
};
return {
...setup,
chat: chat as unknown as ChatInstance,
apiCalls: stub.apiCalls,
sendTelegramWebhook,
sendWebhook: sendTelegramWebhook,
latestContext: () => setup.messageContextStore.latest(),
latestThreadId: () => setup.messageContextStore.latestThreadId(),
lastApiCall: (method: string) => stub.apiCalls.filter((call) => call.method === method).at(-1),
lastPost: () => stub.apiCalls.filter((call) => call.method === 'sendMessage').at(-1),
shutdown: async () => {
try {
await setup.shutdown();
} finally {
stub.restore();
}
},
};
}
export function getTelegramInlineCallbackData(
call: TelegramApiCall | undefined,
): string | undefined {
// The real adapter sends `reply_markup` as a JSON object in the request body
// (Telegram accepts it inline in JSON mode); tolerate a stringified form too.
const replyMarkup = call?.body.reply_markup;
let markup: unknown = replyMarkup;
if (typeof replyMarkup === 'string') {
try {
markup = JSON.parse(replyMarkup);
} catch {
return undefined;
}
}
if (!markup || typeof markup !== 'object') return undefined;
const inlineKeyboard = (markup as { inline_keyboard?: Array<Array<{ callback_data?: string }>> })
.inline_keyboard;
return inlineKeyboard?.[0]?.[0]?.callback_data;
}
@@ -0,0 +1,137 @@
import type {
TelegramChatFixture,
TelegramMessageFixture,
TelegramReplayFixtures,
TelegramUpdateFixture,
TelegramUserFixture,
} from './replay-test-context';
const DEFAULT_DATE = 1_719_000_000;
export const telegramBot = (overrides: Partial<TelegramUserFixture> = {}): TelegramUserFixture => ({
id: 777000,
is_bot: true,
first_name: 'n8n Agent',
username: 'n8n_agent_bot',
...overrides,
});
export const telegramUser = (
overrides: Partial<TelegramUserFixture> = {},
): TelegramUserFixture => ({
id: 123456,
is_bot: false,
first_name: 'Alice',
username: 'alice_dev',
...overrides,
});
export const telegramPrivateChat = (
overrides: Partial<TelegramChatFixture> = {},
): TelegramChatFixture => ({
id: 123456,
type: 'private',
first_name: 'Alice',
username: 'alice_dev',
...overrides,
});
export const telegramGroupChat = (
overrides: Partial<TelegramChatFixture> = {},
): TelegramChatFixture => ({
id: -1001234567890,
type: 'supergroup',
title: 'Agent Test Group',
...overrides,
});
export const telegramMessage = (
overrides: Partial<TelegramMessageFixture> = {},
): TelegramMessageFixture => ({
message_id: 11,
from: telegramUser(),
chat: telegramPrivateChat(),
date: DEFAULT_DATE,
text: 'hello agent',
...overrides,
});
export const telegramMessageUpdate = (
overrides: Omit<Partial<TelegramUpdateFixture>, 'message'> & {
message?: Partial<TelegramMessageFixture>;
} = {},
): TelegramUpdateFixture => ({
update_id: overrides.update_id ?? 10001,
message: telegramMessage(overrides.message),
});
export const telegramCallbackQueryUpdate = (options: {
data: string;
message?: Partial<TelegramMessageFixture>;
from?: Partial<TelegramUserFixture>;
updateId?: number;
callbackId?: string;
}): TelegramUpdateFixture => ({
update_id: options.updateId ?? 10004,
callback_query: {
id: options.callbackId ?? 'callback-1',
from: telegramUser(options.from),
message: telegramMessage({
message_id: 1000,
from: telegramBot(),
text: 'Approval required',
...options.message,
}),
chat_instance: 'chat-instance-1',
data: options.data,
},
});
export const telegramReplayFixtures = (
overrides: Partial<TelegramReplayFixtures> = {},
): TelegramReplayFixtures => {
const bot = overrides.bot ?? telegramBot();
const user = overrides.user ?? telegramUser();
const chat = overrides.chat ?? telegramPrivateChat();
const mention =
overrides.mention ??
telegramMessageUpdate({
message: { from: user, chat },
});
const mentionMessage = mention.message ?? telegramMessage({ from: user, chat });
return {
bot,
user,
chat,
mention,
followUp:
overrides.followUp ??
telegramMessageUpdate({
update_id: mention.update_id + 1,
message: {
...mentionMessage,
message_id: mentionMessage.message_id + 1,
text: 'follow up',
},
}),
selfMessage:
overrides.selfMessage ??
telegramMessageUpdate({
update_id: mention.update_id + 2,
message: {
...mentionMessage,
message_id: mentionMessage.message_id + 2,
from: bot,
text: 'bot echo',
},
}),
callbackBase:
overrides.callbackBase ??
telegramCallbackQueryUpdate({
data: 'placeholder',
message: { from: bot, chat },
from: user,
}),
};
};
@@ -0,0 +1,64 @@
import type { Mock } from 'vitest';
import type { ChannelIntegrationRecorder } from '../recording/channel-integration-recorder';
import { recordAdapterCalls } from '../recording/recording-adapter';
describe('recordAdapterCalls', () => {
function enabledRecorder(recordApiCall: Mock): ChannelIntegrationRecorder {
return {
isEnabled: true,
recordApiCall,
} as unknown as ChannelIntegrationRecorder;
}
it('does not fail the adapter call when recording fails', async () => {
const adapter = {
postMessage: vi.fn(async () => ({ id: 'message-1' })),
};
const recorder = enabledRecorder(
vi.fn(async () => {
throw new Error('disk full');
}),
);
const wrapped = recordAdapterCalls('slack', adapter, recorder) as typeof adapter;
await expect(wrapped.postMessage()).resolves.toEqual({ id: 'message-1' });
});
it('records only stream chunks consumed by the adapter', async () => {
let pulledChunks = 0;
async function* textStream() {
pulledChunks++;
yield 'first';
pulledChunks++;
yield 'second';
}
const adapter = {
stream: vi.fn(async (_threadId: string, stream: AsyncIterable<string>) => {
const iterator = stream[Symbol.asyncIterator]();
return await iterator.next();
}),
};
const recordApiCall = vi.fn(async () => undefined);
const wrapped = recordAdapterCalls(
'slack',
adapter,
enabledRecorder(recordApiCall),
) as typeof adapter;
await expect(wrapped.stream('thread-1', textStream())).resolves.toMatchObject({
value: 'first',
done: false,
});
expect(pulledChunks).toBe(1);
expect(recordApiCall).toHaveBeenCalledWith(
'slack',
'stream',
['thread-1', { streamChunks: ['first'] }],
expect.objectContaining({ value: 'first', done: false }),
undefined,
);
});
});
@@ -27,6 +27,8 @@ import { AgentChatSubscriptionStateService } from './agent-chat-subscription-sta
import { ComponentMapper, type ShortenCallback } from './component-mapper';
import { loadChatSdk, loadMemoryState } from './esm-loader';
import { buildIntegrationConnectionId } from './integration-tools';
import { channelIntegrationRecorder } from './recording/channel-integration-recorder';
import { recordAdapterCalls } from './recording/recording-adapter';
import type { Agent } from '../entities/agent.entity';
import { AgentRepository } from '../repositories/agent.repository';
@@ -201,7 +203,8 @@ export class ChatIntegrationService {
}
// Delegate adapter construction to the platform implementation.
const adapter = await integrationImpl.createAdapter(ctx);
const adapter = recordAdapterCalls(integration.type, await integrationImpl.createAdapter(ctx));
channelIntegrationRecorder.startFetchRecording();
// Dynamic imports — chat packages are ESM-only, use loader to bypass CJS transform
const { Chat } = await loadChatSdk();
@@ -0,0 +1,138 @@
import { readFileSync } from 'fs';
import { jsonParse } from 'n8n-workflow';
import { join } from 'path';
import {
createLinearReplayContext,
type LinearAgentSessionEventFixture,
type LinearReplayFixtures,
} from '../../../__tests__/helpers/linear/replay-test-context';
import type { ChannelIntegrationRecord } from '../../../recording/channel-integration-recorder';
const recordedSession = jsonParse<ChannelIntegrationRecord[]>(
readFileSync(join(__dirname, '../../../__tests__/fixtures/linear/recorded-session.json'), 'utf8'),
);
function getRecordedWebhook() {
const record = recordedSession.find(
(entry): entry is Extract<ChannelIntegrationRecord, { type: 'webhook' }> =>
entry.type === 'webhook' && entry.platform === 'linear',
);
if (!record) throw new Error('Expected Linear webhook record');
return record;
}
function getRecordedAgentActivityCreate() {
const record = recordedSession.find(
(entry): entry is Extract<ChannelIntegrationRecord, { type: 'fetch' }> => {
if (entry.type !== 'fetch' || !entry.requestBody) return false;
const request = jsonParse<{ variables?: { input?: { agentSessionId?: string } } }>(
entry.requestBody,
);
return request.variables?.input?.agentSessionId === 'AGENT_SESSION_1';
},
);
if (!record?.requestBody) throw new Error('Expected Linear agentActivityCreate fetch record');
return record;
}
function recordedLinearFixtures(): LinearReplayFixtures {
const mention = jsonParse<LinearAgentSessionEventFixture>(getRecordedWebhook().body);
return {
botUser: {
id: mention.appUserId,
name: 'testapp',
displayName: 'testapp',
app: true,
},
mention,
};
}
function recordedAgentActivityInput() {
return jsonParse<{
variables: {
input: {
agentSessionId: string;
content: { type: string; body: string };
};
};
}>(getRecordedAgentActivityCreate().requestBody ?? '{}').variables.input;
}
describe('Linear recorded integration replay', () => {
it('replays the captured Linear agent session webhook and outbound activity', async () => {
const fixtures = recordedLinearFixtures();
const recordedActivityInput = recordedAgentActivityInput();
const ctx = await createLinearReplayContext(fixtures, {
stream: [
{
type: 'text-delta',
id: 'recorded-linear-response',
delta: recordedActivityInput.content.body,
},
{ type: 'finish', finishReason: 'stop' },
],
});
try {
await ctx.sendWebhook(fixtures.mention);
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenCalledWith(
expect.objectContaining({
message: '@testapp hey',
integrationType: 'linear',
}),
);
expect(ctx.latestContext()).toMatchObject({
platform: 'linear',
messageId: 'COMMENT_SOURCE',
interactingUserId: 'USER_ALICE',
target: {
threadId: 'linear:ISSUE_1:c:COMMENT_SOURCE:s:AGENT_SESSION_1',
channelId: 'linear:ISSUE_1',
},
});
expect(ctx.lastPost()?.body).toEqual(recordedActivityInput);
} finally {
await ctx.shutdown();
}
});
it('responds to the current Linear agent session through the action executor', async () => {
const fixtures = recordedLinearFixtures();
const ctx = await createLinearReplayContext(fixtures);
try {
await ctx.sendWebhook(fixtures.mention);
const context = ctx.latestContext();
expect(context).toMatchObject({
platform: 'linear',
target: { threadId: 'linear:ISSUE_1:c:COMMENT_SOURCE:s:AGENT_SESSION_1' },
});
const result = await ctx.actionExecutor.execute({
descriptor: ctx.descriptor,
action: 'respond',
input: { message: { text: 'Action response' } },
awaitResponse: false,
currentMessageContext: context,
});
expect(result).toMatchObject({
ok: true,
messageContext: {
platform: 'linear',
target: { type: 'thread', threadId: 'linear:ISSUE_1:c:COMMENT_SOURCE:s:AGENT_SESSION_1' },
},
});
expect(ctx.lastPost()?.body).toMatchObject({
agentSessionId: 'AGENT_SESSION_1',
content: { type: 'response', body: 'Action response' },
});
} finally {
await ctx.shutdown();
}
});
});
@@ -0,0 +1,163 @@
import { readFileSync } from 'fs';
import { jsonParse } from 'n8n-workflow';
import { join } from 'path';
import {
createSlackReplayContext,
type SlackReplayFixtures,
} from '../../../__tests__/helpers/slack/replay-test-context';
import type { ChannelIntegrationRecord } from '../../../recording/channel-integration-recorder';
const recordedSession = jsonParse<ChannelIntegrationRecord[]>(
readFileSync(join(__dirname, '../../../__tests__/fixtures/slack/recorded-session.json'), 'utf8'),
);
function recordedWebhook(eventId: string) {
const record = recordedSession.find(
(entry): entry is Extract<ChannelIntegrationRecord, { type: 'webhook' }> => {
if (entry.type !== 'webhook' || entry.platform !== 'slack') return false;
const body = jsonParse<{ event_id?: string }>(entry.body);
return body.event_id === eventId;
},
);
if (!record) throw new Error(`Expected Slack webhook record ${eventId}`);
return record;
}
function webhookBody(eventId: string) {
return jsonParse<SlackReplayFixtures['mention']>(recordedWebhook(eventId).body);
}
function recordedSlackFixtures(overrides: Partial<SlackReplayFixtures> = {}): SlackReplayFixtures {
const mention = overrides.mention ?? webhookBody('Ev_APP_MENTION');
const followUp =
overrides.followUp ??
({
...mention,
event: {
...mention.event,
type: 'message',
text: 'follow up',
ts: '1782378391.000000',
thread_ts: mention.event.ts,
},
} as SlackReplayFixtures['followUp']);
return {
botUserId: 'U_BOT',
user: { id: 'U_USER', name: 'user', real_name: 'Recorded User' },
channel: { id: 'C_CHANNEL', name: 'recorded-channel' },
mention,
followUp,
selfMessage: overrides.selfMessage ?? webhookBody('Ev_BOT_CHANNEL_RESPONSE'),
...overrides,
};
}
describe('Slack recorded integration replay', () => {
it('replays a recorded Slack app mention and outbound post', async () => {
const fixtures = recordedSlackFixtures();
const ctx = await createSlackReplayContext(fixtures, {
stream: [
{
type: 'text-delta',
id: 'recorded-slack-response',
delta: 'Hey! 👋 How can I help you today?',
},
{ type: 'finish', finishReason: 'stop' },
],
});
try {
await ctx.sendWebhook(fixtures.mention);
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenCalledWith(
expect.objectContaining({
message: 'hey',
integrationType: 'slack',
}),
);
expect(ctx.latestContext()).toMatchObject({
platform: 'slack',
messageId: '1782378390.841549',
interactingUserId: 'U_USER',
agentUserId: 'U_BOT',
target: {
threadId: 'slack:C_CHANNEL:1782378390.841549',
channelId: 'slack:C_CHANNEL',
},
});
expect(ctx.lastPost()?.body).toMatchObject({
channel: 'C_CHANNEL',
thread_ts: '1782378390.841549',
markdown_text: 'Hey! 👋 How can I help you today?',
});
} finally {
await ctx.shutdown();
}
});
it('ignores duplicate recorded Slack bot-authored channel_join webhooks', async () => {
const fixtures = recordedSlackFixtures({ mention: webhookBody('Ev_CHANNEL_JOIN') });
const retry = recordedWebhook('Ev_CHANNEL_JOIN');
const ctx = await createSlackReplayContext(fixtures);
try {
await ctx.sendWebhook(fixtures.mention);
await ctx.sendWebhook(jsonParse<SlackReplayFixtures['mention']>(retry.body));
expect(ctx.agentExecutor.executeForChatPublished).not.toHaveBeenCalled();
} finally {
await ctx.shutdown();
}
});
it('replays a recorded Slack DM message and outbound DM post', async () => {
const fixtures = recordedSlackFixtures({
mention: webhookBody('Ev_DM_MESSAGE'),
selfMessage: webhookBody('Ev_BOT_DM_RESPONSE'),
});
const ctx = await createSlackReplayContext(fixtures, {
stream: [
{ type: 'text-delta', id: 'dm-response', delta: "I'm Assistant." },
{ type: 'finish', finishReason: 'stop' },
],
});
try {
await ctx.sendWebhook(fixtures.mention);
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenCalledWith(
expect.objectContaining({
message: "DM message. What's your name?",
integrationType: 'slack',
}),
);
expect(ctx.latestContext()).toMatchObject({
platform: 'slack',
messageId: '1782379185.654229',
interactingUserId: 'U_USER',
target: {
threadId: 'slack:D_DM:',
channelId: 'slack:D_DM',
},
});
expect(ctx.lastPost()?.body).toMatchObject({
channel: 'D_DM',
markdown_text: "I'm Assistant.",
});
} finally {
await ctx.shutdown();
}
});
it('ignores recorded Slack bot-authored response webhooks', async () => {
const ctx = await createSlackReplayContext(recordedSlackFixtures());
try {
await ctx.sendWebhook(webhookBody('Ev_BOT_CHANNEL_RESPONSE'));
await ctx.sendWebhook(webhookBody('Ev_BOT_DM_RESPONSE'));
expect(ctx.agentExecutor.executeForChatPublished).not.toHaveBeenCalled();
} finally {
await ctx.shutdown();
}
});
});
@@ -0,0 +1,115 @@
import { createSlackReplayContext } from '../../../__tests__/helpers/slack/replay-test-context';
import {
slackEventCallback,
slackReplayFixtures,
slackUser,
} from '../../../__tests__/helpers/slack/synthetic-fixtures';
import { SlackIntegration } from '../../slack-integration';
describe('Slack channel integration scenarios', () => {
it('handles Slack URL verification without an active connection', () => {
const integration = new SlackIntegration();
expect(
integration.handleUnauthenticatedWebhook({
type: 'url_verification',
challenge: 'url-verification-challenge',
}),
).toEqual({ status: 200, body: { challenge: 'url-verification-challenge' } });
});
it('routes a Slack DM message as a new agent conversation', async () => {
const user = slackUser({ id: 'U_DM_USER', name: 'dm_user', real_name: 'DM User' });
const fixtures = slackReplayFixtures({
user,
mention: slackEventCallback({
event: {
type: 'message',
user: user.id,
text: "DM message. What's your name?",
ts: '1719000100.000100',
channel: 'D_DM',
channel_type: 'im',
thread_ts: undefined,
},
}),
});
const ctx = await createSlackReplayContext(fixtures);
try {
await ctx.sendWebhook(fixtures.mention);
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenCalledWith(
expect.objectContaining({
message: "DM message. What's your name?",
integrationType: 'slack',
}),
);
expect(ctx.latestContext()).toMatchObject({
messageId: '1719000100.000100',
interactingUserId: 'U_DM_USER',
target: {
threadId: 'slack:D_DM:',
channelId: 'slack:D_DM',
},
});
expect(ctx.lastPost()?.body).toMatchObject({
channel: 'D_DM',
markdown_text: 'Got it',
});
} finally {
await ctx.shutdown();
}
});
it('ignores a Slack bot-authored channel_join message', async () => {
const fixtures = slackReplayFixtures({
mention: slackEventCallback({
event: {
type: 'message',
subtype: 'channel_join',
user: 'U_BOT',
bot_id: 'B_BOT',
text: '<@U_BOT> has joined the channel',
ts: '1719000200.000100',
channel: 'C_SUPPORT',
channel_type: 'channel',
},
}),
});
const ctx = await createSlackReplayContext(fixtures);
try {
await ctx.sendWebhook(fixtures.mention);
expect(ctx.agentExecutor.executeForChatPublished).not.toHaveBeenCalled();
expect(ctx.latestContext()).toBeUndefined();
} finally {
await ctx.shutdown();
}
});
it('ignores a Slack bot-authored message in a thread', async () => {
const fixtures = slackReplayFixtures({
mention: slackEventCallback({
event: {
type: 'message',
user: 'U_BOT',
bot_id: 'B_BOT',
text: 'bot echo',
ts: '1719000300.000100',
thread_ts: '1719000000.000100',
channel: 'C_SUPPORT',
channel_type: 'channel',
},
}),
});
const ctx = await createSlackReplayContext(fixtures);
try {
await ctx.sendWebhook(fixtures.mention);
expect(ctx.agentExecutor.executeForChatPublished).not.toHaveBeenCalled();
expect(ctx.latestContext()).toBeUndefined();
} finally {
await ctx.shutdown();
}
});
});
@@ -11,13 +11,13 @@ import type { InstanceSettings } from 'n8n-core';
import { ConflictError } from '@/errors/response-errors/conflict.error';
import type { UrlService } from '@/services/url.service';
import type { Agent } from '../../../entities/agent.entity';
import type { AgentRepository } from '../../../repositories/agent.repository';
import type { AgentChatIntegrationContext } from '../../agent-chat-integration';
import { loadTelegramAdapter } from '../../esm-loader';
import { TelegramIntegration } from '../telegram-integration';
import type { Agent } from '../../../../entities/agent.entity';
import type { AgentRepository } from '../../../../repositories/agent.repository';
import type { AgentChatIntegrationContext } from '../../../agent-chat-integration';
import { loadTelegramAdapter } from '../../../esm-loader';
import { TelegramIntegration } from '../../telegram-integration';
vi.mock('../../esm-loader', () => ({
vi.mock('../../../esm-loader', () => ({
loadTelegramAdapter: vi.fn(),
}));
@@ -239,6 +239,60 @@ describe('TelegramIntegration.isUserAllowed', () => {
});
});
describe('TelegramIntegration.normalizeComponents', () => {
const { integration } = makeIntegration();
it('converts select and radio select options to individual buttons', () => {
expect(
integration.normalizeComponents([
{
type: 'select',
options: [
{ label: 'Approve', value: 'approve' },
{ label: 'Reject', value: 'reject' },
],
},
{
type: 'radio_select',
options: [{ label: 'Escalate', value: 'escalate' }],
},
]),
).toEqual([
{ type: 'button', label: 'Approve', value: 'approve' },
{ type: 'button', label: 'Reject', value: 'reject' },
{ type: 'button', label: 'Escalate', value: 'escalate' },
]);
});
it('converts image components to markdown link sections', () => {
expect(
integration.normalizeComponents([
{
type: 'image',
url: 'https://example.com/chart.png',
altText: 'Workflow chart',
},
]),
).toEqual([
{
type: 'section',
text: '[Workflow chart](https://example.com/chart.png)',
},
]);
});
it('passes supported Telegram components through unchanged', () => {
const components = [
{ type: 'section', text: 'Review this change' },
{ type: 'button', label: 'Approve', value: 'approve', style: 'primary' },
{ type: 'divider' },
{ type: 'fields', fields: [{ label: 'Risk', value: 'Low' }] },
];
expect(integration.normalizeComponents(components)).toEqual(components);
});
});
describe('TelegramIntegration secret token', () => {
const createTelegramAdapter = vi.fn();
@@ -0,0 +1,256 @@
import { readFileSync } from 'fs';
import { jsonParse } from 'n8n-workflow';
import { join } from 'path';
import type { TelegramReplayFixtures } from '../../../__tests__/helpers/telegram/replay-test-context';
import {
callbackPayloadWithData,
createTelegramReplayContext,
getTelegramInlineCallbackData,
} from '../../../__tests__/helpers/telegram/replay-test-context';
import type { ChannelIntegrationRecord } from '../../../recording/channel-integration-recorder';
// The chat SDK + adapters are ESM-only. Production loads them via esm-loader's
// `new Function()` hack to dodge the CJS transform, which can't run under vitest;
// redirect the loaders to native dynamic imports so the real adapters are used.
vi.mock('../../../esm-loader', () => ({
loadChatSdk: async () => await import('chat'),
loadMemoryState: async () => await import('@chat-adapter/state-memory'),
loadTelegramAdapter: async () => await import('@chat-adapter/telegram'),
loadSlackAdapter: async () => await import('@chat-adapter/slack'),
loadLinearAdapter: async () => await import('@chat-adapter/linear'),
}));
const telegramFixtures = jsonParse<TelegramReplayFixtures>(
readFileSync(join(__dirname, '../../../__tests__/fixtures/telegram/basic.json'), 'utf8'),
);
const recordedSession = jsonParse<ChannelIntegrationRecord[]>(
readFileSync(
join(__dirname, '../../../__tests__/fixtures/telegram/recorded-session.json'),
'utf8',
),
);
function getRecordedWebhook() {
const record = recordedSession.find(
(entry): entry is Extract<ChannelIntegrationRecord, { type: 'webhook' }> =>
entry.type === 'webhook' && entry.platform === 'telegram',
);
if (!record) throw new Error('Expected Telegram webhook record');
return record;
}
function getRecordedFetch(method: string) {
const record = recordedSession.find(
(entry): entry is Extract<ChannelIntegrationRecord, { type: 'fetch' }> =>
entry.type === 'fetch' && entry.url.endsWith(`/${method}`),
);
if (!record?.responseBody) throw new Error(`Expected Telegram ${method} fetch record`);
return record;
}
function recordedTelegramFixtures(): TelegramReplayFixtures {
const getMe = jsonParse<{
result: TelegramReplayFixtures['bot'];
}>(getRecordedFetch('getMe').responseBody ?? '{}');
const webhook = jsonParse<TelegramReplayFixtures['mention']>(getRecordedWebhook().body);
const message = webhook.message;
if (!message?.from) throw new Error('Expected recorded webhook message');
return {
bot: getMe.result,
user: message.from,
chat: message.chat,
mention: webhook,
followUp: {
update_id: webhook.update_id + 1,
message: {
...message,
message_id: message.message_id + 1,
text: 'follow up',
},
},
selfMessage: {
update_id: webhook.update_id + 2,
message: {
...message,
message_id: message.message_id + 2,
from: getMe.result,
text: 'bot echo',
},
},
callbackBase: telegramFixtures.callbackBase,
};
}
describe('Telegram recorded integration replay', () => {
afterEach(() => {
vi.restoreAllMocks();
});
it('replays the captured Telegram session webhook and outbound post', async () => {
const fixtures = recordedTelegramFixtures();
const ctx = await createTelegramReplayContext(fixtures, {
stream: [
{ type: 'text-delta', id: 'recorded-response', delta: 'Test response' },
{ type: 'finish', finishReason: 'stop' },
],
});
try {
await ctx.sendTelegramWebhook(fixtures.mention);
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenCalledWith(
expect.objectContaining({
message: 'hey',
integrationType: 'telegram',
}),
);
expect(ctx.latestContext()).toMatchObject({
platform: 'telegram',
messageId: '123456789:178',
interactingUserId: '123456789',
target: {
threadId: 'telegram:123456789',
channelId: 'telegram:123456789',
},
});
expect(ctx.lastApiCall('sendMessage')?.body).toMatchObject({
chat_id: '123456789',
text: 'Test response',
});
} finally {
await ctx.shutdown();
}
});
it('routes allowed private-mode Telegram users and ignores blocked users during replay', async () => {
const ctx = await createTelegramReplayContext(telegramFixtures, {
integration: {
type: 'telegram',
credentialId: 'cred-telegram',
settings: { accessMode: 'private', allowedUsers: ['alice_dev'] },
},
});
try {
await ctx.sendTelegramWebhook(telegramFixtures.mention);
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenCalledTimes(1);
} finally {
await ctx.shutdown();
}
const originalMessage = telegramFixtures.mention.message;
if (!originalMessage) throw new Error('Telegram mention fixture must include a message');
const blockedPayload = {
...telegramFixtures.mention,
message: {
...originalMessage,
from: {
id: 999999,
is_bot: false,
first_name: 'Mallory',
username: 'mallory',
},
},
};
const blockedCtx = await createTelegramReplayContext(telegramFixtures, {
integration: {
type: 'telegram',
credentialId: 'cred-telegram',
settings: { accessMode: 'private', allowedUsers: ['alice_dev'] },
},
});
try {
await blockedCtx.sendTelegramWebhook(blockedPayload);
expect(blockedCtx.agentExecutor.executeForChatPublished).not.toHaveBeenCalled();
} finally {
await blockedCtx.shutdown();
}
});
it('uses short callback data for Telegram rich cards and resumes the agent from callback queries', async () => {
const ctx = await createTelegramReplayContext(telegramFixtures, {
stream: [
{
type: 'tool-call-suspended',
runId: 'run-telegram-1',
toolCallId: 'tool-approval-1',
toolName: 'approval',
suspendPayload: {
type: 'approval',
toolName: 'send_telegram_message',
displayName: 'Send Telegram message',
args: { text: 'Ship it?' },
},
},
{ type: 'finish', finishReason: 'stop' },
],
});
try {
await ctx.sendTelegramWebhook(telegramFixtures.mention);
const cardMessage = ctx.lastApiCall('sendMessage');
const callbackData = getTelegramInlineCallbackData(cardMessage);
expect(callbackData).toBeDefined();
expect(Buffer.byteLength(callbackData ?? '', 'utf8')).toBeLessThanOrEqual(64);
ctx.nextStream([
{ type: 'text-delta', id: 'resume-text', delta: 'Approved' },
{ type: 'finish', finishReason: 'stop' },
]);
await ctx.sendTelegramWebhook(
callbackPayloadWithData(telegramFixtures.callbackBase, callbackData ?? '', 1000),
);
expect(ctx.agentExecutor.resumeForChat).toHaveBeenCalledWith(
expect.objectContaining({
runId: 'run-telegram-1',
toolCallId: 'tool-approval-1',
resumeData: { value: 'true' },
integrationType: 'telegram',
}),
);
expect(ctx.lastApiCall('deleteMessage')?.body).toMatchObject({
chat_id: '123456',
message_id: 1000,
});
expect(ctx.lastApiCall('answerCallbackQuery')?.body).toMatchObject({
callback_query_id: 'callback-1',
});
expect(ctx.lastApiCall('sendMessage')?.body).toMatchObject({
chat_id: '123456',
text: 'Approved',
});
} finally {
await ctx.shutdown();
}
});
it('sends Telegram direct messages through the integration action executor', async () => {
const ctx = await createTelegramReplayContext(telegramFixtures);
try {
const result = await ctx.actionExecutor.execute({
descriptor: ctx.descriptor,
action: 'send_dm',
input: { userId: '123456', message: { text: 'DM from agent' } },
awaitResponse: false,
});
expect(result).toMatchObject({
ok: true,
messageContext: {
platform: 'telegram',
target: { type: 'dm', userId: '123456', threadId: 'telegram:123456' },
},
});
expect(ctx.lastApiCall('sendMessage')?.body).toMatchObject({
chat_id: '123456',
text: 'DM from agent',
});
} finally {
await ctx.shutdown();
}
});
});
@@ -0,0 +1,170 @@
import {
createTelegramReplayContext,
getTelegramInlineCallbackData,
} from '../../../__tests__/helpers/telegram/replay-test-context';
import {
telegramCallbackQueryUpdate,
telegramGroupChat,
telegramMessageUpdate,
telegramReplayFixtures,
telegramUser,
} from '../../../__tests__/helpers/telegram/synthetic-fixtures';
// The chat SDK + adapters are ESM-only. Production loads them via esm-loader's
// `new Function()` hack to dodge the CJS transform, which can't run under vitest;
// redirect the loaders to native dynamic imports so the real adapters are used.
vi.mock('../../../esm-loader', () => ({
loadChatSdk: async () => await import('chat'),
loadMemoryState: async () => await import('@chat-adapter/state-memory'),
loadTelegramAdapter: async () => await import('@chat-adapter/telegram'),
loadSlackAdapter: async () => await import('@chat-adapter/slack'),
loadLinearAdapter: async () => await import('@chat-adapter/linear'),
}));
describe('Telegram Bot API integration scenarios', () => {
it('routes a Telegram group mention to a new agent conversation', async () => {
const group = telegramGroupChat();
const user = telegramUser({ id: 234567, username: 'group_user' });
const fixtures = telegramReplayFixtures({
user,
chat: group,
mention: telegramMessageUpdate({
message: {
message_id: 21,
from: user,
chat: group,
text: '@n8n_agent_bot hello group',
},
}),
});
const ctx = await createTelegramReplayContext(fixtures);
try {
await ctx.sendTelegramWebhook(fixtures.mention);
expect(ctx.agentExecutor.executeForChatPublished).toHaveBeenCalledWith(
expect.objectContaining({
message: '@n8n_agent_bot hello group',
integrationType: 'telegram',
}),
);
expect(ctx.latestContext()).toMatchObject({
messageId: '-1001234567890:21',
interactingUserId: '234567',
target: {
threadId: 'telegram:-1001234567890',
channelId: 'telegram:-1001234567890',
},
});
} finally {
await ctx.shutdown();
}
});
it('ignores a Bot API group message that does not mention the bot', async () => {
const group = telegramGroupChat();
const fixtures = telegramReplayFixtures({
chat: group,
mention: telegramMessageUpdate({
message: {
message_id: 22,
chat: group,
text: 'hello everyone',
},
}),
});
const ctx = await createTelegramReplayContext(fixtures);
try {
await ctx.sendTelegramWebhook(fixtures.mention);
expect(ctx.agentExecutor.executeForChatPublished).not.toHaveBeenCalled();
expect(ctx.latestContext()).toBeUndefined();
} finally {
await ctx.shutdown();
}
});
it('preserves Telegram forum topic IDs in thread context and outbound responses', async () => {
const group = telegramGroupChat();
const fixtures = telegramReplayFixtures({
chat: group,
mention: telegramMessageUpdate({
message: {
message_id: 23,
chat: group,
message_thread_id: 42,
text: '@n8n_agent_bot topic question',
},
}),
});
const ctx = await createTelegramReplayContext(fixtures);
try {
await ctx.sendTelegramWebhook(fixtures.mention);
expect(ctx.latestContext()).toMatchObject({
messageId: '-1001234567890:23',
target: {
threadId: 'telegram:-1001234567890:42',
channelId: 'telegram:-1001234567890',
},
});
expect(ctx.lastApiCall('sendMessage')?.body).toMatchObject({
chat_id: '-1001234567890',
message_thread_id: 42,
text: 'Got it',
});
} finally {
await ctx.shutdown();
}
});
it('resumes a suspended Telegram approval from an inline keyboard callback', async () => {
const fixtures = telegramReplayFixtures();
const ctx = await createTelegramReplayContext(fixtures, {
stream: [
{
type: 'tool-call-suspended',
runId: 'run-callback-1',
toolCallId: 'tool-callback-1',
toolName: 'approval',
suspendPayload: {
type: 'approval',
toolName: 'send_telegram_message',
displayName: 'Send Telegram message',
args: { text: 'Continue?' },
},
},
{ type: 'finish', finishReason: 'stop' },
],
});
try {
await ctx.sendTelegramWebhook(fixtures.mention);
const callbackData = getTelegramInlineCallbackData(ctx.lastApiCall('sendMessage'));
if (!callbackData) throw new Error('Expected inline keyboard markup');
ctx.nextStream([
{ type: 'text-delta', id: 'resume-text', delta: 'Callback handled' },
{ type: 'finish', finishReason: 'stop' },
]);
await ctx.sendTelegramWebhook(
telegramCallbackQueryUpdate({
data: callbackData,
message: fixtures.callbackBase.callback_query?.message,
}),
);
expect(ctx.agentExecutor.resumeForChat).toHaveBeenCalledWith(
expect.objectContaining({
runId: 'run-callback-1',
toolCallId: 'tool-callback-1',
resumeData: { value: 'true' },
integrationType: 'telegram',
}),
);
expect(ctx.lastApiCall('answerCallbackQuery')?.body).toMatchObject({
callback_query_id: 'callback-1',
});
} finally {
await ctx.shutdown();
}
});
});
@@ -0,0 +1,437 @@
import { mkdir, readFile, readdir, rm } from 'fs/promises';
import { jsonParse } from 'n8n-workflow';
import { join, resolve } from 'path';
export interface WebhookRecord {
type: 'webhook';
timestamp: number;
platform: string;
method: string;
url: string;
headers: Record<string, string>;
body: string;
}
export interface ApiCallRecord {
type: 'api-call';
timestamp: number;
platform: string;
method: string;
args: unknown[];
response?: unknown;
error?: string;
}
export interface FetchRecord {
type: 'fetch';
timestamp: number;
method: string;
url: string;
durationMs: number;
requestHeaders?: Record<string, string>;
requestBody?: string;
responseHeaders?: Record<string, string>;
responseBody?: string;
status?: number;
error?: string;
}
export type ChannelIntegrationRecord = WebhookRecord | ApiCallRecord | FetchRecord;
const SENSITIVE_HEADERS = new Set([
'authorization',
'cookie',
'set-cookie',
'x-api-key',
'x-auth-token',
'x-access-token',
'x-refresh-token',
'x-csrf-token',
'x-xsrf-token',
'x-slack-signature',
'x-telegram-bot-api-secret-token',
]);
const DEFAULT_FETCH_URL_PATTERNS = [
/\.slack\.com/i,
/api\.telegram\.org/i,
/api\.linear\.app/i,
/linear\.app/i,
];
const SANITIZED_N8N_HOST = 'https://n8n.host.com';
function sanitizeHeaderValue(key: string, value: string): string {
const normalizedKey = key.toLowerCase();
if (!SENSITIVE_HEADERS.has(normalizedKey)) return value;
return '[REDACTED]';
}
function sanitizeWebhookHeaderValue(key: string, value: string): string {
const normalizedKey = key.toLowerCase();
if (normalizedKey === 'x-forwarded-for') {
return '111.111.111.111';
}
if (normalizedKey === 'host' || normalizedKey === 'x-forwarded-host') {
return SANITIZED_N8N_HOST;
}
return sanitizeHeaderValue(key, value);
}
function sanitizeUrl(url: string): string {
if (url.includes('api.telegram.org')) {
return url.replace(/(.+api\.telegram\.org\/bot)(\d+:\S+)(\/.+)/, '$1123456789:abcdefghijkl$3');
}
return url;
}
function sanitizeWebhookUrl(url: string): string {
try {
const parsed = new URL(url);
return `${SANITIZED_N8N_HOST}${parsed.pathname}${parsed.search}${parsed.hash}`;
} catch {
return url;
}
}
function sanitizeHeaders(
headers: Headers | Record<string, string> | undefined,
sanitizeValue: (key: string, value: string) => string = sanitizeHeaderValue,
) {
if (!headers) return undefined;
const sanitized: Record<string, string> = {};
if (headers instanceof Headers) {
headers.forEach((value, key) => {
sanitized[key] = sanitizeValue(key, value);
});
return sanitized;
}
for (const [key, value] of Object.entries(headers)) {
sanitized[key] = sanitizeValue(key, value);
}
return sanitized;
}
function sanitizeRecord(record: ChannelIntegrationRecord): ChannelIntegrationRecord {
if (record.type === 'webhook') {
return {
...record,
url: sanitizeWebhookUrl(record.url),
headers: sanitizeHeaders(record.headers, sanitizeWebhookHeaderValue) ?? {},
};
}
if (record.type === 'fetch') {
return {
...record,
url: sanitizeUrl(record.url),
requestHeaders: sanitizeHeaders(record.requestHeaders),
responseHeaders: sanitizeHeaders(record.responseHeaders),
};
}
if (record.type === 'api-call') {
return {
...record,
response: sanitizeApiCallResponse(record.response),
};
}
throw new Error('Unsupported channel integration record type');
}
function sanitizeApiCallResponse(response: unknown): unknown {
if (!response || typeof response !== 'object') return response;
if (response instanceof Response) return response;
if (!('headers' in response) || !(response.headers instanceof Headers)) return response;
return {
...response,
headers: sanitizeHeaders(response.headers),
};
}
function sanitizeSessionId(value: string): string {
return value.replace(/[^a-zA-Z0-9._-]/g, '-');
}
function defaultSessionId(): string {
const ref = process.env.N8N_AGENT_INTEGRATION_RECORDING_REF ?? 'local';
return `session-${sanitizeSessionId(ref)}-${Date.now()}`;
}
function defaultRecordingDir(): string {
return resolve(process.cwd(), '.agent-recordings', 'channel-integrations');
}
async function responseToRecordable(response: unknown): Promise<unknown> {
if (response instanceof Response) {
return {
status: response.status,
headers: response.headers,
body: await response.clone().text(),
};
}
return response;
}
function getRequestUrl(input: RequestInfo | URL): string {
if (typeof input === 'string') return input;
if (input instanceof URL) return input.href;
return input.url;
}
function headersToRecord(headersInit?: HeadersInit): Record<string, string> | undefined {
if (!headersInit) return undefined;
if (headersInit instanceof Headers) {
const headers: Record<string, string> = {};
headersInit.forEach((value, key) => {
headers[key] = value;
});
return headers;
}
if (Array.isArray(headersInit)) {
const headers: Record<string, string> = {};
for (const [key, value] of headersInit) {
headers[key] = value;
}
return headers;
}
return headersInit;
}
function getRequestMethod(input: RequestInfo | URL, init?: RequestInit): string {
if (init?.method) return init.method;
if (input instanceof Request) return input.method;
return 'GET';
}
function getRequestHeaders(
input: RequestInfo | URL,
init?: RequestInit,
): Record<string, string> | undefined {
return headersToRecord(init?.headers ?? (input instanceof Request ? input.headers : undefined));
}
async function getRequestBody(
input: RequestInfo | URL,
init?: RequestInit,
): Promise<string | undefined> {
if (typeof init?.body === 'string') return init.body;
if (init?.body !== undefined) return undefined;
if (!(input instanceof Request)) return undefined;
return await input
.clone()
.text()
.catch(() => undefined);
}
export class ChannelIntegrationRecorder {
private readonly enabled: boolean;
private readonly sessionId: string;
private readonly recordingDir: string;
private originalFetch: typeof globalThis.fetch | undefined;
private fetchUrlPatterns = DEFAULT_FETCH_URL_PATTERNS;
private readonly pendingRecords = new Set<Promise<void>>();
constructor(options: { enabled?: boolean; sessionId?: string; recordingDir?: string } = {}) {
this.enabled =
options.enabled ?? process.env.N8N_AGENT_INTEGRATION_RECORDING_ENABLED === 'true';
this.sessionId = sanitizeSessionId(
options.sessionId ??
process.env.N8N_AGENT_INTEGRATION_RECORDING_SESSION_ID ??
defaultSessionId(),
);
this.recordingDir =
options.recordingDir ??
process.env.N8N_AGENT_INTEGRATION_RECORDING_DIR ??
defaultRecordingDir();
}
get isEnabled(): boolean {
return this.enabled;
}
get currentSessionId(): string {
return this.sessionId;
}
get currentSessionPath(): string {
return join(this.recordingDir, `${this.sessionId}.jsonl`);
}
async recordWebhook(platform: string, request: Request): Promise<void> {
if (!this.enabled) return;
await this.recordBestEffort(async () => {
const headers: Record<string, string> = {};
request.headers.forEach((value, key) => {
headers[key] = value;
});
await this.appendRecord({
type: 'webhook',
timestamp: Date.now(),
platform,
method: request.method,
url: request.url,
headers,
body: await request.clone().text(),
});
});
}
async recordApiCall(
platform: string,
method: string,
args: unknown[],
response?: unknown,
error?: Error,
): Promise<void> {
if (!this.enabled) return;
await this.recordBestEffort(async () => {
await this.appendRecord({
type: 'api-call',
timestamp: Date.now(),
platform,
method,
args,
response: await responseToRecordable(response),
...(error ? { error: error.message } : {}),
});
});
}
startFetchRecording(urlPatterns: RegExp[] = DEFAULT_FETCH_URL_PATTERNS): void {
if (!this.enabled || this.originalFetch) return;
this.fetchUrlPatterns = urlPatterns;
this.originalFetch = globalThis.fetch;
const originalFetch = this.originalFetch;
globalThis.fetch = async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
const url = getRequestUrl(input);
const shouldRecord = this.fetchUrlPatterns.some((pattern) => pattern.test(url));
if (!shouldRecord) return await originalFetch(input, init);
const startTime = Date.now();
const requestMethod = getRequestMethod(input, init);
const requestHeaders = getRequestHeaders(input, init);
const requestBody = await getRequestBody(input, init);
let response: Response | undefined;
let error: Error | undefined;
try {
response = await originalFetch(input, init);
return response;
} catch (caught) {
error = caught instanceof Error ? caught : new Error(String(caught));
throw caught;
} finally {
let responseHeaders: Record<string, string> | undefined;
if (response) {
responseHeaders = {};
response.headers.forEach((value, key) => {
responseHeaders![key] = value;
});
}
const responseBody = response
? await response
.clone()
.text()
.catch(() => undefined)
: undefined;
const record: FetchRecord = {
type: 'fetch',
timestamp: Date.now(),
method: requestMethod,
url,
durationMs: Date.now() - startTime,
requestHeaders,
requestBody,
status: response?.status,
responseHeaders,
responseBody,
...(error ? { error: error.message } : {}),
};
this.trackPendingRecord(this.appendRecord(record).catch(() => {}));
}
};
}
stopFetchRecording(): void {
if (!this.originalFetch) return;
globalThis.fetch = this.originalFetch;
this.originalFetch = undefined;
}
async listSessions(): Promise<Array<{ sessionId: string; entries: number }>> {
await mkdir(this.recordingDir, { recursive: true });
const files = await readdir(this.recordingDir);
const sessions = await Promise.all(
files
.filter((file) => file.endsWith('.jsonl'))
.map(async (file) => {
const contents = await readFile(join(this.recordingDir, file), 'utf8');
return {
sessionId: file.slice(0, -'.jsonl'.length),
entries: contents.split('\n').filter(Boolean).length,
};
}),
);
return sessions.sort((a, b) => a.sessionId.localeCompare(b.sessionId));
}
async getRecords(sessionId = this.sessionId): Promise<ChannelIntegrationRecord[]> {
await this.flush();
const filePath = join(this.recordingDir, `${sanitizeSessionId(sessionId)}.jsonl`);
const contents = await readFile(filePath, 'utf8');
return contents
.split('\n')
.filter(Boolean)
.map((line) => jsonParse<ChannelIntegrationRecord>(line));
}
async exportRecords(sessionId = this.sessionId): Promise<string> {
return JSON.stringify(await this.getRecords(sessionId), null, 2);
}
async deleteSession(sessionId = this.sessionId): Promise<void> {
await rm(join(this.recordingDir, `${sanitizeSessionId(sessionId)}.jsonl`), { force: true });
}
async flush(): Promise<void> {
await Promise.all([...this.pendingRecords]);
}
private async appendRecord(record: ChannelIntegrationRecord): Promise<void> {
await mkdir(this.recordingDir, { recursive: true });
const { appendFile } = await import('fs/promises');
await appendFile(
this.currentSessionPath,
`${JSON.stringify(sanitizeRecord(record))}\n`,
'utf8',
);
}
private async recordBestEffort(record: () => Promise<void>): Promise<void> {
await record().catch(() => {});
}
private trackPendingRecord(record: Promise<void>): void {
this.pendingRecords.add(record);
void record.finally(() => this.pendingRecords.delete(record));
}
}
export const channelIntegrationRecorder = new ChannelIntegrationRecorder();
@@ -0,0 +1,111 @@
import {
channelIntegrationRecorder,
type ChannelIntegrationRecorder,
} from './channel-integration-recorder';
const METHODS_BY_PLATFORM: Record<string, string[]> = {
slack: [
'postMessage',
'editMessage',
'deleteMessage',
'addReaction',
'removeReaction',
'startTyping',
'stream',
'openDM',
'fetchMessages',
],
telegram: [
'postMessage',
'editMessage',
'deleteMessage',
'addReaction',
'removeReaction',
'startTyping',
'openDM',
'fetchMessages',
],
linear: ['postMessage', 'editMessage', 'deleteMessage', 'addReaction', 'fetchMessages'],
};
type UnknownMethod = (this: unknown, ...args: unknown[]) => unknown;
function isObject(value: unknown): value is object {
return typeof value === 'object' && value !== null;
}
function isAsyncIterable(value: unknown): value is AsyncIterable<unknown> {
return isObject(value) && Symbol.asyncIterator in value;
}
function wrapStreamArg(args: unknown[]): { args: unknown[]; getChunks: () => unknown[] } {
const [threadId, textStream, ...rest] = args;
if (!isAsyncIterable(textStream)) {
return { args, getChunks: () => [] };
}
const chunks: unknown[] = [];
const wrappedStream: AsyncIterable<unknown> = {
[Symbol.asyncIterator]: () => {
const iterator = textStream[Symbol.asyncIterator]();
return {
async next() {
const result = await iterator.next();
if (!result.done) chunks.push(result.value);
return result;
},
};
},
};
return {
args: [threadId, wrappedStream, ...rest],
getChunks: () => chunks,
};
}
export function recordAdapterCalls(
platform: string,
adapter: unknown,
recorder: ChannelIntegrationRecorder = channelIntegrationRecorder,
methodsToRecord: string[] = METHODS_BY_PLATFORM[platform] ?? [],
): unknown {
if (!isObject(adapter)) return adapter;
if (!recorder.isEnabled || methodsToRecord.length === 0) return adapter;
return new Proxy(adapter, {
get(target, prop, receiver) {
const value = Reflect.get(target, prop, receiver) as unknown;
if (
typeof prop !== 'string' ||
typeof value !== 'function' ||
!methodsToRecord.includes(prop)
) {
return value;
}
const method = value as UnknownMethod;
return async (...args: unknown[]) => {
const streamRecording = prop === 'stream' ? wrapStreamArg(args) : undefined;
const callArgs = streamRecording?.args ?? args;
let response: unknown;
let error: Error | undefined;
try {
response = await method.apply(target, callArgs);
return response;
} catch (caught) {
error = caught instanceof Error ? caught : new Error(String(caught));
throw caught;
} finally {
const recordedArgs =
streamRecording === undefined
? args
: [args[0], { streamChunks: streamRecording.getChunks() }, ...args.slice(2)];
await recorder
.recordApiCall(platform, prop, recordedArgs, response, error)
.catch(() => {});
}
};
},
});
}