From 408aaa685d9c19c842bf3c45cdde97e614681576 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Wed, 4 Nov 2026 16:52:07 -0800 Subject: [PATCH 1/3] feat(discord): a pod's agent lines relay out to its bound Discord channel by mode (F132) Discord becomes the third outbound ChatRelayAdapter. With live relay on, the pod's agent lines go to the bound channel in mirror or attention mode, and mute holds them, as Telegram and Slack do. - Send through the row's stored channel webhook (encrypted webhookUrlRef, ?wait=true); the URL must match Discord's webhook pattern first. No bot token. allowed_mentions { parse: [] } on every send, @everyone/@here defused, names markdown-escaped; truncated at 2000 without splitting. - Pacing: one queue per webhook per replica, >= 400 ms apart; one retry on a short 429, otherwise a cooldown that drops (and logs) lines, which stay in the pod. - Failures: 10003/10015/50001/50013 set the row to error with a reason for a person; 429 and 5xx never change status. - Core stays generic: adapters declare directRows; the dispatcher selects chatRelayDirectRowTypes() and calls internal:.relay, so no platform is named there. Telegram dispatches as before. - Same gate as F131: the row's creator with a proven discordUserId; the send path re-checks linkedUserId === createdBy. - Connectors: a proven Discord row gets Manage, the Relay switch and Attention/Mirror; an unproven one says to add the bot again. Copy says the whole server reads the channel. en + zh-CN. - Gateway pre-filter: a cached discordUserId skips the per-line row read for everyone but the installer. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MDs9DPFVAooJnqCSwrdxVw --- .../unit/services/chatRelayCore.test.js | 21 +- .../connectorDeliveryFailureService.test.js | 70 ++++ .../discordBridgeService.outbound.test.js | 262 +++++++++++++++ .../services/discordBridgeService.test.js | 29 +- .../services/installableEventHandlers.test.js | 37 +++ backend/services/chatRelay/adapters.ts | 12 +- backend/services/chatRelay/chatRelayCore.ts | 11 +- .../connectorDeliveryFailureService.ts | 84 ++++- backend/services/discordBridgeService.ts | 308 +++++++++++++++++- backend/services/discordGatewayService.ts | 13 +- backend/services/installable/eventHandlers.ts | 17 +- backend/services/telegramBridgeService.ts | 2 + docs/integrations/INTEGRATION_CONTRACT.md | 38 ++- frontend/src/i18n/locales/en.json | 8 +- frontend/src/i18n/locales/zh-CN.json | 8 +- .../ConnectorsPage.channels.test.tsx | 78 +++++ frontend/src/v2/connectors/ChannelPanel.tsx | 28 +- frontend/src/v2/connectors/channelModel.ts | 12 + .../src/v2/connectors/useChannelConnectors.ts | 15 +- 19 files changed, 1004 insertions(+), 49 deletions(-) create mode 100644 backend/__tests__/unit/services/discordBridgeService.outbound.test.js diff --git a/backend/__tests__/unit/services/chatRelayCore.test.js b/backend/__tests__/unit/services/chatRelayCore.test.js index 5190fcdd5..7b5137ba1 100644 --- a/backend/__tests__/unit/services/chatRelayCore.test.js +++ b/backend/__tests__/unit/services/chatRelayCore.test.js @@ -269,7 +269,7 @@ describe('the adapter registry drives the dispatcher', () => { }); test('every registered platform has a .relay handler', () => { - expect(chatRelayIntegrationTypes()).toEqual(['telegram', 'slack']); + expect(chatRelayIntegrationTypes()).toEqual(['telegram', 'slack', 'discord']); chatRelayIntegrationTypes().forEach((provider) => { expect(typeof eventHandlers[`${provider}.relay`]).toBe('function'); }); @@ -283,7 +283,20 @@ describe('the adapter registry drives the dispatcher', () => { delete chatRelayAdapters.discordish; } const [[pipeline]] = Integration.aggregate.mock.calls; - expect(pipeline[0].$match.type).toEqual({ $in: ['telegram', 'slack', 'discordish'] }); + expect(pipeline[0].$match.type).toEqual({ $in: ['telegram', 'slack', 'discord', 'discordish'] }); + }); + + test('the direct-row arm reads each adapter’s directRows flag, not a type literal', async () => { + chatRelayAdapters.discordish = { provider: 'discordish', directRows: true }; + try { + await activeHandlersForPod(podId); + } finally { + delete chatRelayAdapters.discordish; + } + const [[pipeline]] = Integration.aggregate.mock.calls; + const projected = JSON.stringify(pipeline.find((stage) => stage.$project && stage.$project.handlers)); + expect(projected).toContain('["telegram","discord","discordish"]'); + expect(projected).toContain('"$concat":["internal:","$type",".relay"]'); }); }); @@ -326,14 +339,14 @@ describe('inbound: only a proven sender reaches the write (F131)', () => { expect(Pod.findById).not.toHaveBeenCalled(); }); - test('every platform that relays inbound declares an inbound adapter; outbound stays telegram and slack', () => { + test('every platform that relays inbound declares an inbound adapter; all three relay outbound (F132)', () => { expect(Object.keys(inboundRelayAdapters)).toEqual(['telegram', 'slack', 'discord']); Object.entries(inboundRelayAdapters).forEach(([provider, adapter]) => { expect(adapter.provider).toBe(provider); expect(typeof adapter.senderIsLinkedUser).toBe('function'); expect(typeof adapter.liveRelayRefusal).toBe('function'); }); - expect(chatRelayIntegrationTypes()).toEqual(['telegram', 'slack']); + expect(chatRelayIntegrationTypes()).toEqual(['telegram', 'slack', 'discord']); }); test('Telegram keeps the private-chat live-relay rule; an undeclared type falls back to it', () => { diff --git a/backend/__tests__/unit/services/connectorDeliveryFailureService.test.js b/backend/__tests__/unit/services/connectorDeliveryFailureService.test.js index 64fcfe235..e195f18ee 100644 --- a/backend/__tests__/unit/services/connectorDeliveryFailureService.test.js +++ b/backend/__tests__/unit/services/connectorDeliveryFailureService.test.js @@ -257,3 +257,73 @@ describe('which chat may flip', () => { expect(await Activity.countDocuments({})).toBe(0); }); }); + +// F132: Discord's webhook send reports failure as an HTTP status with a JSON +// error code. Its bound surface is `config.channelId`, which the row keeps for +// good (F90 fixes a Discord binding), so the flip names the reason and leaves +// the channel in place for a reconnect. +describe('Discord classification (F132)', () => { + const discordFailure = (httpStatus, code, extra = {}) => ({ + provider: 'discord', delivered: false, httpStatus, code, ...extra, + }); + + it('is permanent for the four codes that mean the channel cannot be posted to', () => { + expect(deliveryFailures.classifyDiscordDeliveryFailure(discordFailure(404, 10003))) + .toBe(deliveryFailures.DISCORD_CHANNEL_GONE_REASON); + expect(deliveryFailures.classifyDiscordDeliveryFailure(discordFailure(404, 10015))) + .toBe(deliveryFailures.DISCORD_WEBHOOK_GONE_REASON); + expect(deliveryFailures.classifyDiscordDeliveryFailure(discordFailure(403, 50001))) + .toBe(deliveryFailures.DISCORD_NO_ACCESS_REASON); + expect(deliveryFailures.classifyDiscordDeliveryFailure(discordFailure(403, 50013))) + .toBe(deliveryFailures.DISCORD_NO_PERMISSION_REASON); + }); + + it('never flips for a rate limit, a server error, content, or a success', () => { + [ + discordFailure(429, 0, { retryAfterMs: 1200 }), + discordFailure(429, undefined, { rateLimited: true }), + discordFailure(500, undefined), + discordFailure(400, 50035), + { provider: 'discord', delivered: true, httpStatus: 200, messageId: '1' }, + null, + ].forEach((result) => { + expect(deliveryFailures.classifyDiscordDeliveryFailure(result)).toBeNull(); + }); + }); + + it('sorts a Discord-shaped result to the Discord classifier', () => { + expect(deliveryFailures.classifyDeliveryFailure(discordFailure(404, 10003))) + .toBe(deliveryFailures.DISCORD_CHANNEL_GONE_REASON); + }); + + const makeDiscordRow = () => makeIntegration({ + name: 'Discord', + type: 'discord', + config: { serverId: '111111111111111111', channelId: '222222222222222222' }, + }); + + it('flips a Discord row on its bound channel and keeps the binding', async () => { + const integration = await makeDiscordRow(); + const flipped = await deliveryFailures.noteBoundChatDeliveryFailure( + integration.toObject(), '222222222222222222', discordFailure(403, 50013), + ); + expect(flipped).toBe(true); + const stored = await storedIntegration(integration._id); + expect(stored.status).toBe('error'); + expect(stored.errorMessage).toBe(deliveryFailures.DISCORD_NO_PERMISSION_REASON); + expect(stored.errorMessageUserFacing).toBe(true); + expect(stored.config.channelId).toBe('222222222222222222'); + }); + + it('does not flip a Discord row for a 429, or for a channel it is not bound to', async () => { + const integration = await makeDiscordRow(); + expect(await deliveryFailures.noteBoundChatDeliveryFailure( + integration.toObject(), '222222222222222222', discordFailure(429, 0, { retryAfterMs: 500 }), + )).toBe(false); + expect(await deliveryFailures.noteBoundChatDeliveryFailure( + integration.toObject(), '333333333333333333', discordFailure(404, 10003), + )).toBe(false); + const stored = await storedIntegration(integration._id); + expect(stored.status).toBe('connected'); + }); +}); diff --git a/backend/__tests__/unit/services/discordBridgeService.outbound.test.js b/backend/__tests__/unit/services/discordBridgeService.outbound.test.js new file mode 100644 index 000000000..52d27c1ec --- /dev/null +++ b/backend/__tests__/unit/services/discordBridgeService.outbound.test.js @@ -0,0 +1,262 @@ +// F132 (D249): a Discord row with live relay on carries the pod's agent lines +// back to its bound channel, by mode, through the shared chat-relay core — the +// same ladder Telegram and Slack run. What only Discord knows is pinned here: +// the channel webhook send, the 2000-character limit, a send that can never +// ping a guild, and the webhook's rate limit. +jest.mock('discord.js', () => ({ + Client: jest.fn(), + GatewayIntentBits: { Guilds: 1, GuildMessages: 2, MessageContent: 4 }, +})); +jest.mock('axios', () => ({ post: jest.fn(), get: jest.fn(), create: jest.fn() })); +jest.mock('../../../models/Integration', () => ({ + find: jest.fn(), + findOne: jest.fn(), + findById: jest.fn(), + findByIdAndUpdate: jest.fn(), + findOneAndUpdate: jest.fn(), +})); +jest.mock('../../../models/Pod', () => ({ findById: jest.fn() })); +jest.mock('../../../models/pg/Message', () => ({ create: jest.fn(), findById: jest.fn() })); +jest.mock('../../../services/channelVerdictService', () => ({ record: jest.fn() })); +jest.mock('../../../services/connectorSecrets', () => ({ get: jest.fn() })); + +const axios = require('axios'); +const Integration = require('../../../models/Integration'); +const Pod = require('../../../models/Pod'); +const PGMessage = require('../../../models/pg/Message'); +const connectorSecrets = require('../../../services/connectorSecrets'); +const { relayAgentMessageThroughAdapter } = require('../../../services/chatRelay/chatRelayCore'); +const { + discordRelayAdapter, + discordWebhookPacing, + relayDiscordMessageToPod, + DISCORD_CONTENT_LIMIT, +} = require('../../../services/discordBridgeService'); +const deliveryFailures = require('../../../services/connectorDeliveryFailureService'); + +const POD = '64b7f0c2a1b2c3d4e5f60718'; +const CREATOR = '64b7f0c2a1b2c3d4e5f60719'; +const GUILD = '111111111111111111'; +const CHANNEL = '222222222222222222'; +const INSTALLER_DISCORD = '345678901234567890'; +const WEBHOOK_ID = '456789012345678901'; +const WEBHOOK_URL = `https://discord.com/api/webhooks/${WEBHOOK_ID}/tok-en_123`; + +let rowSeq = 0; +const discordRow = (configOverrides = {}, rowOverrides = {}) => { + rowSeq += 1; + return { + _id: `row-${rowSeq}`, + type: 'discord', + podId: POD, + isActive: true, + status: 'connected', + createdBy: CREATOR, + config: { + serverId: GUILD, + channelId: CHANNEL, + webhookUrlRef: 'secret-ref-1', + liveRelay: true, + relayAllAgentMessages: true, + linkedUserId: CREATOR, + discordUserId: INSTALLER_DISCORD, + ...configOverrides, + }, + ...rowOverrides, + }; +}; + +const ok = (id = '999000111222333444') => ({ status: 200, data: { id }, headers: {} }); +const failure = (status, data, headers = {}) => ({ status, data, headers }); + +const relay = (integration, overrides = {}) => relayAgentMessageThroughAdapter(discordRelayAdapter, { + podId: POD, + agentUsername: 'kai', + displayName: 'Kai', + content: 'The build is green.', + podMessageId: 'pod-msg-1', + integration, + ...overrides, +}); + +const sentBody = (call = 0) => axios.post.mock.calls[call][1]; + +let sleeps; +beforeEach(() => { + jest.clearAllMocks(); + discordWebhookPacing.reset(); + sleeps = []; + discordWebhookPacing.sleep = jest.fn(async (ms) => { sleeps.push(ms); }); + Pod.findById.mockReturnValue({ select: () => ({ lean: async () => ({ name: 'Launch' }) }) }); + connectorSecrets.get.mockResolvedValue(WEBHOOK_URL); + axios.post.mockResolvedValue(ok()); + Integration.findByIdAndUpdate.mockResolvedValue({}); + Integration.findOneAndUpdate.mockResolvedValue(null); +}); + +describe('mode', () => { + test('mirror: every agent line reaches the bound channel webhook', async () => { + await relay(discordRow({ relayAllAgentMessages: true })); + expect(axios.post).toHaveBeenCalledTimes(1); + const [url] = axios.post.mock.calls[0]; + expect(url).toBe(`${WEBHOOK_URL}?wait=true`); + expect(sentBody().content).toBe('[Launch] **Kai**: The build is green.'); + // The sent line is remembered, as on the other platforms. + expect(Integration.findByIdAndUpdate).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ + $push: expect.objectContaining({ + 'config.relayMap': expect.objectContaining({ + $each: [expect.objectContaining({ externalMessageId: '999000111222333444', agentUsername: 'kai' })], + }), + }), + })); + }); + + test('attention: an ordinary line stays in the pod; an escalation reaches the channel', async () => { + const row = discordRow({ relayAllAgentMessages: false }); + await relay(row, { content: 'Just a status note.' }); + expect(axios.post).not.toHaveBeenCalled(); + await relay(row, { content: '[ESCALATE] the deploy needs a human' }); + expect(axios.post).toHaveBeenCalledTimes(1); + expect(sentBody().content).toContain('[ESCALATE] the deploy needs a human'); + }); + + test('mute holds every line until it expires', async () => { + await relay(discordRow({ relayMutedUntil: new Date(Date.now() + 60_000) })); + expect(axios.post).not.toHaveBeenCalled(); + await relay(discordRow({ relayMutedUntil: new Date(Date.now() - 60_000) })); + expect(axios.post).toHaveBeenCalledTimes(1); + }); +}); + +describe('who it sends for', () => { + test.each([ + ['live relay off', discordRow({ liveRelay: false })], + ['an admin pause', discordRow({ adminPause: { reason: 'x' } })], + ['an error row', discordRow({}, { status: 'error' })], + ['no proven installer', discordRow({ discordUserId: undefined })], + ['live relay switched on by someone other than the creator', discordRow({ linkedUserId: '64b7f0c2a1b2c3d4e5f60799' })], + ['a row for another pod', discordRow({}, { podId: '64b7f0c2a1b2c3d4e5f60700' })], + ])('sends nothing for %s', async (_label, row) => { + await relay(row); + expect(axios.post).not.toHaveBeenCalled(); + }); + + test('sends nothing when the stored webhook URL is not a Discord webhook', async () => { + connectorSecrets.get.mockResolvedValue('https://example.com/collect'); + await relay(discordRow()); + expect(axios.post).not.toHaveBeenCalled(); + }); +}); + +describe('content', () => { + test('a long line is cut to Discord’s 2000-character limit, with an ellipsis', async () => { + await relay(discordRow(), { content: 'x'.repeat(5000) }); + const { content } = sentBody(); + expect(content.length).toBe(DISCORD_CONTENT_LIMIT); + expect(content.startsWith('[Launch] **Kai**: xxx')).toBe(true); + expect(content.endsWith('…')).toBe(true); + }); + + test('a cut never splits a surrogate pair', async () => { + const prefixLength = '[Launch] **Kai**: '.length; + // Puts an emoji's high surrogate exactly at the cut point. + const content = `${'x'.repeat(DISCORD_CONTENT_LIMIT - 2 - prefixLength)}😀${'y'.repeat(50)}`; + await relay(discordRow(), { content }); + const sent = sentBody().content; + expect(sent.length).toBeLessThanOrEqual(DISCORD_CONTENT_LIMIT); + expect(/[\ud800-\udbff]…$/.test(sent)).toBe(false); + }); + + test('the send can ping no one: allowed_mentions is empty, and @everyone/@here are defused', async () => { + await relay(discordRow(), { content: '@everyone and @here, see <@123456789012345678> and <@&987654321098765432>' }); + const body = sentBody(); + expect(body.allowed_mentions).toEqual({ parse: [] }); + expect(body.content).not.toMatch(/@everyone|@here/); + expect(body.content).toContain('@​everyone'); + }); + + test('the display name is markdown-escaped so an agent cannot restyle the prefix', async () => { + await relay(discordRow(), { displayName: '**Boss**' }); + expect(sentBody().content.startsWith('[Launch] **\\*\\*Boss\\*\\***: ')).toBe(true); + }); + + test('a decision card renders inside the limit and links back to the pod', async () => { + await relay(discordRow(), { + card: { + title: 'Choose a train', + question: 'Which rollout should ship? '.repeat(200), + options: [{ label: 'Canary' }, { label: 'Fast lane', recommended: true }], + }, + }); + const { content, allowed_mentions: mentions } = sentBody(); + expect(content.length).toBeLessThanOrEqual(DISCORD_CONTENT_LIMIT); + expect(content).toContain('Kai needs a ruling'); + expect(content).toMatch(/]+\/v2\/pods\/64b7f0c2a1b2c3d4e5f60718\?message=pod-msg-1>$/); + expect(mentions).toEqual({ parse: [] }); + }); +}); + +describe('failures', () => { + test.each([ + [404, 10003, 'DISCORD_CHANNEL_GONE_REASON'], + [403, 50013, 'DISCORD_NO_PERMISSION_REASON'], + ])('a %i with code %i marks the bound channel as failing', async (status, code, reasonKey) => { + axios.post.mockResolvedValue(failure(status, { code, message: 'nope' })); + const row = discordRow(); + await relay(row); + expect(Integration.findOneAndUpdate).toHaveBeenCalledWith( + { _id: row._id, 'config.channelId': CHANNEL }, + expect.objectContaining({ + $set: expect.objectContaining({ status: 'error', errorMessage: deliveryFailures[reasonKey] }), + }), + expect.anything(), + ); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); + + test('a 429 waits out retry_after once, then sends — never a tight retry', async () => { + axios.post + .mockResolvedValueOnce(failure(429, { message: 'You are being rate limited.', retry_after: 1.5, global: false })) + .mockResolvedValueOnce(ok()); + await relay(discordRow()); + expect(axios.post).toHaveBeenCalledTimes(2); + expect(sleeps).toContain(1500); + expect(Integration.findOneAndUpdate).not.toHaveBeenCalled(); + }); + + test('a long 429 does not retry; the lane cools down and the next line is not sent into it', async () => { + axios.post.mockResolvedValueOnce(failure(429, { message: 'slow down', retry_after: 30 })); + const row = discordRow(); + await relay(row); + expect(axios.post).toHaveBeenCalledTimes(1); + expect(sleeps.some((ms) => ms >= 30_000)).toBe(false); + await relay(row, { content: 'second line' }); + expect(axios.post).toHaveBeenCalledTimes(1); + expect(Integration.findOneAndUpdate).not.toHaveBeenCalled(); + }); + + test('back-to-back lines to one webhook are paced at about 5 per 2 seconds', async () => { + const row = discordRow(); + await Promise.all([relay(row, { content: 'one' }), relay(row, { content: 'two' }), relay(row, { content: 'three' })]); + expect(axios.post).toHaveBeenCalledTimes(3); + expect(sleeps.filter((ms) => ms > 0).length).toBeGreaterThanOrEqual(2); + sleeps.filter((ms) => ms > 0).forEach((ms) => expect(ms).toBeLessThanOrEqual(discordWebhookPacing.minIntervalMs)); + }); +}); + +describe('echo', () => { + test('the channel’s copy of a relayed agent line is not relayed back into the pod', async () => { + const echo = { + id: '999000111222333444', + type: 0, + webhookId: WEBHOOK_ID, + guild: { id: GUILD }, + channel: { id: CHANNEL }, + author: { id: WEBHOOK_ID, username: 'Commonly Bot', bot: true }, + content: '[Launch] **Kai**: The build is green.', + }; + await expect(relayDiscordMessageToPod({ integration: discordRow(), message: echo })) + .resolves.toEqual({ relayed: false }); + expect(PGMessage.create).not.toHaveBeenCalled(); + }); +}); diff --git a/backend/__tests__/unit/services/discordBridgeService.test.js b/backend/__tests__/unit/services/discordBridgeService.test.js index 20de418d9..1b61c4cc5 100644 --- a/backend/__tests__/unit/services/discordBridgeService.test.js +++ b/backend/__tests__/unit/services/discordBridgeService.test.js @@ -291,7 +291,12 @@ describe('Discord inbound live relay (F131)', () => { describe('the gateway', () => { const target = (overrides = {}) => ({ - integrationId: 'integration-1', webhookListenerEnabled: true, maxBufferSize: 50, liveRelay: true, ...overrides, + integrationId: 'integration-1', + webhookListenerEnabled: true, + maxBufferSize: 50, + liveRelay: true, + discordUserId: INSTALLER_DISCORD, + ...overrides, }); beforeEach(() => { @@ -328,6 +333,28 @@ describe('Discord inbound live relay (F131)', () => { expect(writes()).toHaveLength(0); }); + // F132 review follow-up: the cached proven id pre-filters, so a line from + // anyone but the installer costs no row read at all. + test('does not read the row for a line from anyone but the cached installer', async () => { + discordGatewayService.channelCache.set(`${GUILD}:${CHANNEL}`, [target()]); + + await discordGatewayService.handleMessageCreate( + discordMessage({ author: { id: OTHER_DISCORD, username: 'mallory', bot: false } }), + ); + + expect(Integration.findById).not.toHaveBeenCalled(); + expect(Integration.findByIdAndUpdate).toHaveBeenCalledTimes(1); + }); + + test('does not read the row when the cache holds no proven installer', async () => { + discordGatewayService.channelCache.set(`${GUILD}:${CHANNEL}`, [target({ discordUserId: undefined })]); + + await discordGatewayService.handleMessageCreate(discordMessage()); + + expect(Integration.findById).not.toHaveBeenCalled(); + expect(writes()).toHaveLength(0); + }); + test('does not hand a non-live row to the bridge', async () => { discordGatewayService.channelCache.set(`${GUILD}:${CHANNEL}`, [target({ liveRelay: false })]); diff --git a/backend/__tests__/unit/services/installableEventHandlers.test.js b/backend/__tests__/unit/services/installableEventHandlers.test.js index 2fe196071..2f1d354f1 100644 --- a/backend/__tests__/unit/services/installableEventHandlers.test.js +++ b/backend/__tests__/unit/services/installableEventHandlers.test.js @@ -341,6 +341,43 @@ describe('installable event dispatcher', () => { expect(relay.mock.calls[0][0].podId).toBe(podId); }); + // F132: a Discord row is a direct row (the OAuth install writes it; there is + // no Installable projection), so the selector's direct-row arm must hand it + // to `discord.relay` — and only while its live relay is on. + it('dispatches a direct Discord row with live relay on, and nothing for one without it', async () => { + const podId = freshId(); + const ownerId = freshId(); + await createPod(podId, ownerId, [ownerId]); + const discordConfig = { + serverId: '111111111111111111', channelId: '222222222222222222', discordUserId: '345678901234567890', + }; + const live = await Integration.create({ + podId, type: 'discord', status: 'connected', createdBy: ownerId, isActive: true, + config: { ...discordConfig, liveRelay: true, linkedUserId: ownerId }, + }); + const otherPodId = freshId(); + await createPod(otherPodId, ownerId, [ownerId]); + await Integration.create({ + podId: otherPodId, type: 'discord', status: 'connected', createdBy: ownerId, isActive: true, + config: { ...discordConfig, liveRelay: false }, + }); + const originalDiscordHandler = eventHandlers['discord.relay']; + const relay = jest.fn().mockResolvedValue(undefined); + eventHandlers['discord.relay'] = relay; + try { + await dispatch('chat.message', { + podId, agentUsername: 'kai', displayName: 'Kai', content: 'to the channel', podMessageId: 'm-1', + }); + await dispatch('chat.message', { + podId: otherPodId, agentUsername: 'kai', displayName: 'Kai', content: 'stays home', podMessageId: 'm-2', + }); + } finally { + eventHandlers['discord.relay'] = originalDiscordHandler; + } + expect(relay).toHaveBeenCalledTimes(1); + expect(String(relay.mock.calls[0][0].integration._id)).toBe(String(live._id)); + }); + // TASK-267a: a post from a native turn that read web results // carries noOutboundRelay, and the dispatcher refuses it where bridges are // selected, so a live bridge (here enabled before dispatch, as if during the diff --git a/backend/services/chatRelay/adapters.ts b/backend/services/chatRelay/adapters.ts index 4f04a6d71..efb4b06c0 100644 --- a/backend/services/chatRelay/adapters.ts +++ b/backend/services/chatRelay/adapters.ts @@ -14,16 +14,22 @@ const { telegramRelayAdapter, telegramInboundAdapter } = require('../telegramBri // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { slackRelayAdapter, slackInboundAdapter } = require('../slackBridgeService'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require -const { discordInboundAdapter } = require('../discordBridgeService'); +const { discordRelayAdapter, discordInboundAdapter } = require('../discordBridgeService'); export const chatRelayAdapters: Record> = { telegram: telegramRelayAdapter, slack: slackRelayAdapter, + discord: discordRelayAdapter, }; /** Integration.type values the dispatcher selects for outbound relay. */ export const chatRelayIntegrationTypes = (): string[] => Object.keys(chatRelayAdapters); +/** The types whose rows relay with no Installable parent (see ChatRelayAdapter.directRows). */ +export const chatRelayDirectRowTypes = (): string[] => ( + Object.entries(chatRelayAdapters).filter(([, adapter]) => adapter.directRows === true).map(([type]) => type) +); + /** `{ 'telegram.relay': fn, 'slack.relay': fn }` — the dispatcher's handler entries. */ export const chatRelayEventHandlers = (): Record) => Promise> => ( Object.fromEntries(Object.entries(chatRelayAdapters).map(([provider, adapter]) => [ @@ -34,8 +40,8 @@ export const chatRelayEventHandlers = (): Record> = { diff --git a/backend/services/chatRelay/chatRelayCore.ts b/backend/services/chatRelay/chatRelayCore.ts index d19caa2f4..b57488c1d 100644 --- a/backend/services/chatRelay/chatRelayCore.ts +++ b/backend/services/chatRelay/chatRelayCore.ts @@ -22,7 +22,8 @@ // proof has passed. // // Adding a platform is one adapter in its bridge plus one entry in -// ./adapters.ts — the dispatcher's selector reads the registry's keys. +// ./adapters.ts — the dispatcher's selector reads the registry's keys, and its +// `directRows` flag. // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const IntegrationModel = require('../../models/Integration'); @@ -190,6 +191,14 @@ export interface ChatRelayAdapter.relay` itself; an adapter without this flag + * is reached only through an installed event-handler component. + */ + directRows?: boolean; findLiveIntegration: (podId: string) => Promise; isRelayableIntegration: (integration: I, podId: string) => boolean; /** Render and send. Return null to skip without a send (e.g. no bot token). */ diff --git a/backend/services/connectorDeliveryFailureService.ts b/backend/services/connectorDeliveryFailureService.ts index 56709c893..40fe93825 100644 --- a/backend/services/connectorDeliveryFailureService.ts +++ b/backend/services/connectorDeliveryFailureService.ts @@ -37,6 +37,40 @@ const PERMANENT_SLACK_ERRORS: { [error: string]: string } = { is_archived: SLACK_CHANNEL_ARCHIVED_REASON, }; +// Discord's webhook send (F132) reports failure as an HTTP status carrying a +// JSON error `code`. These four mean the channel cannot be posted to until a +// person acts; everything else (429, 5xx, a 400 on content) is transient or +// content and says nothing about the binding. +const DISCORD_CHANNEL_GONE_REASON = 'Discord stopped delivering: this channel no longer exists.'; +const DISCORD_WEBHOOK_GONE_REASON = 'Discord stopped delivering: the channel\'s Commonly webhook was deleted.'; +const DISCORD_NO_ACCESS_REASON = 'Discord stopped delivering: the bot can no longer see this channel.'; +const DISCORD_NO_PERMISSION_REASON = 'Discord stopped delivering: the bot is not allowed to post in this channel.'; + +const PERMANENT_DISCORD_CODES: { [code: number]: string } = { + 10003: DISCORD_CHANNEL_GONE_REASON, // Unknown Channel + 10015: DISCORD_WEBHOOK_GONE_REASON, // Unknown Webhook + 50001: DISCORD_NO_ACCESS_REASON, // Missing Access + 50013: DISCORD_NO_PERMISSION_REASON, // Missing Permissions +}; + +/** What the Discord relay adapter hands over: tagged, so it cannot be mistaken for the other shapes. */ +interface DiscordDeliveryResult { + provider?: 'discord'; + delivered?: boolean; + httpStatus?: number; + code?: number; +} + +/** + * Discord's JSON error `code` → a named reason, or null. A rate limit (429) is + * never permanent: the adapter backs off and the next line tries again. + */ +const classifyDiscordDeliveryFailure = (result?: DiscordDeliveryResult | null): string | null => { + if (!result || result.provider !== 'discord' || result.delivered === true) return null; + if (result.httpStatus === 429 || typeof result.code !== 'number') return null; + return PERMANENT_DISCORD_CODES[result.code] || null; +}; + interface DeliveryResult { success?: boolean; errorCode?: number; @@ -89,6 +123,13 @@ interface FlipInput { integrationId: unknown; failedChatId: unknown; reason: string; + /** + * The config key that holds the bound surface. `chatId` (Telegram, Slack) is + * unset by the flip, which is what lets that connector re-bind. A Discord + * row's `channelId` is fixed for the row's life (F90), so it is matched on + * and kept: a reconnect clears the error on the same channel. + */ + boundChatKey?: 'chatId' | 'channelId'; } /** @@ -135,13 +176,13 @@ const flipConnectorOnPermanentDeliveryFailure = async ({ integrationId, failedChatId, reason, + boundChatKey = 'chatId', }: FlipInput): Promise => { const matched = await Integration.findOneAndUpdate( - { _id: integrationId, 'config.chatId': String(failedChatId) }, - { - $set: userFacingError(reason), - $unset: { 'config.chatId': '' }, - }, + { _id: integrationId, [`config.${boundChatKey}`]: String(failedChatId) }, + boundChatKey === 'chatId' + ? { $set: userFacingError(reason), $unset: { 'config.chatId': '' } } + : { $set: userFacingError(reason) }, { new: true }, ); if (!matched) return false; @@ -149,18 +190,26 @@ const flipConnectorOnPermanentDeliveryFailure = async ({ }; /** - * The two providers report failure in their own shapes — `{success: false, - * errorCode}` and `{ok: false, error}` — and the call sites hand over whatever + * The providers report failure in their own shapes — `{success: false, + * errorCode}`, `{ok: false, error}` and Discord's tagged `{provider: 'discord', + * httpStatus, code}` — and the call sites hand over whatever * their provider returned, so the entry point sorts them by shape rather than * making every caller name its provider. */ -const classifyDeliveryFailure = (result?: (DeliveryResult & SlackDeliveryResult) | null): string | null => ( - result && result.ok === false ? classifySlackDeliveryFailure(result) : classifyTelegramDeliveryFailure(result) -); +type AnyDeliveryResult = DeliveryResult & SlackDeliveryResult & DiscordDeliveryResult; + +const classifyDeliveryFailure = (result?: AnyDeliveryResult | null): string | null => { + if (result && result.provider === 'discord') return classifyDiscordDeliveryFailure(result); + return result && result.ok === false ? classifySlackDeliveryFailure(result) : classifyTelegramDeliveryFailure(result); +}; + +// Where each provider's row keeps the surface it was bound to. Telegram and +// Slack store it as `chatId`; a Discord row as `channelId`. +const boundChatKeyFor = (type: unknown): 'chatId' | 'channelId' => (type === 'discord' ? 'channelId' : 'chatId'); /** * Call this only from a send that targeted the connector's own bound chat — for - * either provider, since both store it in `config.chatId`. + * any provider: `boundChatKeyFor` names where its row stores that chat. * * `sentToChatId` is compared against the stored one on purpose: the sites that * may flip all send to `integration.config.chatId`, and the receive-side replies @@ -169,11 +218,12 @@ const classifyDeliveryFailure = (result?: (DeliveryResult & SlackDeliveryResult) * flipping someone else's working connector. */ const noteBoundChatDeliveryFailure = async ( - integration: { _id?: unknown; config?: { chatId?: unknown } } | null | undefined, + integration: { _id?: unknown; type?: unknown; config?: { chatId?: unknown; channelId?: unknown } } | null | undefined, sentToChatId: unknown, - result?: (DeliveryResult & SlackDeliveryResult) | null, + result?: AnyDeliveryResult | null, ): Promise => { - const boundChatId = integration?.config?.chatId; + const boundChatKey = boundChatKeyFor(integration?.type); + const boundChatId = integration?.config?.[boundChatKey]; if (!boundChatId || !sentToChatId || String(boundChatId) !== String(sentToChatId)) return false; const reason = classifyDeliveryFailure(result); @@ -183,6 +233,7 @@ const noteBoundChatDeliveryFailure = async ( integrationId: integration?._id, failedChatId: boundChatId, reason, + boundChatKey, }); }; @@ -207,6 +258,7 @@ const markConnectorUnavailable = async ( export { classifyDeliveryFailure, + classifyDiscordDeliveryFailure, classifySlackDeliveryFailure, classifyTelegramDeliveryFailure, flipConnectorOnPermanentDeliveryFailure, @@ -217,4 +269,8 @@ export { SLACK_BOT_REMOVED_REASON, SLACK_CHANNEL_ARCHIVED_REASON, SLACK_CHANNEL_GONE_REASON, + DISCORD_CHANNEL_GONE_REASON, + DISCORD_WEBHOOK_GONE_REASON, + DISCORD_NO_ACCESS_REASON, + DISCORD_NO_PERMISSION_REASON, }; diff --git a/backend/services/discordBridgeService.ts b/backend/services/discordBridgeService.ts index ea10400a5..9631dcd00 100644 --- a/backend/services/discordBridgeService.ts +++ b/backend/services/discordBridgeService.ts @@ -8,18 +8,30 @@ // and stored server-side as `config.discordUserId`. Everyone else's lines keep // the gateway's buffer-for-summary path, which this module does not touch. // -// Outbound is not here: Discord is not in the outbound relay registry, and an -// agent's pod message does not reach the channel through this bridge. +// Outbound (F132): with live relay on, the pod's agent lines reach the bound +// channel by mode, through the core's hold/mute/escalate ladder, sent by the +// channel's own webhook. Agent text reaches everyone in the guild, so the send +// can ping no one (`allowed_mentions: { parse: [] }`), and it is paced to the +// webhook's rate limit. The webhook's own echo of each line is a webhook-authored +// message, which `isIgnoredDiscordMessage` drops, so nothing loops back in. // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const Integration = require('../models/Integration'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const axios = require('axios'); +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const { isGatedPodTarget } = require('./connectorRelayPolicy'); +import { resolveDiscordWebhookUrl } from '../utils/discordWebhookUrl'; import { claimInboundMessage, loadInboundAuthor, proveInboundSender, writeInboundAsLinkedUser, } from './chatRelay/chatRelayCore'; -import type { InboundRelayAdapter, LiveRelayRow } from './chatRelay/chatRelayCore'; +import type { + ChatRelayAdapter, InboundRelayAdapter, LiveRelayRow, OutboundRelayOptions, +} from './chatRelay/chatRelayCore'; +import type { DecisionRelayCard } from './decisionCardRelay'; const LOG = '[discord-bridge]'; const AUTHOR_LOAD_ATTEMPTS = 3; @@ -42,6 +54,12 @@ export interface DiscordIntegrationDoc { linkedUserId?: string; discordUserId?: string; adminPause?: unknown; + webhookUrlRef?: string; + webhookUrl?: string; + relayAllAgentMessages?: boolean; + relayMutedUntil?: Date | string; + leadAgentUsername?: string; + gates?: Record; }; } @@ -100,7 +118,7 @@ const isInboundRelayableIntegration = ( const LIVE_RELAY_NEEDS_PROOF = 'Live relay on a Discord channel needs the Discord account of the person who ' + 'added the bot, which the install proves. Add the bot again from Commonly to link it.'; const LIVE_RELAY_INSTALLER_ONLY = 'Only the person who added the bot can turn on live relay for this channel: ' - + 'it posts their own Discord messages into the pod as them.'; + + 'it posts their own Discord messages into the pod as them, and the pod\'s agent messages into the channel.'; export const discordInboundAdapter: InboundRelayAdapter = { provider: 'discord', @@ -230,8 +248,290 @@ export const relayDiscordMessageForIntegration = async ( return relayDiscordMessageToPod({ integration, message }); }; + +// --------------------------------------------------------------------------- +// Outbound (F132) — agent message → the bound channel, by its webhook. +// --------------------------------------------------------------------------- + +/** Discord rejects a message `content` longer than this. */ +export const DISCORD_CONTENT_LIMIT = 2000; + +// The channel webhook the install created, and nothing else: a legacy +// plaintext `config.webhookUrl` is a field a row has carried from a request +// body, so the URL must be Discord's own webhook endpoint before agent text is +// posted to it. +const DISCORD_WEBHOOK_URL = new RegExp( + '^https://(?:(?:ptb|canary)\\.)?discord(?:app)?\\.com/api/(?:v\\d+/)?webhooks/\\d+/[\\w-]+$', +); + +// Same proof as inbound: live relay was switched on by the person whose Discord +// account the install proved (`linkedUserId` is stamped from that caller, and +// liveRelayRefusal admits only the creator holding a `discordUserId`). A row +// that fails it was turned on some other way and stays silent. +const isRelayableIntegration = (integration: DiscordIntegrationDoc, podId: string): boolean => ( + isGatedPodTarget(integration, podId) + && integration.type === 'discord' + && integration.isActive === true + && integration.status !== 'error' + && integration.config?.liveRelay === true + && !integration.config?.adminPause + && Boolean(integration.config?.serverId) + && Boolean(integration.config?.channelId) + && Boolean(integration.config?.discordUserId) + && Boolean(integration.config?.linkedUserId) + && String(integration.config?.linkedUserId) === idOf(integration.createdBy) +); + +const findLiveIntegration = async (podId: string): Promise => ( + Integration.findOne({ + type: 'discord', + isActive: true, + status: { $ne: 'error' }, + podId, + 'config.liveRelay': true, + 'config.channelId': { $exists: true, $ne: null }, + 'config.adminPause': { $exists: false }, + }).lean() +); + +// Bold and the other inline marks would let a display or pod name restyle the +// prefix; a backslash makes each one literal. +const escapeDiscordMarkdown = (value: string): string => value.replace(/([\\*_~`|>])/g, '\\$1'); + +// `allowed_mentions` already stops every ping. This also keeps the words from +// reading as one: a zero-width space after the @ renders the same and is no +// longer the mass-mention token. +const defuseMassMentions = (value: string): string => value.replace(/@(everyone|here)/gi, '@​$1'); + +/** Cut to `limit` characters with an ellipsis, never leaving half a surrogate pair. */ +const fitWithin = (value: string, limit: number): string => { + if (value.length <= limit) return value; + if (limit <= 1) return '…'.slice(0, Math.max(0, limit)); + let cut = value.slice(0, limit - 1); + if (/[\ud800-\udbff]$/.test(cut)) cut = cut.slice(0, -1); + return `${cut}…`; +}; + +const podNameForTag = async (podId: string): Promise => { + try { + // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require + const Pod = require('../models/Pod'); + const pod = await Pod.findById(podId).select('name').lean(); + return String(pod?.name || podId); + } catch (error) { + console.warn(`${LOG} pod name lookup failed, tagging with the id:`, (error as Error).message); + return podId; + } +}; + +/** `[Pod] **Agent**: text`, the shape the Slack and Telegram lines take, in 2000 characters. */ +export const renderDiscordAgentLine = (opts: { podName: string; name: string; content: string }): string => { + const prefix = `[${escapeDiscordMarkdown(opts.podName)}] **${escapeDiscordMarkdown(opts.name)}**: `; + return fitWithin(prefix + defuseMassMentions(String(opts.content || '')), DISCORD_CONTENT_LIMIT); +}; + +// Discord has no reply-to-rule path yet (no relayMap reader on inbound), so the +// card sends the reader to the pod rather than asking for a reply here. The +// link sits in angle brackets so Discord does not unfurl it. +export const renderDiscordDecisionCard = (opts: { + card: DecisionRelayCard; + name: string; + link: string; +}): string => { + const { card, name, link } = opts; + const footer = `\nRule on it in Commonly → <${link}>`; + const options = card.options.map((option, index) => { + const description = option.description ? `\n ${fitWithin(String(option.description), 100)}` : ''; + const star = option.recommended ? ' ★ recommended' : ''; + return `\n${index + 1}. **${escapeDiscordMarkdown(option.label)}**${star}${description}`; + }).join(''); + const title = `**${escapeDiscordMarkdown(name)} needs a ruling · ${escapeDiscordMarkdown(card.title)}**`; + const header = `${title}\n${card.question}`; + const optionsBlock = fitWithin(defuseMassMentions(options), Math.floor(DISCORD_CONTENT_LIMIT / 2)); + const body = fitWithin( + defuseMassMentions(header), + DISCORD_CONTENT_LIMIT - footer.length - optionsBlock.length, + ); + return body + optionsBlock + footer; +}; + +/** What a webhook send hands the core and `connectorDeliveryFailureService`. */ +export interface DiscordDeliveryResult { + provider: 'discord'; + delivered: boolean; + httpStatus?: number; + code?: number; + messageId?: string; + retryAfterMs?: number; + /** Not sent: the webhook's lane is still cooling down from a 429. */ + rateLimited?: boolean; +} + +/** + * The webhook's rate limit, about 5 messages per 2 seconds per webhook. Each + * row's webhook gets one lane: its sends run one at a time, at least + * `minIntervalMs` apart. A 429 waits out Discord's `retry_after` once and + * retries; a longer wait is not taken inline — the lane cools down for it and + * lines arriving meanwhile are dropped and logged, never retried in a loop. + * Per process: each backend replica paces its own sends, and Discord's 429 is + * the backstop between them. + */ +interface WebhookLane { tail: Promise; nextAt: number; coolUntil: number } +export const discordWebhookPacing = { + minIntervalMs: 400, + maxRetryWaitMs: 5_000, + defaultRetryAfterMs: 1_000, + now: (): number => Date.now(), + sleep: (ms: number): Promise => new Promise((resolve) => { setTimeout(resolve, ms); }), + lanes: new Map(), + reset(): void { this.lanes.clear(); }, +}; + +const retryAfterMsOf = (response: { data?: Record; headers?: Record }): number => { + const body = Number(response?.data?.retry_after); + if (Number.isFinite(body) && body >= 0) return Math.ceil(body * 1000); + const header = Number(response?.headers?.['retry-after']); + if (Number.isFinite(header) && header >= 0) return Math.ceil(header * 1000); + return discordWebhookPacing.defaultRetryAfterMs; +}; + +const postToWebhook = async (url: string, payload: Record): Promise => { + const response = await axios.post(url, payload, { + headers: { 'Content-Type': 'application/json' }, + timeout: 10_000, + validateStatus: () => true, + }); + const status = Number(response?.status); + const data = (response?.data || {}) as { id?: unknown; code?: unknown }; + if (status >= 200 && status < 300 && data.id) { + return { provider: 'discord', delivered: true, httpStatus: status, messageId: String(data.id) }; + } + return { + provider: 'discord', + delivered: false, + httpStatus: status, + ...(typeof data.code === 'number' ? { code: data.code } : {}), + ...(status === 429 ? { retryAfterMs: retryAfterMsOf(response) } : {}), + }; +}; + +const sendThroughLane = ( + laneKey: string, + url: string, + payload: Record, +): Promise => { + const pacing = discordWebhookPacing; + let lane = pacing.lanes.get(laneKey); + if (!lane) { + lane = { tail: Promise.resolve(), nextAt: 0, coolUntil: 0 }; + pacing.lanes.set(laneKey, lane); + } + const current = lane; + const run = async (): Promise => { + if (pacing.now() < current.coolUntil) { + return { + provider: 'discord', delivered: false, httpStatus: 429, rateLimited: true, + retryAfterMs: current.coolUntil - pacing.now(), + }; + } + const wait = current.nextAt - pacing.now(); + if (wait > 0) await pacing.sleep(wait); + let result = await postToWebhook(url, payload); + current.nextAt = pacing.now() + pacing.minIntervalMs; + if (result.httpStatus === 429) { + const backoff = result.retryAfterMs ?? pacing.defaultRetryAfterMs; + if (backoff <= pacing.maxRetryWaitMs) { + await pacing.sleep(backoff); + result = await postToWebhook(url, payload); + current.nextAt = pacing.now() + pacing.minIntervalMs; + } + if (result.httpStatus === 429) { + current.coolUntil = pacing.now() + (result.retryAfterMs ?? pacing.defaultRetryAfterMs); + } + } + return result; + }; + const next = current.tail.then(run); + current.tail = next.catch(() => undefined); + return next; +}; + +const webhookUrlFor = async (integration: DiscordIntegrationDoc): Promise => { + let legacyPlatformUrl: unknown; + if (!integration.config?.webhookUrlRef && !integration.config?.webhookUrl) { + // Rows connected before the webhook URL was encrypted kept it on the + // platform document; the migration script moves it into the ref. + // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require + const DiscordIntegration = require('../models/DiscordIntegration'); + const platform = await DiscordIntegration.findOne({ integrationId: integration._id }).select('webhookUrl').lean(); + legacyPlatformUrl = platform?.webhookUrl; + } + const url = await resolveDiscordWebhookUrl( + integration.config?.webhookUrlRef, + integration.config?.webhookUrl, + legacyPlatformUrl, + ); + if (!url || !DISCORD_WEBHOOK_URL.test(url)) return null; + return url; +}; + +const withWait = (url: string): string => { + const parsed = new URL(url); + parsed.searchParams.set('wait', 'true'); + return parsed.toString(); +}; + +export const discordRelayAdapter: ChatRelayAdapter = { + provider: 'discord', + relayMapKey: 'externalMessageId', + // The OAuth install writes the row; there is no Installable projection. + directRows: true, + findLiveIntegration, + isRelayableIntegration, + send: async ({ + integration, podId, agentUsername, displayName, content, card, link, + }: OutboundRelayOptions & { integration: DiscordIntegrationDoc; link: string }) => { + const webhookUrl = await webhookUrlFor(integration); + const chatId = integration.config?.channelId; + if (!webhookUrl) { + console.warn(`${LOG} outbound skipped integration=${String(integration._id)}: no Discord webhook on the row`); + return null; + } + const name = displayName || agentUsername; + const text = card + ? renderDiscordDecisionCard({ card, name, link }) + : renderDiscordAgentLine({ podName: await podNameForTag(podId), name, content }); + const result = await sendThroughLane(String(integration._id), withWait(webhookUrl), { + content: text, + // Agent text goes to everyone in the guild: no @everyone, @here, role or + // user ping, whatever the text says. + allowed_mentions: { parse: [] }, + }); + return { chatId, externalMessageId: result.delivered ? result.messageId || null : null, result }; + }, + deliveryFailureError: (result) => { + const r = (result || {}) as DiscordDeliveryResult; + if (r.rateLimited) { + return new Error(`webhook lane cooling down after a 429 (${r.retryAfterMs}ms left); line dropped`); + } + const code = r.code !== undefined ? ` code ${r.code}` : ''; + return new Error(`webhook send failed: HTTP ${String(r.httpStatus)}${code}`); + }, + logOutboundFailure: (error, opts) => { + console.warn( + `${LOG} outbound relay failed integration=${String(opts.integration?._id || 'lookup')} ` + + `channel=${String(opts.integration?.config?.channelId || 'unknown')}: ${error.message}`, + ); + }, +}; + module.exports = { discordInboundAdapter, + discordRelayAdapter, + discordWebhookPacing, + renderDiscordAgentLine, + renderDiscordDecisionCard, + DISCORD_CONTENT_LIMIT, isIgnoredDiscordMessage, renderDiscordInboundContent, relayDiscordMessageToPod, diff --git a/backend/services/discordGatewayService.ts b/backend/services/discordGatewayService.ts index ca672eaf7..bece8056d 100644 --- a/backend/services/discordGatewayService.ts +++ b/backend/services/discordGatewayService.ts @@ -18,6 +18,10 @@ interface ChannelTarget { maxBufferSize: number; // Only a hint for which rows to hand the bridge; the bridge re-reads the row. liveRelay?: boolean; + // The installer's proven Discord id (F131). Server-owned and fixed at the + // row's creation, so a minute-stale copy is still right: any other author's + // line skips the per-message row read, since the bridge would refuse it. + discordUserId?: string; } interface IntegrationDoc { @@ -28,6 +32,7 @@ interface IntegrationDoc { webhookListenerEnabled?: boolean; maxBufferSize?: number; liveRelay?: boolean; + discordUserId?: string; }; } @@ -124,7 +129,7 @@ class DiscordGatewayService { isActive: true, }) .select('_id config.channelId config.serverId config.webhookListenerEnabled config.maxBufferSize ' - + 'config.liveRelay') + + 'config.liveRelay config.discordUserId') .lean() as IntegrationDoc[]; this.channelCache.clear(); @@ -141,6 +146,7 @@ class DiscordGatewayService { webhookListenerEnabled: !!integration?.config?.webhookListenerEnabled, maxBufferSize: integration?.config?.maxBufferSize || 1000, liveRelay: integration?.config?.liveRelay === true, + ...(integration?.config?.discordUserId ? { discordUserId: String(integration.config.discordUserId) } : {}), }); this.channelCache.set(key, existing); }); @@ -202,7 +208,10 @@ class DiscordGatewayService { // F131: a live-relay row also gets the line in its pod, when the bridge // proves the author. The buffer above is unchanged for every line, relayed // or not, so channel summaries still see the whole channel. - const liveTargets = targets.filter((target) => target.liveRelay); + const authorId = String(message.author?.id || ''); + const liveTargets = targets.filter((target) => ( + target.liveRelay && Boolean(target.discordUserId) && target.discordUserId === authorId + )); if (!liveTargets.length) return; // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports const { relayDiscordMessageForIntegration } = require('./discordBridgeService'); diff --git a/backend/services/installable/eventHandlers.ts b/backend/services/installable/eventHandlers.ts index 8c96979d9..cd9daf0ec 100644 --- a/backend/services/installable/eventHandlers.ts +++ b/backend/services/installable/eventHandlers.ts @@ -6,7 +6,7 @@ import { Types } from 'mongoose'; const Integration = require('../../models/Integration'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const Pod = require('../../models/Pod'); -import { chatRelayEventHandlers, chatRelayIntegrationTypes } from '../chatRelay/adapters'; +import { chatRelayDirectRowTypes, chatRelayEventHandlers, chatRelayIntegrationTypes } from '../chatRelay/adapters'; // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { handleGithubGig } = require('../githubGigService'); @@ -29,7 +29,7 @@ export type InternalEventHandler = ( // component tests can pin selection at this seam and future component types // add one registration instead of another dispatcher branch. export const eventHandlers: Record = { - // `telegram.relay`, `slack.relay`, … — one per chat-relay adapter. + // `telegram.relay`, `slack.relay`, `discord.relay` — one per chat-relay adapter. ...chatRelayEventHandlers(), // Inbound App activity is selected by the webhook's indexed pod binding. 'github.gig': handleGithubGig, @@ -65,6 +65,7 @@ const activeHandlersForPod = async (podId: string, includeCardHolds = false): Pr if (!memberIds.length) return []; const gateEnabledPath = `config.gates.${String(podId)}.enabled`; const relayTypes = chatRelayIntegrationTypes(); + const directRowTypes = chatRelayDirectRowTypes(); // A decision card must record why it was deliberately withheld. Ordinary // chat retains the narrow selector below; cards additionally reach paused // and gate-off member bindings so their bridge can write a `hold` verdict @@ -137,18 +138,20 @@ const activeHandlersForPod = async (podId: string, includeCardHolds = false): Pr { $project: { integration: '$$ROOT', - // Existing direct Integration rows predate D17. The old live-relay - // shape is Telegram-only; Slack is always an Installable projection - // because its token is an opaque ConnectorSecret reference. + // A direct Integration row (no Installable parent) relays only when + // its adapter declares `directRows`: legacy Telegram binds that predate + // D17, and Discord, whose OAuth install writes the row itself (F132). + // Slack is always an Installable projection because its token is an + // opaque ConnectorSecret reference. handlers: { $cond: [ { $and: [ { $eq: [{ $ifNull: ['$installationId', null] }, null] }, - { $eq: ['$type', 'telegram'] }, + { $in: ['$type', directRowTypes] }, ], }, - [{ config: { eventHandler: 'internal:telegram.relay' } }], + [{ config: { eventHandler: { $concat: ['internal:', '$type', '.relay'] } } }], { $let: { vars: { parent: { $arrayElemAt: ['$installation', 0] } }, diff --git a/backend/services/telegramBridgeService.ts b/backend/services/telegramBridgeService.ts index 6ca7d179a..1de871668 100644 --- a/backend/services/telegramBridgeService.ts +++ b/backend/services/telegramBridgeService.ts @@ -277,6 +277,8 @@ const podNameForTag = async (podId: string): Promise => { export const telegramRelayAdapter: ChatRelayAdapter = { provider: 'telegram', relayMapKey: 'tgMessageId', + // Legacy direct binds predate D17 and still relay. + directRows: true, findLiveIntegration, isRelayableIntegration, send: async ({ diff --git a/docs/integrations/INTEGRATION_CONTRACT.md b/docs/integrations/INTEGRATION_CONTRACT.md index c81a4cef8..bf37a90b1 100644 --- a/docs/integrations/INTEGRATION_CONTRACT.md +++ b/docs/integrations/INTEGRATION_CONTRACT.md @@ -158,8 +158,42 @@ delivery per update. A gateway socket per replica doesn't. Once the claim is held, no other replica writes the line. So the read-only author load (`loadInboundAuthor`) is retried up to three times on a thrown error, then logged with the same two ids. The write itself is single-attempt. -Inbound and outbound are separate registries: -Discord relays inbound only, and is not in `chatRelayAdapters`. +Inbound and outbound are separate registries (`inboundRelayAdapters` and +`chatRelayAdapters`); since F132 all three chat platforms are in both. + +## Agent lines out to the chat (F132) +With `liveRelay` on, an agent's pod message reaches the bound chat through the +core's ladder (`relayAgentMessageThroughAdapter`): card hold, relayable row, +mute (`relayMutedUntil`), then mode (`connectorRelayPolicy.shouldEscalate`: +`mirror` sends every line, `attention` only escalations and the lead). Each +platform is one `ChatRelayAdapter` in its bridge, registered in +`chatRelay/adapters.ts`. The dispatcher (`installable/eventHandlers.ts`) selects +rows by the registry's keys, and hands a row with no Installable parent to +`.relay` only when its adapter declares `directRows` (Telegram's legacy +binds, and every Discord row, which the OAuth install writes directly). + +**Discord** sends through the channel webhook the install created +(`config.webhookUrlRef`), never a bot token in the browser: +- `allowed_mentions: { parse: [] }`, so agent text pings nobody in the guild, + and `@everyone`/`@here` are defused in the text too; +- `[Pod] **Agent**: text`, cut to Discord's 2000 characters with an ellipsis + (the Telegram and Slack adapters truncate too; none splits a line); +- one lane per webhook, about 5 sends per 2 s. A 429 waits out `retry_after` + once (up to 5 s) and retries; a longer wait cools the lane, and lines that + arrive meanwhile are dropped and logged. **An under-delivering guard, by + design (rule 19):** the line is still in the pod, and the audience is the + operator log, which names the row and channel. Pacing is per replica; + Discord's 429 is the backstop between replicas; +- 10003, 10015, 50001 and 50013 flip the row to `status: 'error'` with a + reason written for a person (`connectorDeliveryFailureService`). The flip + keeps `config.channelId` (F90 fixes the binding); `POST /:id/connect` sets + the row connected again once the bot can post. It does not recreate a + deleted webhook (10015) while `webhookUrlRef` is set, so that case needs the + bot added again. 429 and 5xx never flip. + +The webhook's copy of each line is webhook-authored, and the inbound gateway +ignores webhook messages, so nothing loops back into the pod. Who may switch +it on is F131's rule: only the row's creator holding a proven `discordUserId`. ## Testing contract - Shared Jest contract tests under `backend/__tests__/contracts/integrationProvider.test.js`: diff --git a/frontend/src/i18n/locales/en.json b/frontend/src/i18n/locales/en.json index f728d2da4..c505ce86c 100644 --- a/frontend/src/i18n/locales/en.json +++ b/frontend/src/i18n/locales/en.json @@ -2664,7 +2664,13 @@ "untitledPod": "Untitled pod", "anotherPod": "another pod", "thisSlackWorkspace": "This Slack workspace", - "yourSlackUser": "your Slack user" + "yourSlackUser": "your Slack user", + "discordRowMirror": "every agent line reaches everyone in the channel", + "discordRowAttention": "escalations reach everyone in the channel", + "discordMirrorHint": "Every agent message is posted in the Discord channel, where everyone in the server can read it.", + "discordAttentionHint": "Only escalations and the lead agent are posted in the Discord channel, where everyone in the server can read them.", + "discordInboundNote": "Your own messages in the channel reach the pod as you.", + "discordNeedsProof": "Add the bot again from Commonly to turn on relay for this channel." }, "artifacts": { "title": "Artifacts", diff --git a/frontend/src/i18n/locales/zh-CN.json b/frontend/src/i18n/locales/zh-CN.json index 5687032f4..77678acde 100644 --- a/frontend/src/i18n/locales/zh-CN.json +++ b/frontend/src/i18n/locales/zh-CN.json @@ -2665,7 +2665,13 @@ "linear": "你自己的 Linear 工作空间中的议题、项目、周期和发布。", "supabase": "项目、数据库结构、Edge Functions 和分支。断开连接会删除 Commonly 的令牌，但在有人到 OAuth Apps 里移除之前，Commonly 在该组织里的授权仍然有效。", "notion": "你的 Notion 工作空间中的页面、评论和数据库。" - } + }, + "discordRowMirror": "智能体的每条消息都会发给频道里的所有人", + "discordRowAttention": "升级消息会发给频道里的所有人", + "discordMirrorHint": "每条智能体消息都会发到这个 Discord 频道，服务器里的每个人都能看到。", + "discordAttentionHint": "只有需要关注的消息和主导智能体的消息会发到这个 Discord 频道，服务器里的每个人都能看到。", + "discordInboundNote": "你在频道里发的消息，会以你的身份进入 Pod。", + "discordNeedsProof": "从 Commonly 重新添加一次机器人，就能为这个频道开启中继。" }, "artifacts": { "title": "产物", diff --git a/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx b/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx index 11809462b..1602ac284 100644 --- a/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx +++ b/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx @@ -1709,3 +1709,81 @@ const ConnectorsPageForRoutes = () => { const Page = require('../connectors/ConnectorsPage').default; return ; }; + +// F132 (D249): a Discord row relays agent lines out to a SHARED channel, so its +// relay switch is offered only on a row whose installer the bot-add proved — +// the server stamps `discordUserId` on the creator's row and withholds it from +// everyone else's view. The copy says who reads the channel. +describe('Connectors page: Discord live relay (F132)', () => { + const discordRow = (config = {}) => ({ + _id: 'i-discord', + type: 'discord', + status: 'connected', + createdAt: new Date(Date.now() - 3_600_000).toISOString(), + updatedAt: new Date(Date.now() - 3_600_000).toISOString(), + config: { serverId: '111111111111111111', channelId: '222222222222222222', ...config }, + podId: { _id: 'p2', name: 'Ops' }, + }); + + beforeEach(() => { + jest.clearAllMocks(); + window.history.replaceState({}, '', '/v2/connectors'); + }); + + afterEach(() => jest.restoreAllMocks()); + + it('offers the installer the relay switch, and turning it on PATCHes liveRelay', async () => { + mockGets([discordRow({ discordUserId: '345678901234567890', liveRelay: false })]); + axios.patch.mockResolvedValue({ data: {} }); + renderPage(); + const drawer = await openDrawer('Discord'); + expect(within(drawer).getByText('What the channel sees')).toBeInTheDocument(); + expect(within(drawer).getByText('Messages stay in the pod.')).toBeInTheDocument(); + fireEvent.click(within(drawer).getByRole('checkbox', { name: 'Relay' })); + await waitFor(() => expect(axios.patch).toHaveBeenCalledWith( + '/api/integrations/i-discord', + { config: { liveRelay: true } }, + expect.anything(), + )); + }); + + it('says the whole server reads the channel, and switches mode', async () => { + mockGets([discordRow({ discordUserId: '345678901234567890', liveRelay: true, relayAllAgentMessages: true })]); + axios.patch.mockResolvedValue({ data: {} }); + renderPage(); + const drawer = await openDrawer('Discord'); + expect(within(drawer).getByText( + 'Every agent message is posted in the Discord channel, where everyone in the server can read it.', + )).toBeInTheDocument(); + expect(within(drawer).getByText('Your own messages in the channel reach the pod as you.')).toBeInTheDocument(); + fireEvent.click(within(drawer).getByRole('button', { name: 'Attention' })); + await waitFor(() => expect(axios.patch).toHaveBeenCalledWith( + '/api/integrations/i-discord', + { config: { relayAllAgentMessages: false } }, + expect.anything(), + )); + }); + + it('shows no relay switch on a row whose installer was not proven, and says how to fix it', async () => { + mockGets([discordRow({})]); + renderPage(); + const drawer = await openDrawer('Discord'); + expect(within(drawer).queryByRole('checkbox', { name: 'Relay' })).toBeNull(); + expect(within(drawer).getByText('Add the bot again from Commonly to turn on relay for this channel.')).toBeInTheDocument(); + }); + + it('renders the shared-channel copy in zh-CN', async () => { + await i18nReady; + await act(async () => { await i18n.changeLanguage('zh-CN'); }); + try { + mockGets([discordRow({ discordUserId: '345678901234567890', liveRelay: true, relayAllAgentMessages: false })]); + renderPage(); + const drawer = await openDrawer('Discord'); + expect(within(drawer).getByText( + '只有需要关注的消息和主导智能体的消息会发到这个 Discord 频道，服务器里的每个人都能看到。', + )).toBeInTheDocument(); + } finally { + await act(async () => { await i18n.changeLanguage('en'); }); + } + }); +}); diff --git a/frontend/src/v2/connectors/ChannelPanel.tsx b/frontend/src/v2/connectors/ChannelPanel.tsx index b790b622b..7ce5df3fd 100644 --- a/frontend/src/v2/connectors/ChannelPanel.tsx +++ b/frontend/src/v2/connectors/ChannelPanel.tsx @@ -11,6 +11,7 @@ import { codeIsLive, connectorRelativeTimeOptions, botMembers, + discordRelayReady, } from './channelModel'; import type { Connector, @@ -229,8 +230,12 @@ const ChannelPanel: React.FC = ({ m, item: panelItem }) => { const connector = item.connector as Connector; const isTelegram = connector.type === 'telegram'; const isSlack = connector.type === 'slack'; + const isDiscord = connector.type === 'discord'; const connected = connector.status === 'connected'; - const supportsRelayControls = isTelegram || isSlack; + // F132: a Discord row shows the card always (its hint says why a switch is + // missing) and the switch only when its installer is proven. + const supportsRelayControls = isTelegram || isSlack || isDiscord; + const relaySwitchOffered = !isDiscord || discordRelayReady(connector); const mirror = connector.config?.relayAllAgentMessages === true; const pendingBind = connector.config?.pendingBind; const activePod = podList.find((pod) => pod._id === connectorPodId(connector)); @@ -337,11 +342,23 @@ const ChannelPanel: React.FC = ({ m, item: panelItem }) => {

