feat(slack): stream Search tool progress

Author: TheodoreSpeaks

apps/sim/lib/knowledge/application/operations.test.ts

@@ -71,6 +71,8 @@ describe('knowledge operation registry', () => {
       'knowledge.connectors.create',
       'knowledge.connectors.update',
       'knowledge.connectors.access.update',
+      'knowledge.search.personal-integrations.connect',
+      'knowledge.search.personal-integrations.list',
       'knowledge.search.sources.list',
       'knowledge.search.sources.overview',
       'knowledge.search.sources.progress',

apps/sim/lib/slack-search/assistant-stream.test.ts

@@ -21,6 +21,10 @@ vi.mock('@/executor/utils/resolved-secret-content-projection', () => ({
   projectResolvedSecretDiagnosticContent: api.project,
 }))
 
+import type {
+  ToolCallStreamEvent,
+  ToolResultStreamEvent,
+} from '@/lib/copilot/request/session/contract'
 import type { OrchestratorResult } from '@/lib/copilot/request/types'
 import { publicSlackAnswer, SlackSearchAssistantStream } from '@/lib/slack-search/assistant-stream'
 import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
@@ -80,6 +84,149 @@ function setup(deliverConnections = vi.fn().mockResolvedValue(undefined)) {
     }),
   }
 }
