feat(connector-slack): @peek in a channel — app_mention + threaded auto-continue

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<Array<unknown>> } }).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<string, unknown>) => Promise<void>;
+}
+
+// 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<Array<unknown>> } }).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']);
+  });
+});

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<string, Route>();
   private msgHandler?: (m: InboundMessage) => void;
   private consentHandler?: (r: ConsentResponse) => void;
+  #activeThreads = new Set<string>();
+  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,17 +208,41 @@ 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;
         text?: string;
         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);
     });