{connector.config?.liveRelay ? (mirror - ? t('connectors.mirrorHint', { defaultValue: 'Every agent message reaches the channel.' }) - : t('connectors.attentionHint', { defaultValue: 'Only escalations and the lead agent reach the channel.' })) + ? (isDiscord + ? t('connectors.discordMirrorHint', { defaultValue: 'Every agent message is posted in the Discord channel, where everyone in the server can read it.' }) + : t('connectors.mirrorHint', { defaultValue: 'Every agent message reaches the channel.' })) + : (isDiscord + ? t('connectors.discordAttentionHint', { defaultValue: 'Only escalations and the lead agent are posted in the Discord channel, where everyone in the server can read them.' }) + : t('connectors.attentionHint', { defaultValue: 'Only escalations and the lead agent reach the channel.' }))) : t('connectors.relayOffHint', { defaultValue: 'Messages stay in the pod.' })}

- {connected && ( + {isDiscord && relaySwitchOffered && connector.config?.liveRelay && ( +

{t('connectors.discordInboundNote', { defaultValue: 'Your own messages in the channel reach the pod as you.' })}

+ )} + {isDiscord && !relaySwitchOffered && ( +

+ {t('connectors.discordNeedsProof', { defaultValue: 'Add the bot again from Commonly to turn on relay for this channel.' })} +

+ )} + {connected && relaySwitchOffered && (
)} {connected && userScoped && renderGates(connector)} - {renderRemove(item)} + {/* Discord's removal is its own uninstall path, not this PATCH. */} + {!isDiscord && renderRemove(item)} )} diff --git a/frontend/src/v2/connectors/channelModel.ts b/frontend/src/v2/connectors/channelModel.ts index 02247bdb1..b8ec93fac 100644 --- a/frontend/src/v2/connectors/channelModel.ts +++ b/frontend/src/v2/connectors/channelModel.ts @@ -24,6 +24,10 @@ export interface ConnectorConfig { liveRelay?: boolean; relayAllAgentMessages?: boolean; leadAgentUsername?: string; + // F131/F132: the Discord account the bot-add proved for this row's creator. + // Server-owned, and withheld from every viewer but the creator, so its + // presence is what says "you may switch live relay on here". + discordUserId?: string; gates?: Record; // Projected by the server as { reason, at } only — the admin is never named. adminPause?: { reason?: string; at?: string }; @@ -260,6 +264,14 @@ export const SLACK_REFUSALS: Record ( + connector?.type === 'discord' && Boolean(connector.config?.discordUserId) +); + export const connectorPodId = (connector: Connector | null | undefined): string | null => { if (!connector || !connector.podId) return null; return typeof connector.podId === 'object' ? String(connector.podId._id || '') || null : String(connector.podId); diff --git a/frontend/src/v2/connectors/useChannelConnectors.ts b/frontend/src/v2/connectors/useChannelConnectors.ts index 8ff03c1d6..875638645 100644 --- a/frontend/src/v2/connectors/useChannelConnectors.ts +++ b/frontend/src/v2/connectors/useChannelConnectors.ts @@ -30,6 +30,7 @@ import { projectionMissing, LINK_INSTALL_PROVIDERS, entryOffered, + discordRelayReady, } from './channelModel'; import type { ConnectorGate, @@ -728,13 +729,19 @@ export const useChannelConnectors = () => { const mirror = connector.config?.relayAllAgentMessages === true; const recent = connector.updatedAt && (Date.now() - new Date(connector.updatedAt).getTime()) < RECENT_MS; const markName: MarkName = relay ? (mirror ? 'mirror' : 'attention') : 'off'; + // A Discord channel is shared: everyone in the server reads what lands there. + const isDiscord = connector.type === 'discord'; + const mirrorLabel = isDiscord + ? t('connectors.discordRowMirror', { defaultValue: 'every agent line reaches everyone in the channel' }) + : t('connectors.rowMirror', { defaultValue: 'every agent line reaches the channel' }); + const attentionLabel = isDiscord + ? t('connectors.discordRowAttention', { defaultValue: 'escalations reach everyone in the channel' }) + : t('connectors.rowAttention', { defaultValue: 'escalations reach the channel' }); const markLabel = relay - ? (mirror - ? t('connectors.rowMirror', { defaultValue: 'every agent line reaches the channel' }) - : t('connectors.rowAttention', { defaultValue: 'escalations reach the channel' })) + ? (mirror ? mirrorLabel : attentionLabel) : t('connectors.rowRelayOff', { defaultValue: 'messages stay in the pod' }); return { - action: (isTelegram || isSlack) ? 'manage' : null, + action: (isTelegram || isSlack || discordRelayReady(connector)) ? 'manage' : null, actionLabel: t('connectors.manage', { defaultValue: 'Manage' }), detail: '', mark: { From e74ab43a9dfeb1b0f2fcdce3a791525e6a29ced1 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Wed, 4 Nov 2026 17:01:02 -0800 Subject: [PATCH 2/3] fix(discord): relay stops when the installer leaves the pod; review follow-ups (#3465) - Send path and the live-relay refusal require the row's creator to still be a listed pod member; a pod that can't be read sends nothing. - Connectors: the Relay toggle and Manage need the viewer to be the row's creator; the "add the bot again" note shows only to the creator. - 50027 (revoked webhook token) classified like 10015. - escapeDiscordMarkdown escapes [ ] ( ); pod and agent names in the prefix go through the @everyone/@here defuse. - Per-webhook send queue capped at 20; overflow dropped with the cooldown's log line. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MDs9DPFVAooJnqCSwrdxVw --- .../routes/discordBindingContainment.test.js | 11 ++++ .../unit/services/chatRelayCore.test.js | 12 +++- .../connectorDeliveryFailureService.test.js | 5 ++ .../discordBridgeService.outbound.test.js | 58 +++++++++++++++++- .../services/discordBridgeService.test.js | 15 ++++- backend/routes/integrations.ts | 11 +++- backend/services/chatRelay/chatRelayCore.ts | 2 + .../connectorDeliveryFailureService.ts | 3 +- backend/services/discordBridgeService.ts | 60 ++++++++++++++----- .../ConnectorsPage.channels.test.tsx | 27 ++++++++- frontend/src/v2/connectors/ChannelPanel.tsx | 10 +++- frontend/src/v2/connectors/channelModel.ts | 23 +++++-- .../src/v2/connectors/useChannelConnectors.ts | 6 +- 13 files changed, 209 insertions(+), 34 deletions(-) diff --git a/backend/__tests__/unit/routes/discordBindingContainment.test.js b/backend/__tests__/unit/routes/discordBindingContainment.test.js index eb97261d8..8ab190229 100644 --- a/backend/__tests__/unit/routes/discordBindingContainment.test.js +++ b/backend/__tests__/unit/routes/discordBindingContainment.test.js @@ -919,6 +919,17 @@ describe('Discord binding containment (TASK-123 a)', () => { }); }); + it('refuses an installer who has left the pod (review finding 1)', async () => { + Integration.findById.mockResolvedValue(provenRow()); + Pod.findById.mockResolvedValue({ _id: 'pod-1', members: ['user-3'], createdBy: 'user-3' }); + + const res = await patch({ config: { liveRelay: true } }); + + expect(res.status).toBe(400); + expect(res.body.message).toMatch(/member of this pod/); + expect(Integration.findByIdAndUpdate).not.toHaveBeenCalled(); + }); + it('refuses the pod creator who did not add the bot: their id would author the installer’s lines', async () => { // MEMBER created the pod (canDeleteIntegration passes) but user-3 added the bot. Integration.findById.mockResolvedValue(provenRow({ createdBy: { toString: () => 'user-3' } })); diff --git a/backend/__tests__/unit/services/chatRelayCore.test.js b/backend/__tests__/unit/services/chatRelayCore.test.js index 7b5137ba1..83f6b05ec 100644 --- a/backend/__tests__/unit/services/chatRelayCore.test.js +++ b/backend/__tests__/unit/services/chatRelayCore.test.js @@ -376,9 +376,15 @@ describe('inbound: only a proven sender reaches the write (F131)', () => { }); test('Discord’s rule is its own declaration: a proven installer, not a chat type', () => { - expect(liveRelayRefusalFor('discord', { config: {}, createdBy: 'u' }, 'u')).toMatch(/Discord account/); - expect(liveRelayRefusalFor('discord', { config: { discordUserId: '345678901234567890' }, createdBy: 'u' }, 'u')).toBeNull(); - expect(liveRelayRefusalFor('discord', { config: { discordUserId: '345678901234567890' }, createdBy: 'u' }, 'v')).toMatch(/Only the person/); + const pod = { members: ['u', 'v'] }; + const proven = { discordUserId: '345678901234567890' }; + expect(liveRelayRefusalFor('discord', { config: {}, createdBy: 'u', pod }, 'u')).toMatch(/Discord account/); + expect(liveRelayRefusalFor('discord', { config: proven, createdBy: 'u', pod }, 'u')).toBeNull(); + expect(liveRelayRefusalFor('discord', { config: proven, createdBy: 'u', pod }, 'v')).toMatch(/Only the person/); + // Review finding 1: the installer must still be a listed member of the pod. + expect(liveRelayRefusalFor('discord', { config: proven, createdBy: 'u', pod: { members: ['v'] } }, 'u')) + .toMatch(/member of this pod/); + expect(liveRelayRefusalFor('discord', { config: proven, createdBy: 'u' }, 'u')).toMatch(/member of this pod/); }); test('an inherited property name is not a platform', () => { diff --git a/backend/__tests__/unit/services/connectorDeliveryFailureService.test.js b/backend/__tests__/unit/services/connectorDeliveryFailureService.test.js index e195f18ee..fb1370e5f 100644 --- a/backend/__tests__/unit/services/connectorDeliveryFailureService.test.js +++ b/backend/__tests__/unit/services/connectorDeliveryFailureService.test.js @@ -278,6 +278,11 @@ describe('Discord classification (F132)', () => { .toBe(deliveryFailures.DISCORD_NO_PERMISSION_REASON); }); + it('reads a revoked webhook token (401, 50027) as the webhook being gone', () => { + expect(deliveryFailures.classifyDiscordDeliveryFailure(discordFailure(401, 50027))) + .toBe(deliveryFailures.DISCORD_WEBHOOK_GONE_REASON); + }); + it('never flips for a rate limit, a server error, content, or a success', () => { [ discordFailure(429, 0, { retryAfterMs: 1200 }), diff --git a/backend/__tests__/unit/services/discordBridgeService.outbound.test.js b/backend/__tests__/unit/services/discordBridgeService.outbound.test.js index 52d27c1ec..3179604c5 100644 --- a/backend/__tests__/unit/services/discordBridgeService.outbound.test.js +++ b/backend/__tests__/unit/services/discordBridgeService.outbound.test.js @@ -87,7 +87,7 @@ beforeEach(() => { discordWebhookPacing.reset(); sleeps = []; discordWebhookPacing.sleep = jest.fn(async (ms) => { sleeps.push(ms); }); - Pod.findById.mockReturnValue({ select: () => ({ lean: async () => ({ name: 'Launch' }) }) }); + Pod.findById.mockReturnValue({ select: () => ({ lean: async () => ({ name: 'Launch', members: [CREATOR] }) }) }); connectorSecrets.get.mockResolvedValue(WEBHOOK_URL); axios.post.mockResolvedValue(ok()); Integration.findByIdAndUpdate.mockResolvedValue({}); @@ -141,6 +141,27 @@ describe('who it sends for', () => { expect(axios.post).not.toHaveBeenCalled(); }); + // Review finding 1: the channel is shared and the installer controls it, so + // an installer who left the pod stops receiving the pod's agent lines there. + test('sends nothing once the installer is no longer a listed pod member', async () => { + Pod.findById.mockReturnValue({ + select: () => ({ lean: async () => ({ name: 'Launch', members: ['64b7f0c2a1b2c3d4e5f60799'] }) }), + }); + await relay(discordRow()); + expect(axios.post).not.toHaveBeenCalled(); + }); + + test('sends nothing when the pod cannot be read (fails closed)', async () => { + Pod.findById.mockReturnValue({ select: () => ({ lean: async () => { throw new Error('mongo down'); } }) }); + await relay(discordRow()); + expect(axios.post).not.toHaveBeenCalled(); + }); + + test('a current member installer is unchanged', async () => { + await relay(discordRow()); + expect(axios.post).toHaveBeenCalledTimes(1); + }); + test('sends nothing when the stored webhook URL is not a Discord webhook', async () => { connectorSecrets.get.mockResolvedValue('https://example.com/collect'); await relay(discordRow()); @@ -180,6 +201,19 @@ describe('content', () => { expect(sentBody().content.startsWith('[Launch] **\\*\\*Boss\\*\\***: ')).toBe(true); }); + // Review finding 4: a pod name is user-chosen; it must not become a masked + // link or a mass mention in the prefix. + test('the pod name cannot forge a link or a mass mention in the prefix', async () => { + Pod.findById.mockReturnValue({ + select: () => ({ lean: async () => ({ name: 'x](https://evil) [y @everyone', members: [CREATOR] }) }), + }); + await relay(discordRow()); + const { content } = sentBody(); + expect(content.startsWith('[x\\]\\(https://evil\\) \\[y @\u200beveryone] **Kai**: ')).toBe(true); + expect(content).not.toMatch(/(^|[^\\])\]\(https:\/\/evil/); + expect(content).not.toMatch(/@everyone/); + }); + test('a decision card renders inside the limit and links back to the pod', async () => { await relay(discordRow(), { card: { @@ -235,6 +269,28 @@ describe('failures', () => { expect(Integration.findOneAndUpdate).not.toHaveBeenCalled(); }); + // Review finding 5: a burst cannot grow the lane's queue without bound. + test('a lane holds at most maxQueueDepth lines; the rest are dropped and logged', async () => { + let release; + axios.post.mockImplementationOnce(() => new Promise((resolve) => { release = () => resolve(ok()); })); + const warn = jest.spyOn(console, 'warn').mockImplementation(() => {}); + try { + const row = discordRow(); + const depth = discordWebhookPacing.maxQueueDepth; + const sends = Array.from({ length: depth + 5 }, (_unused, index) => relay(row, { content: `line ${index}` })); + await new Promise((resolve) => { setImmediate(resolve); }); + while (!release) await new Promise((resolve) => { setImmediate(resolve); }); // eslint-disable-line no-await-in-loop + release(); + await Promise.all(sends); + expect(axios.post).toHaveBeenCalledTimes(depth); + const dropped = warn.mock.calls.filter(([line]) => /line dropped/.test(String(line))); + expect(dropped).toHaveLength(5); + expect(String(dropped[0][0])).toMatch(/rate-limited/); + } finally { + warn.mockRestore(); + } + }); + test('back-to-back lines to one webhook are paced at about 5 per 2 seconds', async () => { const row = discordRow(); await Promise.all([relay(row, { content: 'one' }), relay(row, { content: 'two' }), relay(row, { content: 'three' })]); diff --git a/backend/__tests__/unit/services/discordBridgeService.test.js b/backend/__tests__/unit/services/discordBridgeService.test.js index 1b61c4cc5..5b2e1abd6 100644 --- a/backend/__tests__/unit/services/discordBridgeService.test.js +++ b/backend/__tests__/unit/services/discordBridgeService.test.js @@ -388,9 +388,22 @@ describe('Discord inbound live relay (F131)', () => { }); test('refuses anyone but the installer, whose id linkedUserId would be stamped from', () => { - const proven = { config: { discordUserId: INSTALLER_DISCORD }, createdBy: { toString: () => INSTALLER } }; + const proven = { + config: { discordUserId: INSTALLER_DISCORD }, + createdBy: { toString: () => INSTALLER }, + pod: { members: [INSTALLER, 'pod-creator'] }, + }; expect(discordInboundAdapter.liveRelayRefusal(proven, 'pod-creator')).toMatch(/Only the person who added the bot/); expect(discordInboundAdapter.liveRelayRefusal(proven, INSTALLER)).toBeNull(); }); + + test('refuses an installer who is no longer a listed member of the pod', () => { + const left = { + config: { discordUserId: INSTALLER_DISCORD }, + createdBy: { toString: () => INSTALLER }, + pod: { members: ['someone-else'] }, + }; + expect(discordInboundAdapter.liveRelayRefusal(left, INSTALLER)).toMatch(/member of this pod/); + }); }); }); diff --git a/backend/routes/integrations.ts b/backend/routes/integrations.ts index 5b255dee0..afc69108d 100644 --- a/backend/routes/integrations.ts +++ b/backend/routes/integrations.ts @@ -575,7 +575,7 @@ router.post('/', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: Res // as it will be stored, before the row or any webhook exists. if (nextConfig.liveRelay === true) { const refusal = liveRelayRefusalFor( - type, { config: nextConfig, createdBy: req.user?.id }, String(req.user?.id || ''), + type, { config: nextConfig, createdBy: req.user?.id, pod: targetPod }, String(req.user?.id || ''), ); if (refusal) return res.status(400).json({ message: refusal }); } @@ -980,10 +980,15 @@ router.patch('/:id', writeIntegrationsRateLimit, auth, async (req: AuthReq, res: // Relay authors inbound as linkedUserId, which is stamped from this caller. // Each platform declares when that is honest (its inbound adapter's // liveRelayRefusal): a 1:1 private chat for Telegram and Slack (#1289 - // inbound, F2 outbound), the proven installer for Discord (F131). + // inbound, F2 outbound), the proven installer for Discord (F131), who must + // still be a listed member of the pod the row serves (F132 review). const refusal = liveRelayRefusalFor( integration.type, - { config: nextConfig, createdBy: integration.createdBy }, + { + config: nextConfig, + createdBy: integration.createdBy, + pod: integration.podId ? await Pod.findById(String(integration.podId)) : null, + }, requesterId, ); if (refusal) return res.status(400).json({ message: refusal }); diff --git a/backend/services/chatRelay/chatRelayCore.ts b/backend/services/chatRelay/chatRelayCore.ts index b57488c1d..4937fbefc 100644 --- a/backend/services/chatRelay/chatRelayCore.ts +++ b/backend/services/chatRelay/chatRelayCore.ts @@ -324,6 +324,8 @@ export const relayAgentMessageThroughAdapter = async | null; createdBy?: unknown; + /** The pod the row serves, for a platform whose rule reads its members (Discord). */ + pod?: { members?: unknown[] } | null; } /** diff --git a/backend/services/connectorDeliveryFailureService.ts b/backend/services/connectorDeliveryFailureService.ts index 40fe93825..1a8bc463b 100644 --- a/backend/services/connectorDeliveryFailureService.ts +++ b/backend/services/connectorDeliveryFailureService.ts @@ -38,7 +38,7 @@ const PERMANENT_SLACK_ERRORS: { [error: string]: string } = { }; // Discord's webhook send (F132) reports failure as an HTTP status carrying a -// JSON error `code`. These four mean the channel cannot be posted to until a +// JSON error `code`. These codes mean the channel cannot be posted to until a // person acts; everything else (429, 5xx, a 400 on content) is transient or // content and says nothing about the binding. const DISCORD_CHANNEL_GONE_REASON = 'Discord stopped delivering: this channel no longer exists.'; @@ -49,6 +49,7 @@ const DISCORD_NO_PERMISSION_REASON = 'Discord stopped delivering: the bot is not const PERMANENT_DISCORD_CODES: { [code: number]: string } = { 10003: DISCORD_CHANNEL_GONE_REASON, // Unknown Channel 10015: DISCORD_WEBHOOK_GONE_REASON, // Unknown Webhook + 50027: DISCORD_WEBHOOK_GONE_REASON, // Invalid Webhook Token (401): the webhook's token was revoked 50001: DISCORD_NO_ACCESS_REASON, // Missing Access 50013: DISCORD_NO_PERMISSION_REASON, // Missing Permissions }; diff --git a/backend/services/discordBridgeService.ts b/backend/services/discordBridgeService.ts index 9631dcd00..55ab8d5fc 100644 --- a/backend/services/discordBridgeService.ts +++ b/backend/services/discordBridgeService.ts @@ -20,7 +20,7 @@ const Integration = require('../models/Integration'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const axios = require('axios'); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require -const { isGatedPodTarget } = require('./connectorRelayPolicy'); +const { isGatedPodTarget, isListedPodMember } = require('./connectorRelayPolicy'); import { resolveDiscordWebhookUrl } from '../utils/discordWebhookUrl'; import { claimInboundMessage, @@ -117,6 +117,8 @@ const isInboundRelayableIntegration = ( const LIVE_RELAY_NEEDS_PROOF = 'Live relay on a Discord channel needs the Discord account of the person who ' + 'added the bot, which the install proves. Add the bot again from Commonly to link it.'; +const LIVE_RELAY_MEMBERS_ONLY = 'Live relay on a Discord channel needs the person who added the bot to still be ' + + 'a member of this pod: the channel is theirs, and the pod\'s agent messages would go to it.'; const LIVE_RELAY_INSTALLER_ONLY = 'Only the person who added the bot can turn on live relay for this channel: ' + 'it posts their own Discord messages into the pod as them, and the pod\'s agent messages into the channel.'; @@ -136,10 +138,13 @@ export const discordInboundAdapter: InboundRelayAdapter { if (!row.config?.discordUserId) return LIVE_RELAY_NEEDS_PROOF; if (!requesterId || idOf(row.createdBy) !== String(requesterId)) return LIVE_RELAY_INSTALLER_ONLY; + if (!row.pod || !isListedPodMember(row.pod, requesterId)) return LIVE_RELAY_MEMBERS_ONLY; return null; }, }; @@ -295,8 +300,9 @@ const findLiveIntegration = async (podId: string): Promise value.replace(/([\\*_~`|>])/g, '\\$1'); +// prefix, and brackets and parentheses would let `x](https://evil) [y` close +// the tag and open a masked link; a backslash makes each one literal. +const escapeDiscordMarkdown = (value: string): string => value.replace(/([\\*_~`|>[\]()])/g, '\\$1'); // `allowed_mentions` already stops every ping. This also keeps the words from // reading as one: a zero-width space after the @ renders the same and is no @@ -312,21 +318,24 @@ const fitWithin = (value: string, limit: number): string => { return `${cut}…`; }; -const podNameForTag = async (podId: string): Promise => { +// The pod read the send needs twice: its name for the tag, and its listed +// members, because the installer must still be one (F132 review). A failed +// read returns null and the send fails closed — the line stays in the pod. +const podForSend = async (podId: string): Promise<{ name?: string; members?: unknown[] } | null> => { try { // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const Pod = require('../models/Pod'); - const pod = await Pod.findById(podId).select('name').lean(); - return String(pod?.name || podId); + return await Pod.findById(podId).select('name members').lean(); } catch (error) { - console.warn(`${LOG} pod name lookup failed, tagging with the id:`, (error as Error).message); - return podId; + console.warn(`${LOG} pod lookup failed, line not sent:`, (error as Error).message); + return null; } }; /** `[Pod] **Agent**: text`, the shape the Slack and Telegram lines take, in 2000 characters. */ export const renderDiscordAgentLine = (opts: { podName: string; name: string; content: string }): string => { - const prefix = `[${escapeDiscordMarkdown(opts.podName)}] **${escapeDiscordMarkdown(opts.name)}**: `; + const tag = defuseMassMentions(escapeDiscordMarkdown(opts.podName)); + const prefix = `[${tag}] **${defuseMassMentions(escapeDiscordMarkdown(opts.name))}**: `; return fitWithin(prefix + defuseMassMentions(String(opts.content || '')), DISCORD_CONTENT_LIMIT); }; @@ -363,8 +372,9 @@ export interface DiscordDeliveryResult { code?: number; messageId?: string; retryAfterMs?: number; - /** Not sent: the webhook's lane is still cooling down from a 429. */ + /** Not sent: the webhook's lane is still cooling down from a 429, or its queue is full. */ rateLimited?: boolean; + queueFull?: boolean; } /** @@ -376,10 +386,13 @@ export interface DiscordDeliveryResult { * Per process: each backend replica paces its own sends, and Discord's 429 is * the backstop between them. */ -interface WebhookLane { tail: Promise; nextAt: number; coolUntil: number } +interface WebhookLane { tail: Promise; nextAt: number; coolUntil: number; depth: number } export const discordWebhookPacing = { minIntervalMs: 400, maxRetryWaitMs: 5_000, + // Lines waiting on one webhook, the one in flight included. A burst beyond + // this is dropped and logged rather than queued without bound. + maxQueueDepth: 20, defaultRetryAfterMs: 1_000, now: (): number => Date.now(), sleep: (ms: number): Promise => new Promise((resolve) => { setTimeout(resolve, ms); }), @@ -423,10 +436,16 @@ const sendThroughLane = ( const pacing = discordWebhookPacing; let lane = pacing.lanes.get(laneKey); if (!lane) { - lane = { tail: Promise.resolve(), nextAt: 0, coolUntil: 0 }; + lane = { tail: Promise.resolve(), nextAt: 0, coolUntil: 0, depth: 0 }; pacing.lanes.set(laneKey, lane); } const current = lane; + if (current.depth >= pacing.maxQueueDepth) { + return Promise.resolve({ + provider: 'discord', delivered: false, httpStatus: 429, rateLimited: true, queueFull: true, + }); + } + current.depth += 1; const run = async (): Promise => { if (pacing.now() < current.coolUntil) { return { @@ -451,7 +470,7 @@ const sendThroughLane = ( } return result; }; - const next = current.tail.then(run); + const next = current.tail.then(run).finally(() => { current.depth -= 1; }); current.tail = next.catch(() => undefined); return next; }; @@ -497,10 +516,18 @@ export const discordRelayAdapter: ChatRelayAdapter = { console.warn(`${LOG} outbound skipped integration=${String(integration._id)}: no Discord webhook on the row`); return null; } + const pod = await podForSend(podId); + if (!pod) return null; + // The channel is the installer's and shared with their guild: once they + // leave the pod, its agent lines stop going there (F132 review). + if (!isListedPodMember(pod, integration.createdBy)) { + console.warn(`${LOG} outbound skipped integration=${String(integration._id)}: the installer left the pod`); + return null; + } const name = displayName || agentUsername; const text = card ? renderDiscordDecisionCard({ card, name, link }) - : renderDiscordAgentLine({ podName: await podNameForTag(podId), name, content }); + : renderDiscordAgentLine({ podName: String(pod.name || podId), name, content }); const result = await sendThroughLane(String(integration._id), withWait(webhookUrl), { content: text, // Agent text goes to everyone in the guild: no @everyone, @here, role or @@ -512,7 +539,8 @@ export const discordRelayAdapter: ChatRelayAdapter = { deliveryFailureError: (result) => { const r = (result || {}) as DiscordDeliveryResult; if (r.rateLimited) { - return new Error(`webhook lane cooling down after a 429 (${r.retryAfterMs}ms left); line dropped`); + const why = r.queueFull ? 'queue full' : `cooling down after a 429, ${r.retryAfterMs}ms left`; + return new Error(`webhook lane rate-limited (${why}); line dropped`); } const code = r.code !== undefined ? ` code ${r.code}` : ''; return new Error(`webhook send failed: HTTP ${String(r.httpStatus)}${code}`); diff --git a/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx b/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx index 1602ac284..c3cb52ee7 100644 --- a/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx +++ b/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx @@ -11,7 +11,7 @@ import en from '../../i18n/locales/en.json'; import i18n, { i18nReady } from '../../i18n'; import { axios, authValue, mockApi, renderPage, renderWithRoutes, pageReady, tileNode, findTile, clickTile, - openDrawer, openPopover, findDrawer, expandOff, + openDrawer, openPopover, findDrawer, expandOff, adminAuth, } from '../__testkit__/connectorsKit'; jest.mock('axios', () => require('../__testkit__/axiosMock').axiosFactory()); @@ -1715,7 +1715,7 @@ const ConnectorsPageForRoutes = () => { // the server stamps `discordUserId` on the creator's row and withholds it from // everyone else's view. The copy says who reads the channel. describe('Connectors page: Discord live relay (F132)', () => { - const discordRow = (config = {}) => ({ + const discordRow = (config = {}, over = {}) => ({ _id: 'i-discord', type: 'discord', status: 'connected', @@ -1723,6 +1723,8 @@ describe('Connectors page: Discord live relay (F132)', () => { updatedAt: new Date(Date.now() - 3_600_000).toISOString(), config: { serverId: '111111111111111111', channelId: '222222222222222222', ...config }, podId: { _id: 'p2', name: 'Ops' }, + createdBy: 'u1', + ...over, }); beforeEach(() => { @@ -1730,6 +1732,27 @@ describe('Connectors page: Discord live relay (F132)', () => { window.history.replaceState({}, '', '/v2/connectors'); }); + // Review finding 2: the server withholds `discordUserId` from everyone but the + // creator, so its absence on someone else's row says nothing about proof. + it('shows a viewer who is not the creator neither the switch nor the add-the-bot-again note', async () => { + mockGets([discordRow({ liveRelay: true, relayAllAgentMessages: true }, { createdBy: 'u2' })]); + renderPage(); + const drawer = await openDrawer('Discord'); + expect(within(drawer).getByText( + 'Every agent message is posted in the Discord channel, where everyone in the server can read it.', + )).toBeInTheDocument(); + expect(within(drawer).queryByRole('checkbox', { name: 'Relay' })).toBeNull(); + expect(within(drawer).queryByText('Add the bot again from Commonly to turn on relay for this channel.')).toBeNull(); + }); + + it('shows an admin who is not the creator no switch, even though the row reaches them whole', async () => { + mockGets([discordRow({ discordUserId: '345678901234567890', liveRelay: false }, { createdBy: { _id: 'u2' } })]); + renderPage({ auth: adminAuth() }); + const drawer = await openDrawer('Discord'); + expect(within(drawer).queryByRole('checkbox', { name: 'Relay' })).toBeNull(); + expect(within(drawer).queryByText('Add the bot again from Commonly to turn on relay for this channel.')).toBeNull(); + }); + afterEach(() => jest.restoreAllMocks()); it('offers the installer the relay switch, and turning it on PATCHes liveRelay', async () => { diff --git a/frontend/src/v2/connectors/ChannelPanel.tsx b/frontend/src/v2/connectors/ChannelPanel.tsx index 7ce5df3fd..8ee1d294c 100644 --- a/frontend/src/v2/connectors/ChannelPanel.tsx +++ b/frontend/src/v2/connectors/ChannelPanel.tsx @@ -12,6 +12,7 @@ import { connectorRelativeTimeOptions, botMembers, discordRelayReady, + viewerCreatedConnector, } from './channelModel'; import type { Connector, @@ -29,6 +30,7 @@ interface Props { const ChannelPanel: React.FC = ({ m, item: panelItem }) => { const { t } = useTranslation(); const { + viewerId, pods, busyId, confirmRemove, @@ -235,7 +237,11 @@ const ChannelPanel: React.FC = ({ m, item: panelItem }) => { // F132: a Discord row shows the card always (its hint says why a switch is // missing) and the switch only when its installer is proven. const supportsRelayControls = isTelegram || isSlack || isDiscord; - const relaySwitchOffered = !isDiscord || discordRelayReady(connector); + const relaySwitchOffered = !isDiscord || discordRelayReady(connector, viewerId); + // Only the creator is handed `discordUserId`, so only for them does its + // absence mean the bot-add proof is missing. + const discordNeedsProof = isDiscord && viewerCreatedConnector(connector, viewerId) + && !connector.config?.discordUserId; const mirror = connector.config?.relayAllAgentMessages === true; const pendingBind = connector.config?.pendingBind; const activePod = podList.find((pod) => pod._id === connectorPodId(connector)); @@ -353,7 +359,7 @@ const ChannelPanel: React.FC = ({ m, item: panelItem }) => { {isDiscord && relaySwitchOffered && connector.config?.liveRelay && (

{t('connectors.discordInboundNote', { defaultValue: 'Your own messages in the channel reach the pod as you.' })}

)} - {isDiscord && !relaySwitchOffered && ( + {discordNeedsProof && (

{t('connectors.discordNeedsProof', { defaultValue: 'Add the bot again from Commonly to turn on relay for this channel.' })}

diff --git a/frontend/src/v2/connectors/channelModel.ts b/frontend/src/v2/connectors/channelModel.ts index b8ec93fac..e3130c529 100644 --- a/frontend/src/v2/connectors/channelModel.ts +++ b/frontend/src/v2/connectors/channelModel.ts @@ -56,6 +56,8 @@ export interface Connector { updatedAt?: string; config?: ConnectorConfig; podId?: { _id: string; name?: string } | string | null; + // An id, or a populated user on the pod-scoped read. + createdBy?: string | { _id?: string } | null; } export interface CatalogInstallation { @@ -264,12 +266,25 @@ export const SLACK_REFUSALS: Record { + const createdBy = connector?.createdBy; + if (!createdBy) return ''; + return typeof createdBy === 'object' ? String(createdBy._id || '') : String(createdBy); +}; + // A Discord row relays to a shared guild channel, and only the installer the // bot-add proved may switch that on (backend discordInboundAdapter -// .liveRelayRefusal). A row without the proof gets no switch, never one the -// server would refuse. -export const discordRelayReady = (connector: Connector | null | undefined): boolean => ( - connector?.type === 'discord' && Boolean(connector.config?.discordUserId) +// .liveRelayRefusal). The viewer must be the row's creator: an admin is handed +// the row whole, `discordUserId` included, and the server would still refuse +// them. A row without the proof gets no switch, never one the server refuses. +export const viewerCreatedConnector = (connector: Connector | null | undefined, viewerId: string): boolean => ( + Boolean(viewerId) && connectorCreatorId(connector) === viewerId +); + +export const discordRelayReady = (connector: Connector | null | undefined, viewerId: string): boolean => ( + connector?.type === 'discord' + && Boolean(connector.config?.discordUserId) + && viewerCreatedConnector(connector, viewerId) ); export const connectorPodId = (connector: Connector | null | undefined): string | null => { diff --git a/frontend/src/v2/connectors/useChannelConnectors.ts b/frontend/src/v2/connectors/useChannelConnectors.ts index 875638645..04ce4fe20 100644 --- a/frontend/src/v2/connectors/useChannelConnectors.ts +++ b/frontend/src/v2/connectors/useChannelConnectors.ts @@ -4,6 +4,7 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; import { useTranslation } from 'react-i18next'; import { useV2Api } from '../hooks/useV2Api'; +import { useAuth } from '../../context/AuthContext'; import { useRelativeNow } from '../hooks/useRelativeNow'; import { V2Pod } from '../hooks/useV2Pods'; import type { MarkName } from '../icons/glyphs'; @@ -47,6 +48,8 @@ import type { export const useChannelConnectors = () => { const { t, i18n } = useTranslation(); const api = useV2Api(); + const { currentUser } = useAuth(); + const viewerId = currentUser?._id ? String(currentUser._id) : ''; const [connectors, setConnectors] = useState([]); const [catalog, setCatalog] = useState(null); const [hostedCatalog, setHostedCatalog] = useState(null); @@ -741,7 +744,7 @@ export const useChannelConnectors = () => { ? (mirror ? mirrorLabel : attentionLabel) : t('connectors.rowRelayOff', { defaultValue: 'messages stay in the pod' }); return { - action: (isTelegram || isSlack || discordRelayReady(connector)) ? 'manage' : null, + action: (isTelegram || isSlack || discordRelayReady(connector, viewerId)) ? 'manage' : null, actionLabel: t('connectors.manage', { defaultValue: 'Manage' }), detail: '', mark: { @@ -1072,6 +1075,7 @@ export const useChannelConnectors = () => { const busyFor = (item: ListItem): boolean => busyId === item.key || (item.connector !== null && busyId === item.connector._id); return { + viewerId, connectors, setConnectors, catalog, From 7721bed982fbc1ae411883a076ef277d2be439cb Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Wed, 4 Nov 2026 17:09:14 -0800 Subject: [PATCH 3/3] fix(discord): relay only from a connected row; Remove deletes the webhook; CodeRabbit follow-ups (#3465) - A Discord row relays only at status 'connected' (query and predicate), so a disconnected row whose webhook delete failed no longer posts. - Remove is back for connected Discord rows and goes to DELETE /api/integrations/:id, which also deletes the channel webhook. - ChannelVerdict accepts provider 'discord', so a Discord decision-card verdict validates. - Attention-mode copy (en, zh-CN) says the lead agent's lines also reach the channel; the contract doc says outbound uses webhookUrlRef, not a bot token, and lists 50027. - Tests: webhook allowlist bypass shapes; the real cache refresh keeps config.discordUserId in its projection. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MDs9DPFVAooJnqCSwrdxVw --- .../unit/models/ChannelVerdict.test.js | 9 +++++++ .../discordBridgeService.outbound.test.js | 23 ++++++++++++++-- .../services/discordBridgeService.test.js | 26 +++++++++++++++++++ backend/models/ChannelVerdict.ts | 3 ++- backend/services/discordBridgeService.ts | 6 +++-- docs/integrations/INTEGRATION_CONTRACT.md | 7 ++--- frontend/src/i18n/locales/en.json | 2 +- frontend/src/i18n/locales/zh-CN.json | 2 +- .../ConnectorsPage.channels.test.tsx | 24 +++++++++++++++++ frontend/src/v2/connectors/ChannelPanel.tsx | 3 +-- .../src/v2/connectors/useChannelConnectors.ts | 9 ++++++- 11 files changed, 101 insertions(+), 13 deletions(-) diff --git a/backend/__tests__/unit/models/ChannelVerdict.test.js b/backend/__tests__/unit/models/ChannelVerdict.test.js index e4e0141c8..5a3754778 100644 --- a/backend/__tests__/unit/models/ChannelVerdict.test.js +++ b/backend/__tests__/unit/models/ChannelVerdict.test.js @@ -40,6 +40,15 @@ describe('ChannelVerdict', () => { ])); }); + // CodeRabbit (PR #3465): the Discord relay records decision-card verdicts + // with provider 'discord'; the enum must accept it on a real round trip. + it('stores and validates a Discord verdict', async () => { + const created = await ChannelVerdict.create(row({ provider: 'discord' })); + const stored = await ChannelVerdict.findById(created._id); + expect(stored.provider).toBe('discord'); + await expect(stored.validate()).resolves.toBeUndefined(); + }); + it('accepts a lazily-linked decision id and a later channel receipt', async () => { const verdict = new ChannelVerdict(row({ provider: 'slack', diff --git a/backend/__tests__/unit/services/discordBridgeService.outbound.test.js b/backend/__tests__/unit/services/discordBridgeService.outbound.test.js index 3179604c5..2b8cfd9d6 100644 --- a/backend/__tests__/unit/services/discordBridgeService.outbound.test.js +++ b/backend/__tests__/unit/services/discordBridgeService.outbound.test.js @@ -133,6 +133,10 @@ describe('who it sends for', () => { ['live relay off', discordRow({ liveRelay: false })], ['an admin pause', discordRow({ adminPause: { reason: 'x' } })], ['an error row', discordRow({}, { status: 'error' })], + // CodeRabbit 4475336541: disconnect sets 'disconnected' and leaves isActive + // and liveRelay alone, so status itself must be 'connected'. + ['a disconnected row', discordRow({}, { status: 'disconnected' })], + ['a pending row', discordRow({}, { status: 'pending' })], ['no proven installer', discordRow({ discordUserId: undefined })], ['live relay switched on by someone other than the creator', discordRow({ linkedUserId: '64b7f0c2a1b2c3d4e5f60799' })], ['a row for another pod', discordRow({}, { podId: '64b7f0c2a1b2c3d4e5f60700' })], @@ -162,8 +166,23 @@ describe('who it sends for', () => { expect(axios.post).toHaveBeenCalledTimes(1); }); - test('sends nothing when the stored webhook URL is not a Discord webhook', async () => { - connectorSecrets.get.mockResolvedValue('https://example.com/collect'); + test('the fallback lookup selects only a connected row', async () => { + Integration.findOne.mockReturnValue({ lean: async () => null }); + await relayAgentMessageThroughAdapter(discordRelayAdapter, { + podId: POD, agentUsername: 'kai', displayName: 'Kai', content: 'x', podMessageId: 'm', + }); + expect(Integration.findOne).toHaveBeenCalledWith(expect.objectContaining({ status: 'connected' })); + }); + + test.each([ + ['a foreign host', 'https://example.com/collect'], + ['plain http', 'http://discord.com/api/webhooks/1/x'], + ['a look-alike host', 'https://discord.com.evil.example/api/webhooks/1/x'], + ['userinfo before another host', 'https://discord.com@evil.example/api/webhooks/1/x'], + ['a query string', 'https://discord.com/api/webhooks/1/x?foo=bar'], + ['a non-numeric webhook id', 'https://discord.com/api/webhooks/abc/x'], + ])('sends nothing when the stored webhook URL is %s', async (_label, url) => { + connectorSecrets.get.mockResolvedValue(url); await relay(discordRow()); expect(axios.post).not.toHaveBeenCalled(); }); diff --git a/backend/__tests__/unit/services/discordBridgeService.test.js b/backend/__tests__/unit/services/discordBridgeService.test.js index 5b2e1abd6..752bed778 100644 --- a/backend/__tests__/unit/services/discordBridgeService.test.js +++ b/backend/__tests__/unit/services/discordBridgeService.test.js @@ -346,6 +346,32 @@ describe('Discord inbound live relay (F131)', () => { expect(Integration.findByIdAndUpdate).toHaveBeenCalledTimes(1); }); + // CodeRabbit nitpick: the pre-filter is only as good as the cache refresh's + // projection. This runs the real refresh over a find() that drops every + // field the projection does not name. + test('the real cache refresh carries discordUserId, so the installer still relays', async () => { + discordGatewayService.ensureChannelCache.mockRestore(); + discordGatewayService.cacheExpiresAt = 0; + const stored = row(); + Integration.find.mockReturnValue({ + select: (projection) => ({ + lean: async () => { + const fields = String(projection).split(/\s+/); + const config = Object.fromEntries(Object.entries(stored.config) + .filter(([key]) => fields.includes(`config.${key}`))); + return [{ _id: 'integration-1', config }]; + }, + }), + }); + try { + await discordGatewayService.handleMessageCreate(discordMessage()); + expect(Integration.findById).toHaveBeenCalledWith('integration-1'); + expect(writes()).toHaveLength(1); + } finally { + jest.spyOn(discordGatewayService, 'ensureChannelCache').mockResolvedValue(); + } + }); + test('does not read the row when the cache holds no proven installer', async () => { discordGatewayService.channelCache.set(`${GUILD}:${CHANNEL}`, [target({ discordUserId: undefined })]); diff --git a/backend/models/ChannelVerdict.ts b/backend/models/ChannelVerdict.ts index 91a4e1b17..d343e3b54 100644 --- a/backend/models/ChannelVerdict.ts +++ b/backend/models/ChannelVerdict.ts @@ -6,7 +6,8 @@ import mongoose, { Document, Model, Schema, Types } from 'mongoose'; * administrator action): it records what a particular channel did with a * workspace event. */ -export const CHANNEL_VERDICT_PROVIDERS = ['telegram', 'slack'] as const; +// Every chat-relay adapter that records card verdicts (chatRelayCore), Discord since F132. +export const CHANNEL_VERDICT_PROVIDERS = ['telegram', 'slack', 'discord'] as const; export type ChannelVerdictProvider = typeof CHANNEL_VERDICT_PROVIDERS[number]; export const CHANNEL_VERDICTS = ['interrupt', 'digest', 'hold'] as const; diff --git a/backend/services/discordBridgeService.ts b/backend/services/discordBridgeService.ts index 55ab8d5fc..810bba34f 100644 --- a/backend/services/discordBridgeService.ts +++ b/backend/services/discordBridgeService.ts @@ -273,11 +273,13 @@ const DISCORD_WEBHOOK_URL = new RegExp( // account the install proved (`linkedUserId` is stamped from that caller, and // liveRelayRefusal admits only the creator holding a `discordUserId`). A row // that fails it was turned on some other way and stays silent. +// `connected` exactly: disconnect writes 'disconnected' and leaves isActive and +// liveRelay as they were, and a webhook whose delete failed still posts. const isRelayableIntegration = (integration: DiscordIntegrationDoc, podId: string): boolean => ( isGatedPodTarget(integration, podId) && integration.type === 'discord' && integration.isActive === true - && integration.status !== 'error' + && integration.status === 'connected' && integration.config?.liveRelay === true && !integration.config?.adminPause && Boolean(integration.config?.serverId) @@ -291,7 +293,7 @@ const findLiveIntegration = async (podId: string): Promise.relay` only when its adapter declares `directRows` (Telegram's legacy binds, and every Discord row, which the OAuth install writes directly). -**Discord** sends through the channel webhook the install created -(`config.webhookUrlRef`), never a bot token in the browser: +**Discord** sends server-side through the channel webhook the install +created, resolved from the stored `config.webhookUrlRef`. Outbound uses no bot +token at all: - `allowed_mentions: { parse: [] }`, so agent text pings nobody in the guild, and `@everyone`/`@here` are defused in the text too; - `[Pod] **Agent**: text`, cut to Discord's 2000 characters with an ellipsis @@ -184,7 +185,7 @@ binds, and every Discord row, which the OAuth install writes directly). design (rule 19):** the line is still in the pod, and the audience is the operator log, which names the row and channel. Pacing is per replica; Discord's 429 is the backstop between replicas; -- 10003, 10015, 50001 and 50013 flip the row to `status: 'error'` with a +- 10003, 10015, 50001, 50013 and 50027 flip the row to `status: 'error'` with a reason written for a person (`connectorDeliveryFailureService`). The flip keeps `config.channelId` (F90 fixes the binding); `POST /:id/connect` sets the row connected again once the bot can post. It does not recreate a diff --git a/frontend/src/i18n/locales/en.json b/frontend/src/i18n/locales/en.json index c505ce86c..384f8fd4d 100644 --- a/frontend/src/i18n/locales/en.json +++ b/frontend/src/i18n/locales/en.json @@ -2666,7 +2666,7 @@ "thisSlackWorkspace": "This Slack workspace", "yourSlackUser": "your Slack user", "discordRowMirror": "every agent line reaches everyone in the channel", - "discordRowAttention": "escalations reach everyone in the channel", + "discordRowAttention": "escalations and the lead agent’s lines reach everyone in the channel", "discordMirrorHint": "Every agent message is posted in the Discord channel, where everyone in the server can read it.", "discordAttentionHint": "Only escalations and the lead agent are posted in the Discord channel, where everyone in the server can read them.", "discordInboundNote": "Your own messages in the channel reach the pod as you.", diff --git a/frontend/src/i18n/locales/zh-CN.json b/frontend/src/i18n/locales/zh-CN.json index 77678acde..558d34cbe 100644 --- a/frontend/src/i18n/locales/zh-CN.json +++ b/frontend/src/i18n/locales/zh-CN.json @@ -2667,7 +2667,7 @@ "notion": "你的 Notion 工作空间中的页面、评论和数据库。" }, "discordRowMirror": "智能体的每条消息都会发给频道里的所有人", - "discordRowAttention": "升级消息会发给频道里的所有人", + "discordRowAttention": "升级消息和主导智能体的消息会发给频道里的所有人", "discordMirrorHint": "每条智能体消息都会发到这个 Discord 频道，服务器里的每个人都能看到。", "discordAttentionHint": "只有需要关注的消息和主导智能体的消息会发到这个 Discord 频道，服务器里的每个人都能看到。", "discordInboundNote": "你在频道里发的消息，会以你的身份进入 Pod。", diff --git a/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx b/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx index c3cb52ee7..32b61bf58 100644 --- a/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx +++ b/frontend/src/v2/__tests__/ConnectorsPage.channels.test.tsx @@ -1795,6 +1795,30 @@ describe('Connectors page: Discord live relay (F132)', () => { expect(within(drawer).getByText('Add the bot again from Commonly to turn on relay for this channel.')).toBeInTheDocument(); }); + // CodeRabbit 4475336552: the drawer is the only removal control, so a + // connected Discord row must keep Remove. It goes through the full DELETE + // route, which deletes the channel webhook too, not the isActive PATCH. + it('removes a connected Discord row through DELETE /api/integrations/:id', async () => { + mockGets([discordRow({ discordUserId: '345678901234567890', liveRelay: true })]); + axios.delete.mockResolvedValue({ data: {} }); + renderPage(); + const drawer = await openDrawer('Discord'); + fireEvent.click(within(drawer).getByRole('button', { name: 'Remove' })); + fireEvent.click(within(drawer).getByRole('button', { name: 'Really remove — the channel goes quiet' })); + await waitFor(() => expect(axios.delete).toHaveBeenCalledWith('/api/integrations/i-discord', expect.anything())); + expect(axios.patch).not.toHaveBeenCalled(); + }); + + // CodeRabbit (body, minor): attention mode also sends every lead-agent line. + // The row mark has no renderer on the v2 page, so the copy is asserted where + // it lives, in both locales. + it('says the lead agent’s lines reach a Discord channel in attention mode, in en and zh-CN', () => { + // eslint-disable-next-line global-require + const zh = require('../../i18n/locales/zh-CN.json'); + expect(en.connectors.discordRowAttention).toMatch(/lead agent/); + expect(zh.connectors.discordRowAttention).toMatch(/主导智能体/); + }); + it('renders the shared-channel copy in zh-CN', async () => { await i18nReady; await act(async () => { await i18n.changeLanguage('zh-CN'); }); diff --git a/frontend/src/v2/connectors/ChannelPanel.tsx b/frontend/src/v2/connectors/ChannelPanel.tsx index 8ee1d294c..437aa86ca 100644 --- a/frontend/src/v2/connectors/ChannelPanel.tsx +++ b/frontend/src/v2/connectors/ChannelPanel.tsx @@ -398,8 +398,7 @@ const ChannelPanel: React.FC = ({ m, item: panelItem }) => { )} {connected && userScoped && renderGates(connector)} - {/* Discord's removal is its own uninstall path, not this PATCH. */} - {!isDiscord && renderRemove(item)} + {renderRemove(item)} )} diff --git a/frontend/src/v2/connectors/useChannelConnectors.ts b/frontend/src/v2/connectors/useChannelConnectors.ts index 04ce4fe20..467b01d7d 100644 --- a/frontend/src/v2/connectors/useChannelConnectors.ts +++ b/frontend/src/v2/connectors/useChannelConnectors.ts @@ -630,6 +630,11 @@ export const useChannelConnectors = () => { try { if (connector.installationId) { await api.del(installableLifecyclePath(connector.type)); + } else if (connector.type === 'discord') { + // A Discord row is written by the OAuth install, with no Installable + // parent. The full DELETE removes its channel webhook as well; the + // isActive PATCH below would leave that webhook able to post. + await api.del(`/api/integrations/${connector._id}`); } else { await api.patch(`/api/integrations/${connector._id}`, { isActive: false }); } @@ -738,7 +743,9 @@ export const useChannelConnectors = () => { ? t('connectors.discordRowMirror', { defaultValue: 'every agent line reaches everyone in the channel' }) : t('connectors.rowMirror', { defaultValue: 'every agent line reaches the channel' }); const attentionLabel = isDiscord - ? t('connectors.discordRowAttention', { defaultValue: 'escalations reach everyone in the channel' }) + ? t('connectors.discordRowAttention', { + defaultValue: 'escalations and the lead agent’s lines reach everyone in the channel', + }) : t('connectors.rowAttention', { defaultValue: 'escalations reach the channel' }); const markLabel = relay ? (mirror ? mirrorLabel : attentionLabel)