+
+function toolCall(toolName = 'list_integrations', toolCallId = 'tool-1'): ToolCallStreamEvent {
+  return {
+    type: 'tool',
+    payload: {
+      phase: 'call',
+      toolName,
+      toolCallId,
+      executor: 'sim',
+      mode: 'sync',
+      status: 'executing',
+    },
+  }
+}
+
+function toolResult(
+  toolName = 'list_integrations',
+  toolCallId = 'tool-1',
+  success = true
+): ToolResultStreamEvent {
+  return {
+    type: 'tool',
+    payload: { phase: 'result', toolName, toolCallId, executor: 'sim', mode: 'sync', success },
+  }
+}
+
+describe('Slack tool progress', () => {
+  it.each([
+    ['list_integrations', 'Listing connected integrations…'],
+    ['search_workspace', 'Searching documents…'],
+    ['read_document', 'Reading documents…'],
+  ])('shows %s as a task and completes that same task once', async (name, title) => {
+    const { stream } = setup()
+    await stream.start()
+    await stream.onEvent(toolCall(name))
+    await stream.onEvent(toolCall(name))
+    await stream.onEvent(toolResult(name))
+    await stream.onEvent(toolResult(name))
+    await stream.finish(result)
+    const chunks = api.append.mock.calls.flatMap((call) => call[3])
+    expect(chunks).toEqual([
+      { type: 'task_update', id: expect.any(String), title, status: 'in_progress' },
+      { type: 'task_update', id: chunks[0].id, title, status: 'complete' },
+    ])
+    expect(api.stop.mock.calls[0][6]).toEqual([])
+  })
+
+  it('keeps parallel calls separate when their results arrive out of order', async () => {
+    const { stream } = setup()
+    await stream.start()
+    await stream.onEvent(toolCall('search_workspace', 'search-1'))
+    await stream.onEvent(toolCall('search_workspace', 'search-2'))
+    await stream.onEvent(toolResult('search_workspace', 'search-2'))
+    await stream.onEvent(toolResult('search_workspace', 'search-1'))
+    const chunks = api.append.mock.calls.flatMap((call) => call[3])
+    expect(chunks[0].id).not.toBe(chunks[1].id)
+    expect(chunks[2]).toEqual({ ...chunks[1], status: 'complete' })
+    expect(chunks[3]).toEqual({ ...chunks[0], status: 'complete' })
+  })
+
+  it('withholds partial, hidden, internal, subagent, and unsupported tools', async () => {
+    const { stream } = setup()
+    await stream.start()
+    for (const attributes of [
+      { partial: true },
+      { status: 'generating' as const },
+      { ui: { hidden: true } },
+      { ui: { internal: true } },
+    ]) {
+      const event = toolCall()
+      await stream.onEvent({ ...event, payload: { ...event.payload, ...attributes } })
+    }
+    await stream.onEvent({ ...toolCall(), scope: { lane: 'subagent', agentId: 'private-agent' } })
+    await stream.onEvent(toolCall('internal_tool'))
+    await stream.onEvent(toolResult())
+    expect(api.append).not.toHaveBeenCalled()
+    await stream.onEvent(toolCall())
+    expect(api.append).toHaveBeenCalledOnce()
+  })
+
+  it('reports failed tools without exposing arguments, account labels, or backend errors', async () => {
+    const { stream } = setup()
+    await stream.start()
+    const call = toolCall()
+    await stream.onEvent({
+      ...call,
+      payload: { ...call.payload, arguments: { query: 'private argument' } },
+    })
+    const failed = toolResult('list_integrations', 'tool-1', false)
+    await stream.onEvent({
+      ...failed,
+      payload: {
+        ...failed.payload,
+        error: 'private error',
+        output: { accountLabel: 'private account' },
+      },
+    })
+    const chunks = api.append.mock.calls.flatMap((call) => call[3])
+    expect(chunks[1]).toEqual({ ...chunks[0], status: 'error' })
+    expect(JSON.stringify(chunks)).not.toContain('private')
+  })
+
+  it('marks unfinished tasks failed when the Assistant fails', async () => {
+    const { stream } = setup()
+    await stream.start()
+    await stream.onEvent(toolCall())
+    await stream.finishWithError()
+    expect(api.stop.mock.calls[0][6]).toEqual([
+      { ...api.append.mock.calls[0][3][0], status: 'error' },
+    ])
+  })
+
+  it('aborts an ambiguous progress send and cleans up once without replaying it', async () => {
+    const { stream, controller } = setup()
+    await stream.start()
+    api.append.mockRejectedValueOnce(new Error('progress response lost'))
+    await expect(stream.onEvent(toolCall())).rejects.toThrow('progress response lost')
+    expect(controller.signal.aborted).toBe(true)
+    await expect(stream.onEvent(toolCall())).rejects.toThrow('progress response lost')
+    await stream.terminateAfterFailure()
+    await stream.terminateAfterFailure()
+    expect(api.append).toHaveBeenCalledOnce()
+    expect(api.stop).toHaveBeenCalledOnce()
+    expect(api.stop.mock.calls[0][6]).toEqual([
+      { ...api.append.mock.calls[0][3][0], status: 'error' },
+    ])
+  })
+
+  it('does not send task updates after cancellation or revoked delivery authority', async () => {
+    const { stream, controller, beforeDelivery } = setup()
+    await stream.start()
+    beforeDelivery.mockRejectedValueOnce(new Error('authority revoked'))
+    await expect(stream.onEvent(toolCall())).rejects.toThrow('authority revoked')
+    expect(controller.signal.aborted).toBe(true)
+    expect(api.append).not.toHaveBeenCalled()
+    const cancelled = setup()
+    await cancelled.stream.start()
+    cancelled.controller.abort(new Error('stopped'))
+    await expect(cancelled.stream.onEvent(toolCall())).rejects.toThrow('stopped')
+    expect(api.append).not.toHaveBeenCalled()
+  })
+})
+
 describe('Slack Assistant delivery', () => {
   it('withholds split connection tags, delivers validated controls, and leaves a visible next step', async () => {
     const deliver = vi.fn().mockResolvedValue(undefined)
@@ -249,7 +396,8 @@ describe('Slack Assistant delivery', () => {
           type: 'section',
           text: { type: 'plain_text', text: 'I couldn’t complete this search. Please try again.' },
         },
-      ]
+      ],
+      []
     )
     expect(api.append).not.toHaveBeenCalled()
   })

apps/sim/lib/slack-search/assistant-stream.ts

