diff --git a/packages/connector-slack/src/slack-adapter.test.ts b/packages/connector-slack/src/slack-adapter.test.ts index 4d60440..afd03f9 100644 --- a/packages/connector-slack/src/slack-adapter.test.ts +++ b/packages/connector-slack/src/slack-adapter.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it, vi } from 'vitest'; -import { SlackAdapter } from './slack-adapter.js'; +import { SlackAdapter, stripMention } from './slack-adapter.js'; import { parseConsentValue, suggestedPrompts } from './slack-adapter.js'; describe('parseConsentValue', () => { @@ -118,3 +118,300 @@ describe('SlackAdapter thinking status', () => { expect(setStatus).not.toHaveBeenCalled(); }); }); + +describe('stripMention', () => { + it('removes the bot mention token(s) and collapses whitespace', () => { + expect(stripMention('<@U0BOT> what failed?', 'U0BOT')).toBe('what failed?'); + expect(stripMention('hey <@U0BOT> make a repro', 'U0BOT')).toBe('hey make a repro'); + expect(stripMention('<@U0BOT> <@U0BOT> show errors', 'U0BOT')).toBe('show errors'); + }); + it('leaves text without the bot mention unchanged (trimmed)', () => { + expect(stripMention(' what failed? ', 'U0BOT')).toBe('what failed?'); + }); + it('returns empty string for a mention-only message', () => { + expect(stripMention('<@U0BOT>', 'U0BOT')).toBe(''); + }); +}); + +// Helper to extract the Nth handler registered with app.event/app.message/app.command/app.action. +// After wire() runs: +// listeners[0] = app_mention handler +// listeners[1] = app.message handler +// listeners[2] = /peek command handler +// listeners[3] = peek_approve action handler +// listeners[4] = peek_deny action handler +// Each listeners[N] is a chain; the last element is the actual user-supplied callback. +// +// IMPORTANT: This mapping is positional and depends on Bolt's internal listener layout. +// The tripwire test below ("Bolt internal listener layout tripwire") asserts that +// listeners.length === 5. If a Bolt version bump shifts the layout, that test fails +// loudly rather than silently exercising the wrong handler. +function getBoltHandler(adapter: SlackAdapter, index: number) { + const boltApp = (adapter as unknown as { app: { listeners: Array> } }).app; + if (index < 0 || index >= boltApp.listeners.length) { + throw new Error( + `getBoltHandler: index ${index} is out of range (listeners.length = ${boltApp.listeners.length}). If a Bolt version bump changed the listener layout, update the index mapping above.`, + ); + } + const chain = boltApp.listeners[index]; + if (!chain || chain.length === 0) { + throw new Error( + `getBoltHandler: listeners[${index}] is empty or undefined. The Bolt internal listener layout may have changed — re-check the index mapping.`, + ); + } + const cb = chain[chain.length - 1]; + if (cb === undefined || cb === null) { + throw new Error( + `getBoltHandler: last element of listeners[${index}] chain is ${cb}. Expected the user-supplied callback but got nothing — re-check the index mapping.`, + ); + } + return cb as (args: Record) => Promise; +} + +// Tripwire: if Bolt's internal listener layout changes and the index mapping above becomes +// stale, this assertion fails loudly (instead of silently exercising the wrong handler). +describe('Bolt internal listener layout tripwire', () => { + it('boltApp.listeners has exactly 5 entries after wire() — one per registered handler', () => { + const { adapter } = makeAdapter(); + const boltApp = (adapter as unknown as { app: { listeners: Array> } }).app; + // If this count changes, re-verify the positional index mapping in getBoltHandler. + expect(boltApp.listeners).toHaveLength(5); + }); +}); + +describe('SlackAdapter app_mention handler', () => { + it('emits once with conversationId === ts and stripped query on a top-level mention', async () => { + const { adapter } = makeAdapter(); + const received: Array<{ conversationId: string; userId: string; text: string }> = []; + adapter.onMessage((m) => + received.push({ conversationId: m.conversationId, userId: m.userId, text: m.text }), + ); + const handler = getBoltHandler(adapter, 0); + await handler({ + event: { + text: '<@U0BOT> what failed?', + ts: 'TS1', + thread_ts: undefined, + user: 'U1', + channel: 'C1', + }, + context: { botUserId: 'U0BOT' }, + }); + expect(received).toHaveLength(1); + expect(received[0]).toEqual({ conversationId: 'TS1', userId: 'U1', text: 'what failed?' }); + }); + + it('activates the thread so a follow-up app.message reply also emits', async () => { + const { adapter } = makeAdapter(); + const received: string[] = []; + adapter.onMessage((m) => received.push(m.text)); + const mentionHandler = getBoltHandler(adapter, 0); + await mentionHandler({ + event: { + text: '<@U0BOT> show errors', + ts: 'TS2', + thread_ts: undefined, + user: 'U1', + channel: 'C1', + }, + context: { botUserId: 'U0BOT' }, + }); + received.length = 0; // reset — only care about the follow-up + // Verify thread is now active by checking that a reply emits + const msgHandler = getBoltHandler(adapter, 1); + await msgHandler({ + message: { + ts: 'TS2-reply', + thread_ts: 'TS2', + text: 'a follow-up', + user: 'U1', + channel: 'C1', + channel_type: 'channel', + }, + context: { botUserId: 'U0BOT' }, + }); + expect(received).toEqual(['a follow-up']); + }); +}); + +describe('SlackAdapter app.message handler (firehose fix)', () => { + it('emits for a DM message (channel_type: im)', async () => { + const { adapter } = makeAdapter(); + const received: string[] = []; + adapter.onMessage((m) => received.push(m.text)); + const handler = getBoltHandler(adapter, 1); + await handler({ + message: { ts: 'TS3', text: 'hello peek', user: 'U1', channel: 'C2', channel_type: 'im' }, + context: { botUserId: 'U0BOT' }, + }); + expect(received).toEqual(['hello peek']); + }); + + it('does NOT emit for a channel message that is not in an active thread', async () => { + const { adapter } = makeAdapter(); + const received: string[] = []; + adapter.onMessage((m) => received.push(m.text)); + const handler = getBoltHandler(adapter, 1); + await handler({ + message: { + ts: 'TS4', + text: 'some channel chatter', + user: 'U1', + channel: 'C1', + channel_type: 'channel', + }, + context: { botUserId: 'U0BOT' }, + }); + expect(received).toHaveLength(0); + }); + + it('emits for a channel thread reply when that thread was activated by app_mention', async () => { + const { adapter } = makeAdapter(); + const received: string[] = []; + adapter.onMessage((m) => received.push(m.text)); + // Activate the thread via app_mention + const mentionHandler = getBoltHandler(adapter, 0); + await mentionHandler({ + event: { + text: '<@U0BOT> what failed?', + ts: 'TS5', + thread_ts: undefined, + user: 'U1', + channel: 'C1', + }, + context: { botUserId: 'U0BOT' }, + }); + received.length = 0; // reset — only care about the follow-up + // Now send a reply in the same thread + const msgHandler = getBoltHandler(adapter, 1); + await msgHandler({ + message: { + ts: 'TS5-reply', + thread_ts: 'TS5', + text: 'follow-up question', + user: 'U1', + channel: 'C1', + channel_type: 'channel', + }, + context: { botUserId: 'U0BOT' }, + }); + expect(received).toEqual(['follow-up question']); + }); + + it('does NOT emit when the message text contains the bot mention (dedupe with app_mention)', async () => { + const { adapter } = makeAdapter(); + const received: string[] = []; + adapter.onMessage((m) => received.push(m.text)); + const handler = getBoltHandler(adapter, 1); + await handler({ + message: { + ts: 'TS6', + text: '<@U0BOT> what broke?', + user: 'U1', + channel: 'C1', + channel_type: 'channel', + }, + context: { botUserId: 'U0BOT' }, + }); + expect(received).toHaveLength(0); + }); + + it('does NOT dedupe when botUserId is absent — a message with a mention token falls through to normal gating', async () => { + // When botId is empty/absent, the dedupe guard `!isDM && botId && text.includes('<@...>')` + // short-circuits on the falsy botId and does NOT suppress the message. A DM with a + // mention-looking token must still emit (it reaches the isDM gate instead). + const { adapter } = makeAdapter(); + const received: string[] = []; + adapter.onMessage((m) => received.push(m.text)); + const handler = getBoltHandler(adapter, 1); + await handler({ + message: { + ts: 'TS7', + text: '<@SOMEONE> what broke?', // looks like a mention, but botUserId is absent + user: 'U1', + channel: 'C3', + channel_type: 'im', // DM — passes the isDM gate + }, + context: { botUserId: '' }, // absent / empty → dedupe check is skipped + }); + // Falls through the (skipped) dedupe guard → hits isDM gate → emits + expect(received).toEqual(['<@SOMEONE> what broke?']); + }); + + it('emits for a DM whose text contains the bot mention token (Fix 1 regression: DM + mention was incorrectly dropped)', async () => { + // Bug: the old code ran the mention-dedupe guard BEFORE isDM. Slack never emits + // app_mention for DMs, so a DM with "<@BOTID>" was silently dropped — nothing handled it. + // Fix: dedupe guard is now gated on `!isDM`, so DMs always fall through to the isDM check. + const { adapter } = makeAdapter(); + const received: string[] = []; + adapter.onMessage((m) => received.push(m.text)); + const handler = getBoltHandler(adapter, 1); + await handler({ + message: { + ts: 'TS8', + text: '<@UBOT> what broke?', + user: 'U1', + channel: 'D1', + channel_type: 'im', // DM — app_mention never fires here + }, + context: { botUserId: 'UBOT' }, // non-empty botUserId — was the exact trigger of the bug + }); + // Must emit: DMs with a mention token must NOT be deduped + expect(received).toEqual(['<@UBOT> what broke?']); + }); +}); + +describe('SlackAdapter #activeThreads cap (Fix 4)', () => { + it('stays at or below 1000 entries after more than 1000 distinct app_mention events', async () => { + const { adapter } = makeAdapter(); + const mentionHandler = getBoltHandler(adapter, 0); + + // Fire 1005 distinct top-level mentions (each gets a unique ts → unique cid) + for (let i = 0; i < 1005; i++) { + await mentionHandler({ + event: { + text: `<@U0BOT> query ${i}`, + ts: `TS-cap-${i}`, + thread_ts: undefined, + user: 'U1', + channel: 'C1', + }, + context: { botUserId: 'U0BOT' }, + }); + } + + // #activeThreads is private — verify the observable cap effect: + // the oldest thread (TS-cap-0) should have been evicted, so a follow-up + // reply in that thread is NOT treated as an active thread and does NOT emit. + const received: string[] = []; + adapter.onMessage((m) => received.push(m.text)); + const msgHandler = getBoltHandler(adapter, 1); + await msgHandler({ + message: { + ts: 'TS-cap-0-reply', + thread_ts: 'TS-cap-0', // oldest — should be evicted + text: 'evicted thread reply', + user: 'U1', + channel: 'C1', + channel_type: 'channel', + }, + context: { botUserId: 'U0BOT' }, + }); + // If eviction worked correctly, the oldest thread is gone → no emit + expect(received).toHaveLength(0); + + // Sanity: a recent thread (TS-cap-1004) is still active → emits + await msgHandler({ + message: { + ts: 'TS-cap-1004-reply', + thread_ts: 'TS-cap-1004', + text: 'recent thread reply', + user: 'U1', + channel: 'C1', + channel_type: 'channel', + }, + context: { botUserId: 'U0BOT' }, + }); + expect(received).toEqual(['recent thread reply']); + }); +}); diff --git a/packages/connector-slack/src/slack-adapter.ts b/packages/connector-slack/src/slack-adapter.ts index 40ce85f..e091d26 100644 --- a/packages/connector-slack/src/slack-adapter.ts +++ b/packages/connector-slack/src/slack-adapter.ts @@ -32,6 +32,11 @@ export function suggestedPrompts(): { }; } +/** Strip every `<@BOTID>` token for the connector's own bot user id, collapse whitespace. */ +export function stripMention(text: string, botUserId: string): string { + return text.replaceAll(`<@${botUserId}>`, ' ').replace(/\s+/g, ' ').trim(); +} + export function parseConsentValue( raw: string | undefined, ): { correlationId: string; conversationId: string } | null { @@ -52,6 +57,8 @@ export class SlackAdapter implements SurfaceAdapter { private routes = new Map(); private msgHandler?: (m: InboundMessage) => void; private consentHandler?: (r: ConsentResponse) => void; + #activeThreads = new Set(); + readonly #MAX_ACTIVE_THREADS = 1000; constructor(config: SlackConfig) { this.app = new App({ @@ -160,6 +167,16 @@ export class SlackAdapter implements SurfaceAdapter { this.msgHandler?.({ conversationId, userId, text }); } + #trackThread(cid: string): void { + // Bounded to avoid unbounded growth on a long-running socket process. + // Simple insertion-order eviction (oldest first); a TTL is deferred past alpha. + if (this.#activeThreads.size >= this.#MAX_ACTIVE_THREADS && !this.#activeThreads.has(cid)) { + const oldest = this.#activeThreads.values().next().value; + if (oldest !== undefined) this.#activeThreads.delete(oldest); + } + this.#activeThreads.add(cid); + } + private wire(): void { const assistant = new Assistant({ threadStarted: async ({ say, setSuggestedPrompts }) => { @@ -191,8 +208,23 @@ export class SlackAdapter implements SurfaceAdapter { }); this.app.assistant(assistant); - this.app.message(async ({ message }) => { - // message payload shape — subtype discriminates bot/system messages + this.app.event('app_mention', async ({ event, context }) => { + const e = event as { + text?: string; + ts: string; + thread_ts?: string; + user?: string; + channel: string; + }; + const botId = context.botUserId ?? ''; + const query = e.text ? stripMention(e.text, botId) : ''; + if (!query || !e.channel) return; + const cid = e.thread_ts ?? e.ts; + this.#trackThread(cid); + this.emit(cid, e.channel, cid, e.user ?? 'unknown', query); + }); + + this.app.message(async ({ message, context }) => { const m = message as { thread_ts?: string; ts: string; @@ -200,8 +232,17 @@ export class SlackAdapter implements SurfaceAdapter { subtype?: string; user?: string; channel?: string; + channel_type?: string; }; if (m.subtype || !m.text || !m.channel) return; + const botId = context.botUserId ?? ''; + const isDM = m.channel_type === 'im'; + // Dedupe only in channels: a channel mention fires BOTH app_mention and message, + // so skip it here (app_mention handles it). DMs never emit app_mention, so a DM + // whose text contains a mention token must NOT be dropped. + if (!isDM && botId && m.text.includes(`<@${botId}>`)) return; + const inActiveThread = m.thread_ts !== undefined && this.#activeThreads.has(m.thread_ts); + if (!isDM && !inActiveThread) return; // ignore unrelated channel chatter const cid = m.thread_ts ?? m.ts; this.emit(cid, m.channel, cid, m.user ?? 'unknown', m.text); });