fix(slack-search): finalize failed streams and explain query limits

Author: TheodoreSpeaks

apps/sim/lib/knowledge/application/slack-search/assistant.test.ts

@@ -17,6 +17,8 @@ const m = vi.hoisted(() => ({
   start: vi.fn(),
   finish: vi.fn(),
   finishWithError: vi.fn(),
+  terminate: vi.fn(),
+  streamOptions: vi.fn(),
   payload: vi.fn(),
   outcome: vi.fn(),
   memberAuthorization: vi.fn(),
@@ -99,9 +101,13 @@ vi.mock('@/lib/copilot/request/session/abort', () => ({
 }))
 vi.mock('@/lib/slack-search/assistant-stream', () => ({
   SlackSearchAssistantStream: class {
+    constructor(options: unknown) {
+      m.streamOptions(options)
+    }
     start = m.start
     finish = m.finish
     finishWithError = m.finishWithError
+    terminateAfterFailure = m.terminate
     onEvent = vi.fn()
     assertHealthy = vi.fn()
   },
@@ -308,9 +314,24 @@ describe('organization Assistant from Slack', () => {
     m.finish.mockRejectedValueOnce(new Error('ambiguous send'))
     await expect(run()).rejects.toThrow('ambiguous send')
     expect(m.run).toHaveBeenCalledOnce()
+    expect(m.terminate).toHaveBeenCalledOnce()
     expect(m.updateRun).toHaveBeenCalledWith('run1', 'error')
     expect(m.release).toHaveBeenCalledOnce()
   })
+  it('rechecks cleanup authority with a fresh signal after aborting execution', async () => {
+    m.run.mockRejectedValueOnce(new Error('Assistant disconnected'))
+    m.terminate.mockImplementationOnce(async () => {
+      const options = m.streamOptions.mock.calls[0][0]
+      expect(options.controller.signal.aborted).toBe(true)
+      await options.beforeCleanup(new AbortController().signal)
+    })
+    await expect(run()).rejects.toThrow('Assistant disconnected')
+    expect(m.terminate).toHaveBeenCalledOnce()
+    expect(m.memberAuthorization).toHaveBeenCalled()
+    expect(m.terminate.mock.invocationCallOrder[0]).toBeLessThan(
+      m.release.mock.invocationCallOrder[0]
+    )
+  })
   it('closes a healthy Slack stream when the Assistant reports failure', async () => {
     m.run.mockResolvedValueOnce({ success: false, content: '', contentBlocks: [], toolCalls: [] })
     await expect(run()).rejects.toThrow('Organization Assistant did not complete')
@@ -328,9 +349,10 @@ describe('organization Assistant from Slack', () => {
   })
   it('marks private history as stopped only when the turn was cancelled by native Stop', async () => {
     m.run.mockRejectedValueOnce(new Error('Stream stopped'))
-    m.stopped.mockResolvedValueOnce(true)
+    m.stopped.mockResolvedValue(true)
     await expect(run()).rejects.toThrow('Stream stopped')
     expect(m.stoppedMessage).toHaveBeenCalledOnce()
     expect(m.finishWithError).not.toHaveBeenCalled()
+    expect(m.terminate).not.toHaveBeenCalled()
   })
 })

apps/sim/lib/knowledge/application/slack-search/assistant.ts

@@ -140,9 +140,10 @@ export async function runSlackSearchAssistant(
   let failed = true
   let failure: Error | undefined
   let registry: ResolvedSecretTraceRegistry | undefined
+  let stream: SlackSearchAssistantStream | undefined
   let checking = false
-  const checkAccess = async () => {
-    controller.signal.throwIfAborted()
+  const checkAccess = async (signal: AbortSignal = controller.signal) => {
+    signal.throwIfAborted()
     await requireSlackSearchTurnLease(turnId, leaseId)
     if (!(await authorizeSlackSearchInstallation(principal, job)))
       throw new OrchestrationError('forbidden', 'Slack Search is disabled')
@@ -223,15 +224,17 @@ export async function runSlackSearchAssistant(
       })
       if (!run) throw new Error('Could not persist Assistant execution')
       runId = run.id
-      const stream = new SlackSearchAssistantStream({
+      const responseStream = new SlackSearchAssistantStream({
         token: secret.botToken,
         channel: job.message.channelId,
         threadTs: job.message.threadTs ?? job.message.messageTs,
         slackUserId: job.message.userId,
         controller,
         registry: environmentContext.resolvedSecretTraceRegistry,
         beforeDelivery: checkAccess,
+        beforeCleanup: checkAccess,
       })
+      stream = responseStream
       const payload = await buildCopilotRequestPayload(
         {
           message: job.message.query,
@@ -248,7 +251,7 @@ export async function runSlackSearchAssistant(
         actorUserId: userId,
         organizationId: installation.organizationId,
       })
-      await stream.start()
+      await responseStream.start()
       result = await runHeadlessCopilotLifecycle(payload, {
         userId,
         organizationId: installation.organizationId,
@@ -264,28 +267,36 @@ export async function runSlackSearchAssistant(
         autoExecuteTools: true,
         onEvent: async (event) => {
           try {
-            await stream.onEvent(event)
+            await responseStream.onEvent(event)
           } catch (error) {
             controller.abort(error)
             throw error
           }
         },
       })
-      stream.assertHealthy()
+      responseStream.assertHealthy()
       controller.signal.throwIfAborted()
       if (!result.success) {
-        await stream.finishWithError()
+        await responseStream.finishWithError()
         throw new Error('Organization Assistant did not complete')
       }
       await checkAccess()
-      await stream.finish(result)
+      await responseStream.finish(result)
       failed = false
       await recordSlackSearchOutcome(installation, 'success')
     }
   } catch (error) {
     failure = toError(error)
     controller.abort(error)
+    const failedStream = stream
     const outcomes = await Promise.allSettled([
+      ...(failedStream
+        ? [
+            wasSlackSearchTurnStopped(turnId, leaseId).then((stopped) =>
+              stopped ? undefined : failedStream.terminateAfterFailure()
+            ),
+          ]
+        : []),
       recordSlackSearchOutcome(installation, 'assistant_or_delivery_failed'),
       ...(runId ? [updateRunStatus(runId, 'error')] : []),
     ])
@@ -295,7 +306,7 @@ export async function runSlackSearchAssistant(
     if (errors.length)
       failure = new AggregateError(
         [failure, ...errors],
-        'Slack turn and outcome persistence failed'
+        'Slack turn cleanup or outcome persistence failed'
       )
   } finally {
     await titleTask

apps/sim/lib/knowledge/application/slack-search/mention.test.ts

@@ -0,0 +1,100 @@
+/** @vitest-environment node */
+import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@sim/testing'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+const mocks = vi.hoisted(() => ({ authorize: vi.fn(), openDm: vi.fn(), post: vi.fn() }))
+vi.mock('@/lib/knowledge/application/slack-search/authorization', () => ({
+  authorizeSlackSearchInstallation: mocks.authorize,
+}))
+vi.mock('@/lib/internal/slack/client', () => ({
+  openSlackDm: mocks.openDm,
+  postSlackMessage: mocks.post,
+  slackString: (value: Record<string, unknown>, key: string) =>
+    typeof value[key] === 'string' ? value[key] : undefined,
+}))
+
+import { routeSlackSearchMentionToDm } from '@/lib/knowledge/application/slack-search/mention'
+import { SLACK_SEARCH_QUERY_TOO_LONG } from '@/lib/slack-search/constants'
+import { slackSearchJobSchema } from '@/lib/slack-search/types'
+
+const principal = {
+  kind: 'slack_installation',
+  credentialId: 'c1',
+  credentialVersion: 'v1',
+  appId: 'A1',
+  teamId: 'T1',
+  eventId: 'Ev1',
+  receivedAt: new Date(),
+} as const
+
+beforeEach(() => {
+  vi.clearAllMocks()
+  resetDbChainMock()
+  mocks.authorize.mockResolvedValue({ secret: { botToken: 'test-token' } })
+  mocks.openDm.mockResolvedValue('D1')
+  mocks.post.mockResolvedValue({
+    status: 200,
+    data: { ok: true, channel: 'D1', ts: '1800000000.2' },
+  })
+})
+
+describe('private Slack mention roots', () => {
+  it.each([false, true])(
+    'creates a nonempty private root for queryTooLong=%s',
+    async (queryTooLong) => {
+      const job = slackSearchJobSchema.parse({
+        installationId: 'i1',
+        revision: 'r1',
+        credentialId: 'c1',
+        credentialVersion: 'v1',
+        receivedAt: principal.receivedAt.getTime(),
+        message: {
+          appId: 'A1',
+          teamId: 'T1',
+          eventId: 'Ev1',
+          channelId: 'C1',
+          userId: 'U1',
+          messageTs: '1800000000.1',
+          origin: { channelId: 'C1', threadTs: '1800000000.1', messageTs: '1800000000.1' },
+          query: queryTooLong ? '' : 'Find the release notes',
+          queryTooLong,
+        },
+      })
+      queueTableRows(schemaMock.slackSearchInstallation, [
+        { enabled: true, revision: 'r1', credentialVersion: 'v1' },
+      ])
+      queueTableRows(schemaMock.slackSearchTurn, [
+        {
+          status: 'running',
+          leaseId: 'lease1',
+          leaseExpiresAt: new Date(Date.now() + 60_000),
+          payload: job,
+        },
+      ])
+      const routed = await routeSlackSearchMentionToDm(principal, {
+        job,
+        turnId: 'turn1',
+        leaseId: 'lease1',
+        signal: new AbortController().signal,
+      })
+      expect(mocks.post).toHaveBeenCalledWith(
+        'test-token',
+        expect.objectContaining({
+          channel: 'D1',
+          blocks: [
+            {
+              type: 'section',
+              text: {
+                type: 'plain_text',
+                text: queryTooLong ? SLACK_SEARCH_QUERY_TOO_LONG : job.message.query,
+              },
+            },
+          ],
+        }),
+        expect.any(AbortSignal)
+      )
+      expect(routed.message).toMatchObject({ channelId: 'D1', threadTs: '1800000000.2' })
+      expect(dbChainMockFns.set).toHaveBeenCalledWith(expect.objectContaining({ payload: routed }))
+    }
+  )
+})

apps/sim/lib/knowledge/application/slack-search/mention.ts

@@ -5,6 +5,7 @@ import { and, eq } from 'drizzle-orm'
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { openSlackDm, postSlackMessage, slackString } from '@/lib/internal/slack/client'
 import { authorizeSlackSearchInstallation } from '@/lib/knowledge/application/slack-search/authorization'
+import { SLACK_SEARCH_QUERY_TOO_LONG } from '@/lib/slack-search/constants'
 import { slackSearchConversationKey } from '@/lib/slack-search/conversation'
 import { type SlackSearchJob, slackSearchJobSchema } from '@/lib/slack-search/types'
 
@@ -72,8 +73,18 @@ export async function routeSlackSearchMentionToDm(
       context.secret.botToken,
       {
         channel: channelId,
-        text: 'Your question for Sim Search',
-        blocks: [{ type: 'section', text: { type: 'plain_text', text: job.message.query } }],
+        text: job.message.queryTooLong
+          ? SLACK_SEARCH_QUERY_TOO_LONG
+          : 'Your question for Sim Search',
+        blocks: [
+          {
+            type: 'section',
+            text: {
+              type: 'plain_text',
+              text: job.message.queryTooLong ? SLACK_SEARCH_QUERY_TOO_LONG : job.message.query,
+            },
+          },
+        ],
         unfurl_links: false,
         unfurl_media: false,
       },