@@ -1,12 +1,17 @@
 import { toError } from '@sim/utils/errors'
+import { generateId } from '@sim/utils/id'
 import { truncate } from '@sim/utils/string'
 import {
   collectRetrievalCitationEvidence,
   parseCitationRecord,
   type RetrievalCitationBlock,
 } from '@/lib/copilot/chat/citation-evidence'
 import { redactSensitiveContent } from '@/lib/copilot/chat/sim-key-redaction'
-import type { StreamEvent } from '@/lib/copilot/request/session/contract'
+import type {
+  StreamEvent,
+  ToolCallStreamEvent,
+  ToolResultStreamEvent,
+} from '@/lib/copilot/request/session/contract'
 import type { OrchestratorResult } from '@/lib/copilot/request/types'
 import {
   parseSearchConnectionTargets,
@@ -15,6 +20,7 @@ import {
 import { SLACK_SEARCH_FAILED_ANSWER } from '@/lib/slack-search/constants'
 import {
   appendSlackAgentStream,
+  type SlackStreamChunk,
   setSlackAgentSessionStatus,
   startSlackAgentStream,
   stopSlackAgentStream,
@@ -100,6 +106,14 @@ const FAILURE_BLOCKS: Record<string, unknown>[] = [
   { type: 'section', text: { type: 'plain_text', text: SLACK_SEARCH_FAILED_ANSWER } },
 ]
 
+const TOOL_PROGRESS_TITLES = new Map([
+  ['list_integrations', 'Listing connected integrations…'],
+  ['search_workspace', 'Searching documents…'],
+  ['read_document', 'Reading documents…'],
+])
+
+type ToolProgress = Extract<SlackStreamChunk, { type: 'task_update' }>
+
 /** Serial delivery through the same provider primitives as Slack blocks; ambiguous sends are terminal. */
 export class SlackSearchAssistantStream {
   private stream?: { channel: string; ts: string }
@@ -111,6 +125,7 @@ export class SlackSearchAssistantStream {
   private closeAttempted = false
   private separateNextText = false
   private evidence = new Map<string, Record<string, unknown>>()
+  private toolProgress = new Map<string, { toolName: string; chunk: ToolProgress }>()
   constructor(private readonly options: AssistantStreamOptions) {}
 
   private async deliver(action: () => Promise<void>) {
@@ -159,7 +174,15 @@ export class SlackSearchAssistantStream {
         },
       ])
     }
-    if (event.type === 'tool' && !event.scope) this.separateNextText = true
+    if (event.type === 'tool' && !event.scope) {
+      this.separateNextText = true
+      if (
+        'phase' in event.payload &&
+        (event.payload.phase === 'call' || event.payload.phase === 'result')
+      ) {
+        await this.updateToolProgress(event.payload)
+      }
+    }
     if (event.type !== 'text' || event.payload.channel !== 'assistant' || event.scope) return
     if (this.separateNextText && this.text) this.text += '\n\n'
     this.separateNextText = false
@@ -168,6 +191,57 @@ export class SlackSearchAssistantStream {
     if (Date.now() - this.lastSentAt >= 750) await this.flush(false)
   }
 
+  /** Only static labels reach Slack; arguments, account details, and backend errors stay private. */
+  private async updateToolProgress(
+    payload: ToolCallStreamEvent['payload'] | ToolResultStreamEvent['payload']
+  ) {
+    const title = TOOL_PROGRESS_TITLES.get(payload.toolName)
+    if (!title) return
+    const existing = this.toolProgress.get(payload.toolCallId)
+    let chunk: ToolProgress
+    if (payload.phase === 'call') {
+      if (
+        existing ||
+        payload.partial ||
+        payload.ui?.hidden ||
+        payload.ui?.internal ||
+        (payload.status !== undefined && payload.status !== 'executing')
+      )
+        return
+      chunk = { type: 'task_update', id: generateId(), title, status: 'in_progress' }
+      this.toolProgress.set(payload.toolCallId, { toolName: payload.toolName, chunk })
+    } else {
+      if (!existing || existing.chunk.status !== 'in_progress') return
+      if (existing.toolName !== payload.toolName)
+        throw new Error('Slack tool progress identity changed')
+      chunk = {
+        ...existing.chunk,
+        status:
+          payload.success && (!payload.status || payload.status === 'success')
+            ? 'complete'
+            : 'error',
+      }
+    }
+    await this.deliver(async () => {
+      if (!this.stream || this.closed) throw new Error('Slack stream is not active')
+      await appendSlackAgentStream(
+        this.options.token,
+        this.stream.channel,
+        this.stream.ts,
+        [chunk],
+        this.options.controller.signal
+      )
+    })
+    this.toolProgress.set(payload.toolCallId, { toolName: payload.toolName, chunk })
+  }
+
+  /** Finalize interrupted tasks in the single stop request, including ambiguous progress sends. */
+  private interruptedToolProgress(): ToolProgress[] {
+    return [...this.toolProgress.values()]
+      .filter(({ chunk }) => chunk.status === 'in_progress')
+      .map(({ chunk }) => ({ ...chunk, status: 'error' }))
+  }
+
   private collectSources(blocks: readonly RetrievalCitationBlock[]) {
     for (const [id, source] of collectRetrievalCitationEvidence(blocks)) {
       if (!this.evidence.has(id)) this.evidence.set(id, source)
@@ -263,7 +337,8 @@ export class SlackSearchAssistantStream {
       this.stream.ts,
       'active',
       signal,
-      FAILURE_BLOCKS
+      FAILURE_BLOCKS,
+      this.interruptedToolProgress()
     )
     this.closed = true
   }
@@ -278,7 +353,8 @@ export class SlackSearchAssistantStream {
         this.stream.ts,
         'active',
         this.options.controller.signal,
-        blocks
+        blocks,
+        this.interruptedToolProgress()
       )
       this.closed = true
     })

apps/sim/lib/webhooks/slack-agent-api.test.ts

@@ -85,6 +85,28 @@ describe('Slack agent API transport', () => {
     })
   })
 
+  it('finishes task updates in the same stop request as the final blocks', async () => {
+    const fetchMock = vi.fn().mockResolvedValue(new Response(JSON.stringify({ ok: true })))
+    vi.stubGlobal('fetch', fetchMock)
+    const chunks = [
+      {
+        type: 'task_update' as const,
+        id: 'task-1',
+        title: 'Searching documents…',
+        status: 'error' as const,
+      },
+    ]
+    const blocks = [{ type: 'section', text: { type: 'plain_text', text: 'Please try again.' } }]
+    await stopSlackAgentStream('xoxb-test', 'D1', '101.2', 'active', undefined, blocks, chunks)
+    expect(JSON.parse(fetchMock.mock.calls[0][1].body)).toEqual({
+      channel: 'D1',
+      ts: '101.2',
+      session_status: 'active',
+      blocks,
+      chunks,
+    })
+  })
+
   it('sets the human initiator when creating a processing session', async () => {
     const fetchMock = vi
       .fn()

apps/sim/lib/webhooks/slack-agent-api.ts

@@ -148,12 +148,19 @@ export async function stopSlackAgentStream(
   ts: string,
   sessionStatus: 'active' | 'processing' | 'suspended',
   signal?: AbortSignal,
-  blocks?: Record<string, unknown>[]
+  blocks?: Record<string, unknown>[],
+  chunks?: SlackStreamChunk[]
 ): Promise<void> {
   await callSlackAgentApi(
     'chat.stopStream',
     token,
-    { channel, ts, session_status: sessionStatus, ...(blocks?.length ? { blocks } : {}) },
+    {
+      channel,
+      ts,
+      session_status: sessionStatus,
+      ...(blocks?.length ? { blocks } : {}),
+      ...(chunks?.length ? { chunks } : {}),
+    },
     signal
   )
 }